Spark:将DataFrame写入Mysql

Spark将DataFrame进行一些列处理后,需要将之写入mysql,下面是实现过程

1.mysql的信息

mysql的信息我保存在了外部的配置文件,这样方便后续的配置添加。

1 //配置文件示例:
2 [hdfs@iptve2e03 tmp_lillcol]$ cat job.properties 
3 #mysql数据库配置
4 mysql.driver=com.mysql.jdbc.Driver
5 mysql.url=jdbc:mysql://127.0.0.1:3306/database1?useSSL=false&autoReconnect=true&failOverReadOnly=false&rewriteBatchedStatements=true
6 mysql.username=user
7 mysql.password=123456

2.需要的jar依赖(sbt版本,maven的对应修改即可)

 1 libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0-cdh5.7.2"
 2 libraryDependencies += "org.apache.spark" % "spark-sql_2.10" % "1.6.0-cdh5.7.2"
 3 libraryDependencies += "org.apache.spark" % "spark-hive_2.10" % "1.6.0-cdh5.7.2"
 4 libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.2.0-cdh5.7.2"
 5 libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.2.0-cdh5.7.2"
 6 libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.2.0-cdh5.7.2"
 7 libraryDependencies += "org.apache.hbase" % "hbase-protocol" % "1.2.0-cdh5.7.2"
 8 libraryDependencies += "mysql" % "mysql-connector-java" % "5.1.38"
 9 libraryDependencies += "org.apache.spark" % "spark-streaming_2.10" % "1.6.0-cdh5.7.2"
10 libraryDependencies += "com.yammer.metrics" % "metrics-core" % "2.2.0"

 

3.完整实现代码

 

  1 import java.io.FileInputStream
  2 import java.sql.{Connection, DriverManager}
  3 import java.util.Properties
  4 
  5 import org.apache.spark.sql.hive.HiveContext
  6 import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
  7 import org.apache.spark.{SparkConf, SparkContext}
  8 
  9 /**
 10   * @author Administrator
 11   *         2018/10/16-10:15
 12   *
 13   */
 14 object SaveDataFrameASMysql {
 15   var hdfsPath: String = ""
 16   var proPath: String = ""
 17   var DATE: String = ""
 18 
 19   val sparkConf: SparkConf = new SparkConf().setAppName(getClass.getSimpleName)
 20   val sc: SparkContext = new SparkContext(sparkConf)
 21   val sqlContext: SQLContext = new HiveContext(sc)
 22 
 23   def main(args: Array[String]): Unit = {
 24     hdfsPath = args(0)
 25     proPath = args(1)
 26     //不过滤读取
 27     val dim_sys_city_dict: DataFrame = readMysqlTable(sqlContext, "TestMysqlTble1", proPath)
 28     dim_sys_city_dict.show(10)
 29 
 30     //保存mysql
 31     saveASMysqlTable(dim_sys_city_dict, "TestMysqlTble2", SaveMode.Append, proPath)
 32   }
 33 
 34   /**
 35     * 将DataFrame保存为Mysql表
 36     *
 37     * @param dataFrame 需要保存的dataFrame
 38     * @param tableName 保存的mysql 表名
 39     * @param saveMode  保存的模式 :Append、Overwrite、ErrorIfExists、Ignore
 40     * @param proPath   配置文件的路径
 41     */
 42   def saveASMysqlTable(dataFrame: DataFrame, tableName: String, saveMode: SaveMode, proPath: String) = {
 43     var table = tableName
 44     val properties: Properties = getProPerties(proPath)
 45     val prop = new Properties //配置文件中的key 与 spark 中的 key 不同 所以 创建prop 按照spark 的格式 进行配置数据库
 46     prop.setProperty("user", properties.getProperty("mysql.username"))
 47     prop.setProperty("password", properties.getProperty("mysql.password"))
 48     prop.setProperty("driver", properties.getProperty("mysql.driver"))
 49     prop.setProperty("url", properties.getProperty("mysql.url"))
 50     if (saveMode == SaveMode.Overwrite) {
 51       var conn: Connection = null
 52       try {
 53         conn = DriverManager.getConnection(
 54           prop.getProperty("url"),
 55           prop.getProperty("user"),
 56           prop.getProperty("password")
 57         )
 58         val stmt = conn.createStatement
 59         table = table.toUpperCase
 60         stmt.execute(s"truncate table $table") //为了不删除表结构,先truncate 再Append
 61         conn.close()
 62       }
 63       catch {
 64         case e: Exception =>
 65           println("MySQL Error:")
 66           e.printStackTrace()
 67       }
 68     }
 69     dataFrame.write.mode(SaveMode.Append).jdbc(prop.getProperty("url"), table, prop)
 70   }
 71 
 72   /**
 73     * 获取 Mysql 表的数据
 74     *
 75     * @param sqlContext
 76     * @param tableName 读取Mysql表的名字
 77     * @param proPath   配置文件的路径
 78     * @return 返回 Mysql 表的 DataFrame
 79     */
 80   def readMysqlTable(sqlContext: SQLContext, tableName: String, proPath: String) = {
 81     val properties: Properties = getProPerties(proPath)
 82     sqlContext
 83       .read
 84       .format("jdbc")
 85       .option("url", properties.getProperty("mysql.url"))
 86       .option("driver", properties.getProperty("mysql.driver"))
 87       .option("user", properties.getProperty("mysql.username"))
 88       .option("password", properties.getProperty("mysql.password"))
 89       //        .option("dbtable", tableName.toUpperCase)
 90       .option("dbtable", tableName)
 91       .load()
 92 
 93   }
 94 
 95   /**
 96     * 获取 Mysql 表的数据 添加过滤条件
 97     *
 98     * @param sqlContext
 99     * @param table           读取Mysql表的名字
100     * @param filterCondition 过滤条件
101     * @param proPath         配置文件的路径
102     * @return 返回 Mysql 表的 DataFrame
103     */
104   def readMysqlTable(sqlContext: SQLContext, table: String, filterCondition: String, proPath: String) = {
105     val properties: Properties = getProPerties(proPath)
106     var tableName = ""
107     tableName = "(select * from " + table + " where " + filterCondition + " ) as t1"
108     sqlContext
109       .read
110       .format("jdbc")
111       .option("url", properties.getProperty("mysql.url"))
112       .option("driver", properties.getProperty("mysql.driver"))
113       .option("user", properties.getProperty("mysql.username"))
114       .option("password", properties.getProperty("mysql.password"))
115       .option("dbtable", tableName)
116       .load()
117   }
118 
119   /**
120     * 获取配置文件
121     *
122     * @param proPath
123     * @return
124     */
125   def getProPerties(proPath: String) = {
126     val properties: Properties = new Properties()
127     properties.load(new FileInputStream(proPath))
128     properties
129   }
130 }

 

 

 

