Spark jdbc批处理不插入所有记录答案

【问题标题】：Spark jdbc batch processing not inserting all recordsSpark jdbc批处理不插入所有记录
【发布时间】：2021-09-20 00:00:56
【问题描述】：

在我的 spark 作业中，我使用 jdbc 批处理将记录插入 MySQL。但我注意到所有记录都没有进入 MySQL。例如；

//count records before insert
println(s"dataframe: ${dataframe.count()}")

dataframe.foreachPartition(partition => {

  Class.forName(jdbcDriver)
  val dbConnection: Connection = DriverManager.getConnection(jdbcUrl, username, password)

  var preparedStatement: PreparedStatement = null
  dbConnection.setAutoCommit(false)
  val batchSize = 100

  partition.grouped(batchSize).foreach(batch => {
    batch.foreach(row => {
      val productName = row.getString(row.fieldIndex("productName"))
      val quantity = row.getLong(row.fieldIndex("quantity"))
      val sqlString =
        s"""
           |INSERT INTO myDb.product (productName, quantity)
           |VALUES (?, ?)
          """.stripMargin

      preparedStatement = dbConnection.prepareStatement(sqlString)
      preparedStatement.setString(1, productName)
      preparedStatement.setLong(2, quantity)

      preparedStatement.addBatch()
    })

    preparedStatement.executeBatch()
    dbConnection.commit()
    preparedStatement.close()
  })
  dbConnection.close()
})

我在dataframe.count 中看到了 650 条记录，但是当我检查 mysql 时，我看到了 195 条记录。这是确定性的。我尝试了不同的批量大小，但仍然看到相同的数字。但是当我将preparedStatement.executeBatch() 移动到batch.foreach() 中时，即preparedStatement.addBatch() 之后的下一行时，我在mysql 中看到了完整的650 条记录。它不再对插入语句进行批处理，因为它在将插入语句添加到单个语句后立即执行它迭代。阻止批处理查询的问题可能是什么？

【问题讨论】：

标签： mysql scala apache-spark jdbc

【解决方案1】：

您似乎在每次迭代中都创建了一个新的 preparedStatement，这意味着 preparedStatement.executeBatch() 仅应用于最后一批，即 195 条记录而不是 650 条记录。相反，您应该创建一个preparedStatement，然后替换迭代中的参数，如下所示：

dataframe.foreachPartition(partition => {

  Class.forName(jdbcDriver)
  val dbConnection: Connection = DriverManager.getConnection(jdbcUrl, username, password)

  val sqlString =
        s"""
           |INSERT INTO myDb.product (productName, quantity)
           |VALUES (?, ?)
          """.stripMargin

  var preparedStatement: PreparedStatement = dbConnection.prepareStatement(sqlString)

  dbConnection.setAutoCommit(false)
  val batchSize = 100

  partition.grouped(batchSize).foreach(batch => {
    batch.foreach(row => {
      val productName = row.getString(row.fieldIndex("productName"))
      val quantity = row.getLong(row.fieldIndex("quantity"))
      

      preparedStatement = dbConnection.prepareStatement(sqlString)
      preparedStatement.setString(1, productName)
      preparedStatement.setLong(2, quantity)

      preparedStatement.addBatch()
    })

    preparedStatement.executeBatch()
    dbConnection.commit()
    preparedStatement.close()
  })
  dbConnection.close()
})

【讨论】：