Spark任务执行时遭遇Task Not Serializable异常求助

从你的错误栈信息可以明确看到问题根源：java.io.NotSerializableException: org.apache.spark.SparkContext，这是因为你在分布式执行的闭包（foreachPartition内部的代码）中直接引用了SparkContext和SQLContext对象——这些Spark核心上下文对象是不可序列化的，当Spark尝试把任务逻辑序列化后分发到Executor节点时，就会抛出这个异常。

问题代码中的核心问题点 #

1. 在foreachPartition的闭包里调用了context.parallelize(Seq(r1))，直接使用了Driver端的SparkContext
2. 闭包里使用sqlCotext.read.schema(schema).json(rdd.toDS)，同样依赖了不可序列化的SQLContext
3. 在Executor端对initialDF1进行union操作，不仅会导致序列化问题，还会引发分布式环境下的并发修改问题

修复方案与重构代码 #

下面是重构后的代码，核心思路是把Spark上下文相关操作限制在Driver端，Executor端只做纯数据处理（curl调用+JSON解析）：

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext, HiveContext}
import org.apache.spark.sql.types._
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import scala.util.Try

object th extends Serializable {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("th").setMaster("local")
    conf.set("spark.debug.maxToStringFields", "10000000")
    val context = new SparkContext(conf)
    val sqlContext = new SQLContext(context)
    val hiveContext = new HiveContext(context)
    import hiveContext.implicits._

    // 1. Driver端读取Hive数据，获取application_number列表
    val appNumList = hiveContext.sql("select application_number from tableA")
      .map(_.getAs[Long]("application_number")) // 直接提取Long类型的application_number
      .take(100)
    val appNumRDD = context.parallelize(appNumList).repartition(10)

    // 2. Driver端定义最终输出的Schema
    val outputSchema = StructType(
      StructField("application_number", LongType, true) ::
      StructField("event_code", StringType, true) ::
      StructField("event_description", StringType, true) ::
      StructField("event_recorded_date", StringType, true) :: Nil
    )

    // 3. 定义JSON解析的工具（Driver端初始化，广播到Executor）
    val mapper = new ObjectMapper()
    mapper.registerModule(DefaultScalaModule)
    val broadcastMapper = context.broadcast(mapper)

    // 4. 分布式处理：每个分区内调用curl并解析结果，返回Row
    val resultRDD = appNumRDD.mapPartitions { iter =>
      val localMapper = broadcastMapper.value
      iter.flatMap { appNum =>
        // 构造curl命令并执行
        val cmd = Seq(
          "curl", "-X", "POST", "--insecure",
          "--header", "Content-Type: application/json",
          "--header", "Accept: application/json",
          "-d", s"""{"searchText":"$appNum","qf":"applId"}""",
          "https://ped.uspto.gov/api/queries"
        )

        // 执行命令并处理结果（增加异常捕获，避免单个请求失败导致整个分区失败）
        Try(cmd.!!).toOption.flatMap { response =>
          // 解析JSON响应，提取transactions数据
          Try {
            val jsonNode = localMapper.readTree(response)
            val docs = jsonNode.at("/queryResults/searchResponse/response/docs")
            val transactions = docs.findValues("transactions").flatMap(_.elements()).flatMap(_.elements())
            
            // 把每个transaction转换成Row
            transactions.map { tx =>
              val code = tx.get("code").asText()
              val description = tx.get("description").asText()
              val recordDate = tx.get("recordDate").asText()
              Row(appNum, code, description, recordDate)
            }
          }.toOption.getOrElse(Seq.empty[Row])
        }
      }
    }

    // 5. Driver端将RDD转换为DataFrame，写入Hive
    val resultDF = sqlContext.createDataFrame(resultRDD, outputSchema)
    resultDF.registerTempTable("curlTH")
    hiveContext.sql("""
      INSERT INTO TABLE default.ipg_tableB 
      SELECT application_number, event_code, event_description, event_recorded_date 
      FROM curlTH
    """)

    context.stop()
  }
}

关键修复说明 #

1. 移除闭包中的Spark上下文引用：不再在Executor端使用SparkContext创建RDD，也不再用SQLContext解析JSON，改用纯Scala的Jackson库在Executor端完成JSON解析。
2. 广播JSON解析工具：把ObjectMapper通过广播变量分发到Executor，避免每个Task重复初始化，提升性能。
3. 异常安全处理：用Try包裹curl调用和JSON解析，避免单个请求失败导致整个分区任务失败。
4. 使用RDD的mapPartitions：直接在分区内处理数据并返回Row，最后在Driver端统一转换为DataFrame，避免分布式环境下的DataFrame并发修改问题。
5. 简化数据提取：从Hive读取时直接提取application_number的Long值，避免后续字符串替换的繁琐操作。

额外建议 #

• 考虑增加curl请求的重试机制，比如使用scala.util.control.Retry来处理临时网络问题。
• 控制并发请求数，避免对目标API（uspto.gov）造成过大压力，可通过调整repartition的分区数来控制并发度。
• 如果curl请求耗时较长，可考虑使用异步HTTP客户端（比如AsyncHttpClient）替代同步的cmd.!!，提升任务执行效率。

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