4.测试

 1 def main(args: Array[String]): Unit = {
 2 hdfsPath = args(0)
 3 proPath = args(1)
 4 //不过滤读取
 5 val dim_sys_city_dict: DataFrame = readMysqlTable(sqlContext, "TestMysqlTble1", proPath)
 6 dim_sys_city_dict.show(10)
 7 
 8 //保存mysql
 9 saveASMysqlTable(dim_sys_city_dict, "TestMysqlTble2", SaveMode.Append, proPath)
10 }

 

5.运行结果数据敏感进行过处理

 1 +-------+-------+---------+---------+--------+----------+---------+--------------------+----+-----------+
 2 |dict_id|city_id|city_name|city_code|group_id|group_name|area_code| bureau_id|sort|bureau_name|
 3 +-------+-------+---------+---------+--------+----------+---------+--------------------+----+-----------+
 4 |     1|    249|       **|    **_ab|     100|      **按时|    **-查到|xcaasd...| 21|    张三公司|
 5 |     2|    240|       **|    **_ab|     300|      **按时|    **-查到|xcaasd...| 21|    张三公司|
 6 |     3|    240|       **|    **_ab|     100|      **按时|    **-查到|xcaasd...| 21|    张三公司|
 7 |     4|    242|       **|    **_ab|     300|      **按时|    **-查到|xcaasd...| 01|    张三公司|
 8 |     5|    246|       **|    **_ab|     100|      **按时|    **-查到|xcaasd...| 01|    张三公司|
 9 |     6|    246|       **|    **_ab|     300|      **按时|    **-查到|xcaasd...| 01|    张三公司|
