Use a method inside a UDF function Spark Scala

Question

I want to use a method located in another class inside a user-designed function but it's not working.

I have a method:

 def traitementDataFrameEleve(sc:SparkSession, dfRedis:DataFrame, domainMail:String, dir:String):Boolean ={
     def loginUDF = udf((sn: String, givenName:String) => {
            LoginClass.GenerateloginPersone(sn,givenName,dfr)
          })

    dfEleve.withColumn("ENTPersonLogin",loginUDF(dfEleve("sn"),dfEleve("givenName")))
}

LoginClass is a class that contains the GenerateloginPersone method.

Output error :

org.apache.spark.SparkException: Failed to execute user defined function(anonfun$loginUDF$1$1: (string, string) => string)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:231)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:225)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:99)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: java.lang.NullPointerException
    at org.apache.spark.sql.Dataset.schema(Dataset.scala:410)
    at org.apache.spark.sql.Dataset.printSchema(Dataset.scala:419)
    at IntegrationDonneesENTLea_V1_AcBordeaux.LoginClass$.GenerateloginPersone(LoginClass.scala:16)
    at IntegrationDonneesENTLea_V1_AcBordeaux.Eleve$$anonfun$loginUDF$1$1.apply(Eleve.scala:25)
    at IntegrationDonneesENTLea_V1_AcBordeaux.Eleve$$anonfun$loginUDF$1$1.apply(Eleve.scala:23)
    ... 16 more

Thank you.

user7920791 · Accepted Answer · 2017-04-25 16:58:06Z

3

It is not allowed to access:

distributed data structures (like Dataset or RDD).
SparkConext / SparkSession

from Spark task (transformation, udf application). This is why you get a NPE.

answered Apr 25, 2017 at 16:58

user7920791

311 bronze badge

Sign up to request clarification or add additional context in comments.

1 Comment

Tzach Zohar Over a year ago

@MounirHamdane to do what? You can call a method from within a UDF (per your post title), the problem here is the contents of your method - GenerateloginPersone seems to call Dataset.printSchema - which is something you can't do. Since we can't see the implementation of GenerateloginPersone, we can't offer any more insight as to how it can or should be implemented - we don't know what it does. So, to summarize: yes, you can call methods from within UDFs; But these methods can't try to use SparkContext / SparkSession or anything that uses those (like Dataset / RDD).

Collectives™ on Stack Overflow

Use a method inside a UDF function Spark Scala

1 Answer 1

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related