Spark2升级至Spark3后DataFrame的foreachPartition方法无法正常运行

开发者阿泽

2026-6-28

问题分析与修复方案

你的代码在Spark 3.4.2 + Scala 2.12.15下无法运行，核心是代码本身存在多个硬伤，和版本升级的语法规则变化关联不大，下面逐个拆解问题并给出修复方案：

不可变变量赋值错误：外部定义的val connection: Connection = null是不可变的，在foreachPartition里给它赋值直接违反Scala的val规则，必须把connection的定义放到分区处理逻辑内部，用var声明。
变量名不匹配：代码里从Row取的是id、a_id，但SQL参数里写的是lead_id、agency_id，这两个变量根本没定义，直接会编译报错，得把参数名改成和你取的变量一致。
PreparedStatement创建位置错误：现在你在每个Row循环里都新建一个PreparedStatement，等于放弃了批量插入的意义，应该在每个batch的外面创建一次，循环添加批量操作即可。
数据类型不匹配：Seq里的id是Int类型（1、2），你用row.getString(IdIndx)去取会抛出类型转换异常，得用row.getInt(IdIndx)，再根据数据库字段类型转成对应格式。
资源泄漏风险：没有任何异常捕获和资源释放逻辑，一旦中间出错，数据库连接和Statement不会被关闭，必须用try-catch-finally或者Scala的Using语法（Scala 2.12+支持）来保证资源释放。
闭包序列化问题：外部定义的JDBC配置变量虽然是String，但放到foreachPartition内部更安全，避免Spark序列化闭包时的潜在问题。

修复后的完整代码：

import java.sql.{Connection, DriverManager, PreparedStatement}
import scala.util.Using

// 定义JDBC配置
val targetDbDriver: String = "org.postgresql.Driver"
val targetDbUrl: String = "jdbc:postgresql://server:portnumber/Dbname"
val targetDbUser: String = "*****"
val targetDbPassword: String = "******"
val targetDbSchema: String = "abc"

// 构造测试DataFrame
val df = Seq((1, "a1", "aaaa"), (2, "b1", "bbbb"))
  .toDF("id", "a_id", "first_nm")

df.coalesce(1).foreachPartition(parti => {
  // 加载驱动（Spark 3.x里JDBC4.0以上可自动加载，保留也没问题）
  Class.forName(targetDbDriver)

  // 使用Using语法自动管理连接和Statement资源
  Using(DriverManager.getConnection(targetDbUrl, targetDbUser, targetDbPassword)) { connection =>
    val batchSize = 10
    // 提前定义SQL，避免重复拼接
    val sql = s"INSERT INTO ${targetDbSchema}.agency_leads VALUES (?,?,?) " +
              "ON CONFLICT (lead_id) " +
              "DO UPDATE SET " +
              "agency_id = EXCLUDED.agency_id, " +
              "first_nm = EXCLUDED.first_nm"
    
    Using(connection.prepareStatement(sql)) { updatePrepareStmt =>
      parti.grouped(batchSize).foreach(batch => {
        batch.foreach { row =>
          // 正确获取对应类型的字段值
          val id = row.getInt(row.fieldIndex("id"))
          val aId = row.getString(row.fieldIndex("a_id"))
          val firstNm = row.getString(row.fieldIndex("first_nm"))

          // 设置参数，注意索引从1开始
          updatePrepareStmt.setInt(1, id)
          updatePrepareStmt.setString(2, aId)
          updatePrepareStmt.setString(3, firstNm)

          updatePrepareStmt.addBatch()
        }
        // 执行批量操作
        updatePrepareStmt.executeBatch()
        // 清空当前batch的参数，避免下一批次干扰
        updatePrepareStmt.clearBatch()
      })
    }
  }
})

额外说明：

用Scala的Using工具类（Scala 2.12.1+引入）可以自动关闭实现了AutoCloseable的资源（Connection、PreparedStatement都实现了），不用手动写finally，更简洁安全。
把SQL语句提前定义在batch循环外面，避免重复拼接字符串，提升性能。
执行完batch后调用clearBatch()清空参数，防止下一批次的参数残留。
如果你的数据库lead_id字段是String类型，把setInt(1, id)改成setString(1, id.toString)即可。

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