10 |     7|    248|       **|    **_ab|     200|      **按时|    **-查到|xcaasd...| 01|    张三公司|
11 |     8|    242|       **|    **_ab|     400|      **按时|    **-查到|xcaasd...| 01|    张三公司|
12 |     9|    247|       **|    **_ab|     200|      **按时|    **-查到|xcaasd...| 01|    张三公司|
13 |     0|    243|       **|    **_ab|     400|      **按时|    **-查到|xcaasd...| 01|    张三公司|
14 +-------+-------+---------+---------+--------+----------+---------+--------------------+----+-----------+
15 
16 mysql> desc TestMysqlTble1;
17 +-------------+-------------+------+-----+---------+-------+
18 | Field       | Type        | Null | Key | Default | Extra |
19 +-------------+-------------+------+-----+---------+-------+
20 | dict_id     | varchar(32) | YES  |     | NULL    |       |
21 | city_id     | varchar(32) | YES  |     | NULL    |       |
22 | city_name   | varchar(32) | YES  |     | NULL    |       |
23 | city_code   | varchar(32) | YES  |     | NULL    |       |
24 | group_id    | varchar(32) | YES  |     | NULL    |       |
25 | group_name  | varchar(32) | YES  |     | NULL    |       |
26 | area_code   | varchar(32) | YES  |     | NULL    |       |
27 | bureau_id   | varchar(64) | YES  |     | NULL    |       |
28 | sort        | varchar(32) | YES  |     | NULL    |       |
29 | bureau_name | varchar(32) | YES  |     | NULL    |       |
30 +-------------+-------------+------+-----+---------+-------+
31 10 rows in set (0.00 sec)
32 
33 mysql> desc TestMysqlTble2;
34 +-------------+------+------+-----+---------+-------+
35 | Field       | Type | Null | Key | Default | Extra |
36 +-------------+------+------+-----+---------+-------+
37 | dict_id     | text | YES  |     | NULL    |       |
38 | city_id     | text | YES  |     | NULL    |       |
39 | city_name   | text | YES  |     | NULL    |       |
40 | city_code   | text | YES  |     | NULL    |       |
41 | group_id    | text | YES  |     | NULL    |       |
42 | group_name  | text | YES  |     | NULL    |       |
43 | area_code   | text | YES  |     | NULL    |       |
44 | bureau_id   | text | YES  |     | NULL    |       |
45 | sort        | text | YES  |     | NULL    |       |
46 | bureau_name | text | YES  |     | NULL    |       |
47 +-------------+------+------+-----+---------+-------+
48 10 rows in set (0.00 sec)
49 
50 
51 mysql> select count(1) from TestMysqlTble1;
52 +----------+
53 | count(1) |
54 +----------+
55 |       21 |
56 +----------+
57 1 row in set (0.00 sec)
58 
59 mysql> select count(1) from TestMysqlTble2;
60 +----------+
61 | count(1) |
62 +----------+
63 |       21 |
64 +----------+
65 1 row in set (0.00 sec)

 

6.效率问题

一开始直接这么用的时候小数据还没什么,但是数据量大一点的时候速度就不行了,于是想方设法的想优化一下,用了几个手段效果不明显,然后进去看源代码,发现了两个关键的片段

 1  /**
 2    * Saves the content of the [[DataFrame]] to a external database table via JDBC. In the case the
 3    * table already exists in the external database, behavior of this function depends on the
 4    * save mode, specified by the `mode` function (default to throwing an exception).
 5    *
 6    * Don't create too many partitions in parallel on a large cluster; otherwise Spark might crash
 7    * your external database systems.
 8    *
 9    * @param url JDBC database url of the form `jdbc:subprotocol:subname`
10    * @param table Name of the table in the external database.
11    * @param connectionProperties JDBC database connection arguments, a list of arbitrary string
12    *                             tag/value. Normally at least a "user" and "password" property
13    *                             should be included.
14    *
15    * @since 1.4.0
16    */
17   def jdbc(url: String, table: String, connectionProperties: Properties): Unit = {
18     val props = new Properties()
19     extraOptions.foreach { case (key, value) =>
20       props.put(key, value)
21     }
22     // connectionProperties should override settings in extraOptions
23     props.putAll(connectionProperties)
24     val conn = JdbcUtils.createConnectionFactory(url, props)()
25 
26     try {
27       var tableExists = JdbcUtils.tableExists(conn, url, table)
28 
29       if (mode == SaveMode.Ignore && tableExists) {
30         return
31       }
32 
33       if (mode == SaveMode.ErrorIfExists && tableExists) {
34         sys.error(s"Table $table already exists.")
35       }
36 
37       if (mode == SaveMode.Overwrite && tableExists) {
38         JdbcUtils.dropTable(conn, table)
39         tableExists = false
40       }
41 
42       // Create the table if the table didn't exist.
43       if (!tableExists) {
44         val schema = JdbcUtils.schemaString(df, url)
45         val sql = s"CREATE TABLE $table ($schema)"
46         val statement = conn.createStatement
47         try {
48           statement.executeUpdate(sql)
49         } finally {
50           statement.close()
51         }
52       }
53     } finally {
54       conn.close()
55     }
56 
57     JdbcUtils.saveTable(df, url, table, props)//-----------------------------关键点1
58   }
59 
60 
61   /**
62    * Saves the RDD to the database in a single transaction.
63    */
64   def saveTable(
65       df: DataFrame,
66       url: String,
67       table: String,
68       properties: Properties) {
69     val dialect = JdbcDialects.get(url)
70     val nullTypes: Array[Int] = df.schema.fields.map { field =>
71       getJdbcType(field.dataType, dialect).jdbcNullType
72     }
73 
74     val rddSchema = df.schema
75     val getConnection: () => Connection = createConnectionFactory(url, properties)
76     val batchSize = properties.getProperty("batchsize", "1000").toInt
77     df.foreachPartition { iterator => //------------------------------------关键点2
78       savePartition(getConnection, table, iterator, rddSchema, nullTypes, batchSize, dialect)
79     }
80   }

 

