组织。阿帕奇。火花SparkException:任务不可序列化。斯卡拉火花

Question

将现有应用程序从Spark 1.6移动到Spark 2.2*（最终）会导致错误“org.apache.spark.SparkExctive：任务不可序列化”。我过于简化了我的代码，以演示同样的错误。代码查询拼花文件以返回以下数据类型：“org.apache.spark.sql.数据集[org.apache.spark.sql.行]”我应用一个函数来提取字符串和整数，返回字符串。一个固有的问题与Spark 2.2返回数据集而不是数据帧有关。（请参阅之前关于初步错误的帖子）如何编写数据集编码器以支持将函数映射到Scala Spark中的org.apache.spark.sql.数据集[字符串]

var d1 = hive.executeQuery(st)
d1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [cvdt35_message_id_d: string, cvdt35_input_timestamp_s: decimal(16,5) ... 2 more fields]

scala> val parseCVDP_parquet = (s:org.apache.spark.sql.Row) => s.getString(2).split("0x")(1)+","+s.getDecimal(1);
parseCVDP_parquet: org.apache.spark.sql.Row => String = <function1>

scala> var d2 =  d1.map(parseCVDP_parquet)
d2: org.apache.spark.sql.Dataset[String] = [value: string]

scala> def dd(s:String, start: Int) = { s + "some string" }
dd: (s: String, start: Int)String

scala> var d3 = d2.map{s=> dd(s,5) }
d3: org.apache.spark.sql.Dataset[String] = [value: string]

scala> d3.take(1)
org.apache.spark.SparkException: Task not serializable

我目前解决这个问题的方法是嵌入内联代码（见下文），但并不实用，因为我的生产代码包含大量的参数和函数。我还尝试过转换为数据帧（就像spark 1.6中那样）和函数定义的变体，这些都没有证明是可行的解决方案。

scala> var d1 = hive.executeQuery(st)
d1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [cvdt35_message_id_d: string, cvdt35_input_timestamp_s: decimal(16,5) ... 2 more fields]

scala> val parseCVDP_parquet = (s:org.apache.spark.sql.Row) => s.getString(2).split("0x")(1)+","+s.getDecimal(1);
parseCVDP_parquet: org.apache.spark.sql.Row => String = <function1>

scala> var d2 =  d1.map(parseCVDP_parquet)
d2: org.apache.spark.sql.Dataset[String] = [value: string]

scala> var d3 = d2.map{s=> { s + "some string" } }
d3: org.apache.spark.sql.Dataset[String] = [value: string]

scala> d3.take(1)
20/04/30 15:16:17 WARN TaskSetManager: Stage 0 contains a task of very large size (132 KB). The maximum recommended task size is 100 KB.
res1: Array[String] = Array(761f006000705904,1521833533.96682some string)

Answer 1

org。阿帕奇。火花SparkException:任务未序列化

要解决这个问题，请将您的所有功能

通过这种方式，您可以解决大多数序列化问题

示例代码

package common
object AppFunctions {
  def append(s: String, start: Int) = s"${s}some thing"
}

object ExecuteQuery {
 import common.AppFunctions._

 [...]

 val d3 = d2.map(s => append(s,5)) // Pass required values to method.

 [...]


}