mysql 批量修改_优化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()                                                                                   }

修改完的代码，写入效率明显比原来的代码效率高很多。

关注更多IT技术，欢迎关注微信公众号：一点IT技术， 或扫描下方二维码关注一点IT技术

!!!

长安识别点击关注