也就是说,自带的方法就是按照分区来存的,每一个分区开启一个mysql连接,所以最简单的优化方式就是在保存之前对DataFrame进行重新分区,注意数据倾斜问题,不然可能效率没有提升。
当然目前测试过最快的就是文件拿下来直接通过load data的命令导入mysql,但是这个比较麻烦。

下面是分区示例

 1 def main(args: Array[String]): Unit = {
 2     hdfsPath = args(0)
 3     proPath = args(1)
 4     //不过滤读取
 5     val dim_sys_city_dict: DataFrame = readMysqlTable(sqlContext, "TestMysqlTble1", proPath)
 6     dim_sys_city_dict.show(10)
 7 
 8     //保存mysql
 9     saveASMysqlTable(dim_sys_city_dict.repartition(10), "TestMysqlTble2", SaveMode.Append, proPath)
10   }

7.总结

 将DataFrame写入mysql有几点需要注意的地方:

  • 需要保存的表最好事先建好,否则字段类型会使用默认的,Text类型实在是耗资源,对比前后两张表,下面分别为源表TestMysqlTble1和DataFrame保存的mysql表TestMysqlTble2
 1 mysql> desc TestMysqlTble1;                
 2 +-------------+-------------+------+-----+---------+-------+
 3 | Field       | Type        | Null | Key | Default | Extra |
 4 +-------------+-------------+------+-----+---------+-------+
 5 | dict_id     | varchar(32) | YES  |     | NULL    |       |
 6 | city_id     | varchar(32) | YES  |     | NULL    |       |
 7 | city_name   | varchar(32) | YES  |     | NULL    |       |
 8 | city_code   | varchar(32) | YES  |     | NULL    |       |
 9 | group_id    | varchar(32) | YES  |     | NULL    |       |
10 | group_name  | varchar(32) | YES  |     | NULL    |       |
11 | area_code   | varchar(32) | YES  |     | NULL    |       |
12 | bureau_id   | varchar(64) | YES  |     | NULL    |       |
13 | sort        | varchar(32) | YES  |     | NULL    |       |
14 | bureau_name | varchar(32) | YES  |     | NULL    |       |
15 +-------------+-------------+------+-----+---------+-------+
16 10 rows in set (0.00 sec)
17 
18 mysql> desc TestMysqlTble2;
19 +-------------+------+------+-----+---------+-------+
20 | Field       | Type | Null | Key | Default | Extra |
21 +-------------+------+------+-----+---------+-------+
22 | dict_id     | text | YES  |     | NULL    |       |
23 | city_id     | text | YES  |     | NULL    |       |
24 | city_name   | text | YES  |     | NULL    |       |
25 | city_code   | text | YES  |     | NULL    |       |
26 | group_id    | text | YES  |     | NULL    |       |
27 | group_name  | text | YES  |     | NULL    |       |
28 | area_code   | text | YES  |     | NULL    |       |
29 | bureau_id   | text | YES  |     | NULL    |       |
30 | sort        | text | YES  |     | NULL    |       |
31 | bureau_name | text | YES  |     | NULL    |       |
32 +-------------+------+------+-----+---------+-------+
33 10 rows in set (0.00 sec)
  •  关于 SaveMode.Overwrite 
 1 def jdbc(url: String, table: String, connectionProperties: Properties): Unit = {
 2     val props = new Properties()
 3     extraOptions.foreach { case (key, value) =>
 4       props.put(key, value)
 5     }
 6     // connectionProperties should override settings in extraOptions
 7     props.putAll(connectionProperties)
 8     val conn = JdbcUtils.createConnectionFactory(url, props)()
 9 
10     try {
11       var tableExists = JdbcUtils.tableExists(conn, url, table)
12 
13       if (mode == SaveMode.Ignore && tableExists) {
14         return
15       }
16 
17       if (mode == SaveMode.ErrorIfExists && tableExists) {
18         sys.error(s"Table $table already exists.")
19       }
20 
21       if (mode == SaveMode.Overwrite && tableExists) {
22         JdbcUtils.dropTable(conn, table)//----------------------------------------关键点1
23         tableExists = false
24       }
25 
26       // Create the table if the table didn't exist.
27       if (!tableExists) {
28         val schema = JdbcUtils.schemaString(df, url)
29         val sql = s"CREATE TABLE $table ($schema)"
30         val statement = conn.createStatement
31         try {
32           statement.executeUpdate(sql)
33         } finally {
34           statement.close()
35         }
36       }
37     } finally {
38       conn.close()
39     }
40 
41     JdbcUtils.saveTable(df, url, table, props)
42   }
43 
44  /**
45    * Drops a table from the JDBC database.
46    */
47   def dropTable(conn: Connection, table: String): Unit = {
48     val statement = conn.createStatement
49     try {
50       statement.executeUpdate(s"DROP TABLE $table")//-------------------------------------关键点2
51     } finally {
52       statement.close()
53     }
54   }

 

