lambda 查询大量数据速度很慢_优化SparkSQL读取kudu数据向Mysql写入效率低的问题

雨点打透心脏的1/2处 2023-01-01 05:48 173阅读 0赞

记录一次实际工作中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()\}

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

!!!