title: Exception handling in spark udf
body: <p>I have spark user defined function which returns date in certain format</p>
  
  <pre><code>val getEventdatetime: (String,String) =&gt; String = (strLogArrivalDate,strEventDatetime) =&gt;  {
  
          val year = strLogArrivalDate.split('-')(0).toString() //get year part
          val strEventDatetime1 = year + " " + strEventDatetime
          val inputFormat = new SimpleDateFormat("yyyy MMM dd HH:mm:ss")
  
          new Timestamp(inputFormat.parse(strEventDatetime1).getTime).toString()
  
      }
  
      // Defining spark UserDefinedFunction
      val getEventdatetimeUDF = udf[String,String,String](getEventdatetime)
  </code></pre>
  
  <p>I'm receiving the input dates from some logs, if there is any problem in date value, my application fails.
  For e.g.</p>
  
  <pre><code>val strEventDatetime1="2019 Jul 01 00:00:00" //works fine
  val strEventDatetime1="2019 Jul 01 k 00:00:00" //throws error
  </code></pre>
  
  <p>I've tried wrapping everything in try/catch but it doesn't work</p>
  
  <pre><code>val getEventdatetime: (String,String) =&gt; String = (strLogArrivalDate,strEventDatetime) =&gt;  {
  
        try {
          val year = strLogArrivalDate.split('-')(0).toString() //get year part
          val strEventDatetime1 = year + " " + strEventDatetime
          val inputFormat = new SimpleDateFormat("yyyy MMM dd HH:mm:ss")
  
          new Timestamp(inputFormat.parse(strEventDatetime1).getTime).toString()
        }catch{
          case e: Exception =&gt; println(e.getMessage())
          "1900-01-01 00:00::"
        }
  
      }
  </code></pre>
  
  <p>Calling the UDF:</p>
  
  <pre><code>// Adding new column "event_datetime1" to DF , it's value is derived from log_arrival_time &amp; event_datetime
  val dfProcessed = dfValidLogs1.withColumn("event_datetime1", getEventdatetimeUDF.apply($"log_arrival_time",$"event_datetime"))
  </code></pre>
  
  <p>This is giving me '<strong>Task not serializable</strong>' error. I've limited knowledge of how spark udf works. What will the best way to handle error inside the udf?
  Thanks in advance.</p>
  
  <p><strong>Stack trace for error</strong></p>
  
  <pre><code>org.apache.spark.SparkException: Task not serializable
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:345)
  at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:335)
  at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:159)
  at org.apache.spark.SparkContext.clean(SparkContext.scala:2299)
  at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1.apply(RDD.scala:844)
  at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1.apply(RDD.scala:843)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
  at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
  at org.apache.spark.rdd.RDD.mapPartitionsWithIndex(RDD.scala:843)
  at org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:608)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.streaming.ForeachSink.addBatch(ForeachSink.scala:49)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$3$$anonfun$apply$16.apply(MicroBatchExecution.scala:475)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$3.apply(MicroBatchExecution.scala:473)
  at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:271)
  at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch(MicroBatchExecution.scala:472)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:133)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
  at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:271)
  at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:121)
  at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:117)
  at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:279)
  at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:189)
  Caused by: java.io.NotSerializableException: java.lang.Object
  Serialization stack:
  - object not serializable (class: java.lang.Object, value: java.lang.Object@3238cce)
  - field (class: com.sb.spark.accesspoint.AccesspointAnalysis$accesspointAnalysis$$anonfun$6, name: nonLocalReturnKey1$1, type: class java.lang.Object)
  - object (class com.sb.spark.accesspoint.AccesspointAnalysis$accesspointAnalysis$$anonfun$6, &lt;function2&gt;)
  - element of array (index: 9)
  - array (class [Ljava.lang.Object;, size 17)
  - field (class: org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, name: references$1, type: class [Ljava.lang.Object;)
  - object (class org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, &lt;function2&gt;)
  at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
  at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
  at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:342)
  </code></pre>