优化SparkSQL读取kudu数据向Mysql写入效率低的问题

vlambda
2020-04-20

优化SparkSQL读取kudu数据向Mysql写入效率低的问题

记录一次实际工作中SparkSQL向Mysql写入数据，效率很低很慢的解决过程。

原始代码如下：

object Kudu2Mysql {
 def main(args: Array[String]): Unit = {
 val sparkSession: SparkSession = SparkSession.builder() .master("local[6]")      .appName("kudu2Mysql") .getOrCreate()    val kuduMaster = "192.168.100.100:7051,192.168.100.100:7051,192.168.100.100:7051" val kuduTable = "kudu_test"    val kuduSql ="""select * from tmp_kuduTable where day = '2020-03-16'"""    val kuduDatas = sparkSqlReadKudu(sparkSession, kuduMaster, kuduTable, kuduSql)
    val mysqlTable = "kudu_test"    saveToMysql(kuduDatas, mysqlTable)
    sparkSession.stop()
 }
  private def sparkSqlReadKudu(sparkSession: SparkSession, kuduMaster: String, kuduTable: String, sql: String) = {
 val options = Map( "kudu.master" -> kuduMaster, "kudu.table" -> kuduTable ) val kuduDatas: DataFrame = sparkSession.read .format("org.apache.kudu.spark.kudu")      .options(options).load()       kuduDatas.createOrReplaceTempView("tmp_kuduTable") val datas: DataFrame = sparkSession.sql(sql) //datas.show()    datas     }
 private def saveToMysql(dataFrame: DataFrame, tableName:String): Unit ={
 dataFrame.write.format("jdbc").mode(SaveMode.Append)      .option("url", "jdbc:mysql://127.0.0.1:3306/kudu) .option("dbtable", tableName) .option("user", "root")      .option("password", "*********") .save()
 }

}

使用以上代码读取的kudu表数据为50000条，运行的速度很慢，插入几千条数据大概就要花费2分钟左右，数据大时，写入Mysql速度非常慢，且对Mysql造成的负载比较高。后来查询了一些资料，原来是因为每次插入数据都会启动mysql连接，非常消耗性能。查看官网资料，可以看到官方spark sql jdbc部分提供了参数。

url：要连接的JDBC URL。例如：jdbc:mysql://ip:3306

driver:用于连接到此URL的JDBC驱动程序的类名,列如：com.mysql.jdbc.Driver

batchsize：仅适用于write数据。JDBC批量大小，用于确定每次insert的行数。这可以帮助JDBC驱动程序调优性能。默认为1000。

isolationLevel：仅适用于write数据。事务隔离级别，适用于当前连接。它可以是一个NONE，READ_COMMITTED，READ_UNCOMMITTED，REPEATABLE_READ，或SERIALIZABLE，对应于由JDBC的连接对象定义，缺省值为标准事务隔离级别READ_UNCOMMITTED。

为此对saveToMysql方法中的参数进行设置，在url后面加上？rewriteBachedStatements=true，开启批量写入，然后设置批量写入的大小，batchsize，不开启事务。

修改后代码如下：

 private def saveToMysql(dataFrame: DataFrame, tableName:String): Unit ={
 dataFrame.write.format("jdbc").mode(SaveMode.Append) .option("url", "jdbc:mysql://127.0.0.1:3306/kudu?rewriteBatchedStatements=true) .option("dbtable", tableName) .option("user", "root") .option("password", "*********") .option("batchsize",10000)      .option("isolationLevel","NONE") .save()
 }