Scala/Spark中mapPartitions无法读取外部val变量的原因排查

这是Spark闭包序列化机制导致的问题：

• product_type是在Driver端定义的变量，但mapPartitions的逻辑会分发到Executor节点执行。Spark需要将Driver端的变量序列化后传递给Executor，如果你的Config类没有正确实现Serializable接口，或者序列化过程中product_type的值未被正确保留，Executor拿到的变量值就会和Driver端不一致，从而触发else分支加载my_bb.hdf5。
• 你在Driver端打印的product_type值正确，仅代表Driver本地的变量状态，不代表序列化后传递到Executor的版本也正确。

解决步骤： #

1. 确保Config类可序列化
   让Config类实现Serializable接口，保证Spark能正确序列化并传递其内部的product_type值：

   class Config(args: Array[String]) extends Serializable {
     // 你的Config类实现逻辑
     val product_type: String = ...
   }
   

2. 显式提取独立字符串变量
   直接将product_type提取为独立的String变量，避免依赖Config对象的序列化：

   val config = Config(args)
   val product_type: String = config.product_type // 显式声明为String类型，确保可被正确序列化
   

3. 使用广播变量传递
   对于需要在Executor端复用的变量，推荐使用Spark广播变量，既保证传递的正确性，又能减少网络传输开销：

   val config = Config(args)
   val product_type = config.product_type
   val ptBroadcast = spark.sparkContext.broadcast(product_type)
   
   val scores = df.mapPartitions(iterator => {
     val currentPt = ptBroadcast.value
     val inputStream =
       if(currentPt == "AA" ) {
         getClass().getClassLoader().getResourceAsStream("my_aa.hdf5")
       } else {
         getClass().getClassLoader().getResourceAsStream("my_bb.hdf5")
       } 
     // 后续模型加载与处理逻辑
     val multiLayerNetwork: MultiLayerNetwork = KerasModelImport.importKerasSequentialModelAndWeights(inputStream, false)
     val wrapped: ParallelInference = new ParallelInference.Builder(multiLayerNetwork).build()
     iterator.map(row => wrapped.output(row).toDoubleVector)
   })
   

4. 验证Executor端变量值
   在mapPartitions内添加日志打印，确认Executor实际接收的product_type值，方便排查问题：

   val scores = df.mapPartitions(iterator => {
     println(s"Executor received product_type: $product_type") // 查看Executor节点的日志输出
     // 后续逻辑
   })
   

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