スパーク：私はスパーク1.4.1上で次のコマンドを実行しようとすると、DATAFRAME

Question

上のUDFためのタスクシリアライズない私はorg.apache.spark.SparkException: Task not serializableを得る：ここで

import java.sql.{Date, Timestamp} 
import java.text.SimpleDateFormat 

object ConversionUtils { 
    val iso8601 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSX") 

    def tsUTC(s: String): Timestamp = new Timestamp(iso8601.parse(s).getTime) 

    val castTS = udf[Timestamp, String](tsUTC _) 
} 

val df = frame.withColumn("ts", ConversionUtils.castTS(frame("ts_str"))) 
df.first 

、frameはHiveContext以内に住んでいるDataFrameです。そのデータフレームには問題はありません。

私は整数に似たUDFを持っており、問題なく動作します。しかし、タイムスタンプを持つものは問題を引き起こすようです。 documentationによれば、java.sql.TimeStampはSerializableを実装しているため、問題はありません。同じことがhereのようにSimpleDateFormatについても当てはまります。

これは、問題を引き起こしているのがUDFだと信じさせています。しかし、私は何をどのように修正するのかよく分かりません。

トレースの関連セクション：

Caused by: java.io.NotSerializableException: ... 
Serialization stack: 
     - object not serializable (class: ..., value: ...$ConversionUtils$@63ed11dd) 
     - field (class: ...$ConversionUtils$$anonfun$3, name: $outer, type: class ...$ConversionUtils$) 
     - object (class ...$ConversionUtils$$anonfun$3, <function1>) 
     - field (class: org.apache.spark.sql.catalyst.expressions.ScalaUdf$$anonfun$2, name: func$2, type: interface scala.Function1) 
     - object (class org.apache.spark.sql.catalyst.expressions.ScalaUdf$$anonfun$2, <function1>) 
     - field (class: org.apache.spark.sql.catalyst.expressions.ScalaUdf, name: f, type: interface scala.Function1) 
     - object (class org.apache.spark.sql.catalyst.expressions.ScalaUdf, scalaUDF(ts_str#2683)) 
     - field (class: org.apache.spark.sql.catalyst.expressions.Alias, name: child, type: class org.apache.spark.sql.catalyst.expressions.Expression) 
     - object (class org.apache.spark.sql.catalyst.expressions.Alias, scalaUDF(ts_str#2683) AS ts#7146) 
     - element of array (index: 35) 
     - array (class [Ljava.lang.Object;, size 36) 
     - field (class: scala.collection.mutable.ArrayBuffer, name: array, type: class [Ljava.lang.Object;) 
     - object (class scala.collection.mutable.ArrayBuffer, 

Answer 1

試してみてください。

object ConversionUtils extends Serializable { 
    ... 
}