使用 Spark 数据集转换的 TaskNotSerializable答案

【问题标题】：TaskNotSerializable with Spark Dataset transformations使用 Spark 数据集转换的 TaskNotSerializable
【发布时间】：2021-11-05 18:55:05
【问题描述】：

我正在尝试按其中一个字段过滤 Spark 数据集，但由于转换，我收到了 TaskNotSerializable。该方法如下所示：

def filterData(input: Dataset[CustomType], idsList: List[Int]): Dataset[CustomType] =
   input.flatMap { record =>
   val filtered = record.data.filter(rec => idsList.contains(rec.id))
   if (filtered.nonEmpty)
     Seq(record.withFields(filtered))
   else
     Iterable.empty
   }

我尝试在 Spark 数据帧上进行类似的转换，并且没有任何问题：

input.withColumn("arr", explode($”data”))
  .filter($"arr.id".isin(idsList: _*)) //List(34,81,95)
  .drop("arr")
  .as[CustomType]

如何修复数据集转换以避免此错误？

自定义类型结构如下：

|-- url: string (nullable = true)

|-- data: array (nullable = true)

|    |-- element: struct (containsNull = true)

|    |    |-- id: integer (nullable = false)

|    |    |-- expiration: long (nullable = false)

|    |    |-- weight: integer (nullable = false)

|-- queryParams: array (nullable = true)

|-- element: string (containsNull = true)

错误序列化堆栈如下所示：

  Serialization stack:

- object not serializable (class: job.DataSink, value: job.DataSink@6dda8f39)

- field (class: job.DataSink$$anonfun$filterData$1, name: $outer, type: class job.DataSink)

- object (class job.DataSink$$anonfun$filterData$1, <function1>)

- field (class: org.apache.spark.sql.Dataset$$anonfun$flatMap$1, name: func$6, type: interface scala.Function1)

- object (class org.apache.spark.sql.Dataset$$anonfun$flatMap$1, <function1>)

- field (class: org.apache.spark.sql.execution.MapPartitionsExec, name: func, type: interface scala.Function1)

- object (class org.apache.spark.sql.execution.MapPartitionsExec, MapPartitions <function1>, obj#392: schemas.data.CustomType

 +- Scan[obj#383]

 )

【问题讨论】：

通常当有TaskNotSerializable 时，Spark 会生成一种堆栈跟踪，表明他在失败时尝试序列化的内容，这样我们就可以看到它是哪个数据结构。您可以浏览您的日志并找到它，还是只发布整个内容，我会尝试找到它？它看起来像这样Serialization stack: - object not serializable (class: testing, value: testing@2dfe2f00) - field (class: testing$$anonfun$1, name: $outer, type: class testing) - object (class testing$$anonfun$1, <function1>)
@Filip 当然，我已经在问题中添加了序列化堆栈
我还需要更多信息，但我最好的猜测是我用可用的 ATM 信息回答的。

标签： scala apache-spark apache-spark-sql apache-spark-dataset

【解决方案1】：

从序列化堆栈中，我会说这个函数是job.DataSink 类的一部分，Spark 正在尝试序列化整个类以将其发送给工作者。如果你想让它被 Spark 序列化，你需要实现Serializable。

【讨论】：

是的，filterData函数不能序列化。有没有办法在不实现 Serializable 的情况下优化方法本身？
很难说没有剩下的代码还能做什么。在某些情况下，可以在 Spark 作业代码中创建变量 lazy，这基本上会导致 Spark 将序列化代码发送给工作人员，而不是序列化对象实例。然后 Spark 会在 worker 上初始化对象并调用函数。在这种情况下，您不需要实现Serializable。为什么要避免实施Serializable？
更多的是代码库风格问题，而不是偏好