Spark使用requests-scala调用API抛出NotSerializableException错误

抛出java.io.NotSerializableException: requests.RequestAuth$Empty$的核心触发逻辑是Spark的闭包序列化校验机制：你将requests-scala的Session实例作为UDF的外部捕获变量传入了执行逻辑，而Session内部默认关联的空鉴权单例对象RequestAuth.Empty未实现Java序列化接口，Spark在将任务序列化分发到Executor节点执行时，会对UDF捕获的所有外部变量做全量序列化校验，碰到不可序列化的对象就会直接抛出该异常。

这类问题的典型错误写法是在Driver端创建全局Session实例，被UDF直接捕获引用：

// 错误实现：跨节点传递不可序列化的Session对象
val session = requests.Session()
val callExternalApiUdf = udf((inputData: String) => {
  // UDF捕获了外层Driver端创建的session对象，触发序列化校验失败
  session.post("外部API服务地址", data = inputData)
})

该写法下，Driver提交任务时会把整个UDF连同捕获的session实例一起序列化后发送给各个Executor节点，而requests.Session持有的RequestAuth$Empty单例不支持序列化，直接中断作业提交流程。

• 禁止在Driver端创建requests-scala的Session/Request相关全局实例作为UDF的捕获变量，这类HTTP客户端实例本身持有TCP连接、本地线程池等非可序列化资源，跨节点传输后也无法正常工作。
• 优先采用Executor本地初始化实例的方式发起调用：如果是简单的单次请求，可以直接在UDF逻辑内部发起请求，无外部对象捕获，自然不会触发序列化问题。
  示例代码：

  // 简单实现：UDF内部直接发起请求，无外部捕获变量
  val callExternalApiUdf = udf((inputData: String) => {
    requests.post("外部API服务地址", data = inputData)
  })
  

• 如果需要复用连接池能力降低请求开销，不要将Session作为UDF捕获变量，改为借助mapPartitions接口在每个分区的本地执行上下文初始化Session，分区内所有数据处理复用该本地实例，分区处理完成后随Executor内存回收即可。
  示例代码：

  // 连接池复用实现：本地初始化Session，无跨节点序列化
  inputDataFrame.rdd.mapPartitions { partitionIterator =>
    // 每个Executor处理每个分区时，本地创建Session实例
    val localSession = requests.Session()
    partitionIterator.map { row =>
      val result = localSession.post("外部API服务地址", data = row.getString(0))
      result.text()
    }
  }.toDF().writeStream // 后续结果写Kafka逻辑
  

• 不要尝试通过包装Session类实现序列化接口的方式规避异常，这类对象持有本地网络资源，跨节点反序列化后无法建立有效网络连接。
• 出现序列化异常时可开启Spark闭包调试参数：-Dspark.streaming.closure.debug=true，快速定位未实现序列化的对象位置。
• 注意requests-scala的RequestAuth.Empty是全局单例的不可序列化对象，不要在Driver端定义任何包含该对象的全局变量被UDF捕获引用。

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