从上述两段关键代码可以看到,在写入的时候会先判断表存不存在,SaveMode.Overwrite 的时候会执行 dropTable(conn: Connection, table: String)把原来的表删除掉,这也意味着你会失去你的表结构,新建的表会出现上一个问题都用默认类型,所以在保存的方法中我添加了下面的操作

 1 if (saveMode == SaveMode.Overwrite) {
 2  51       var conn: Connection = null
 3  52       try {
 4  53         conn = DriverManager.getConnection(
 5  54           prop.getProperty("url"),
 6  55           prop.getProperty("user"),
 7  56           prop.getProperty("password")
 8  57         )
 9  58         val stmt = conn.createStatement
10  59         table = table.toUpperCase
11  60         stmt.execute(s"truncate table $table") //为了不删除表结构,先truncate 再Append
12  61         conn.close()
13  62       }
14  63       catch {
15  64         case e: Exception =>
16  65           println("MySQL Error:")
17  66           e.printStackTrace()
18  67       }

 

truncate仅仅是删除数据,并不删除结构。

如果表一开始不存在

如果一开始不存在需要分两种情况:

1.非SaveMode.Overwrite模式

没有问题,会直接建表,用默认的数据类型

2.SaveMode.Overwrite模式

会报错,下面是在没有TestMysqlTble2的情况下使用SaveMode.Overwrite

 1 com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Table 'iptv.TESTMYSQLTBLE2' doesn't exist
 2         at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
 3         at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
 4         at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
 5         at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
 6         at com.mysql.jdbc.Util.handleNewInstance(Util.java:404)
 7         at com.mysql.jdbc.Util.getInstance(Util.java:387)
 8         at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:939)
 9         at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3878)
10         at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3814)
11         at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2478)
12         at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2625)
13         at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2547)
14         at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2505)
15         at com.mysql.jdbc.StatementImpl.executeInternal(StatementImpl.java:840)
16         at com.mysql.jdbc.StatementImpl.execute(StatementImpl.java:740)
17         at com.iptv.job.basedata.SaveDataFrameASMysql$.saveASMysqlTable(SaveDataFrameASMysql.scala:62)
18         at com.iptv.job.basedata.SaveDataFrameASMysql$.main(SaveDataFrameASMysql.scala:33)
19         at com.iptv.job.basedata.SaveDataFrameASMysql.main(SaveDataFrameASMysql.scala)
20         at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
21         at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
22         at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
23         at java.lang.reflect.Method.invoke(Method.java:498)
24         at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731)
25         at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
26         at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
27         at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
28         at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

报错详情

1 at com.iptv.job.basedata.SaveDataFrameASMysql$.saveASMysqlTable(SaveDataFrameASMysql.scala:62)
2 生面报错位置对应的代码为
3 stmt.execute(s"truncate table $table") //为了不删除表结构,先truncate 再Append
4 即truncate需要表存在

 

 

至此,DataFrame写mysql功能实现

 

文章为个人工作总结,转载请注明出处!!!!!!!

 

    原文作者:spark
    原文地址: https://www.cnblogs.com/lillcol/p/9796935.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