导论

笔者的毕业设计是一个与大数据相关的项目，其中遇到了一个将百万级别数据存储到MySQL 的问题，本来是打算使用Spark SQL 的DataFrame 自带的API来直接将数据存储到MySQL 中，可是当我看到MySQL 监控平台的数据，我瞬间被打了脸，没那么简单，在监控平台上面显示着平均每5 秒插入140条左右的数据，笔者瞬间被吓到了，心想，这可不行，按这个速度，那要运行到何年何月，然后笔者自行Google 了一下，决定使用JDBC 批处理。

Java 高性能批处理

MySQL 之rewriteBatchedStatements

这是MySQL 批处理的一个重要参数，默认为false

当rewriteBatchedStatements 为false 时

执行插入的SQL 语句是

insert into ***_table values (***)

当rewriteBatchedStatements 为true 时

insert语句，满成条件情况下，会整合成形如：

insert into xxx_table values (aa),(bb),(cc).

然后分批次发送给MySQL(会有一次发送的package大小限制，所以需要拆分批次)

示例代码(这里使用的是Scala，其实调用的都是Java 的库)

• 设置rewriteBatchedStatements 为true

val mysqlConfProd = collection.mutable.Map(  
    "driver" -> "com.mysql.jdbc.Driver",  
    "url" -> "jdbc:mysql://******.com:3306/behavior_db?rewriteBatchedStatements=true",  
    "username" -> "root",  
    "password" -> "root"
)

• 执行批处理程序

var index = 0val begin = System.currentTimeMillis()for (item <- items) {  index = index + 1  
// 每次批处理插入10000 条数据  if (index >= 10000) {    println("执行批处理任务，插入一万条数据")    
    preparedStatement.executeBatch()    
    mysql.commit()    
    println("插入成功")   
    index = 1  
}  
preparedStatement.setString(1, item.getString(0))  
preparedStatement.setString(2, item.getString(1))  
preparedStatement.setLong(3, item.getLong(2))  
preparedStatement.setString(4, item.getString(3))  
preparedStatement.addBatch()}

// 执行批处理，完成剩下的任务
preparedStatement.executeBatch()
mysql.commit()
val end = System.currentTimeMillis()
println(end - begin)

测试结果，插入三百万条数据，耗时68.047 秒

耗时还是十分可观的，可见JDBC 的批处理功能还是十分强大的！

运行时截图