Scala用Jackson将RDD[String]转Map报Task not serializable错误怎么解决

org.apache.spark.SparkException: Task not serializable 报错的核心原因是Jackson的ObjectMapper本身不支持Spark序列化。你在Driver端创建的mapper实例被传入RDD算子时，Spark需要将该实例序列化后发送到Executor节点执行，而ObjectMapper未实现序列化接口，直接引用就会触发序列化失败。
额外说明：你给出的遍历代码还存在传参缺失问题，mapper.readValue[Map[String, Object]]没有传入待解析的line变量，需要修正为mapper.readValue[Map[String, Any]](line)才能正常解析。

两种方案均为分布式执行逻辑，所有计算运行在Executor端，无需将全量数据拉取到Driver，不会超出分配内存，适配任意大小的文件。

方案1：使用mapPartitions按分区初始化Mapper（最推荐） #

每个分区仅创建一次Mapper实例，既避免序列化问题，也不会因单条数据创建一次Mapper导致性能损耗，是生产环境的标准处理方案：

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.ScalaObjectMapper

val lines = sc.textFile(fileName)

val processedRDD = lines.mapPartitions(iter => {
  // 分区内本地初始化Mapper，不需要从Driver端传输
  val mapper = new ObjectMapper() with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  iter.map(line => {
    // 解析JSON为Map
    val jsonMap = mapper.readValue[Map[String, Any]](line)
    // 此处编写你对Map的后续处理逻辑，最后返回处理完成的字符串即可
    // 示例：将Map转为指定格式字符串
    jsonMap.map { case (k, v) => s"$k:$v" }.mkString("|")
  })
})

// 直接输出结果到存储，无需collect/take全量数据到Driver
processedRDD.saveAsTextFile("your_output_path")

方案2：用@transient lazy val标记Mapper #

如果需要在多个算子中复用Mapper实例，可以在Driver端将Mapper声明为瞬态懒加载，序列化时会忽略该实例，Executor执行到对应逻辑时自动本地初始化：

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.ScalaObjectMapper

// 声明为瞬态懒加载，避免序列化
@transient lazy val mapper = {
  val mapper = new ObjectMapper() with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  mapper
}

val lines = sc.textFile(fileName)
val processedRDD = lines.map(line => {
  val jsonMap = mapper.readValue[Map[String, Any]](line)
  // 后续处理逻辑
})

processedRDD.saveAsTextFile("your_output_path")

内容的提问来源于stack exchange，提问作者NMcD