本文介绍Spark如何访问MySQL。

Spark RDD访问MySQL

示例代码如下。
val input = getSparkContext.textFile(inputPath, numPartitions)
    input.flatMap(_.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)
      .mapPartitions(e => {
        var conn: Connection = null
        var ps: PreparedStatement = null
        val sql = s"insert into $tbName(word, count) values (?, ?)"
        try {
          conn = DriverManager.getConnection(s"jdbc:mysql://$dbUrl:$dbPort/$dbName", dbUser, dbPwd)
          ps = conn.prepareStatement(sql)
          e.foreach(pair => {
            ps.setString(1, pair._1)
            ps.setLong(2, pair._2)
            ps.executeUpdate()
          })

          ps.close()
          conn.close()
        } catch {
          case e: Exception => e.printStackTrace()
        } finally {
          if (ps != null) {
            ps.close()
          }
          if (conn != null) {
            conn.close()
          }
        }
      Iterator.empty
    }).count()

spark-sql访问MySQL

访问命令如下。
spark-sql --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/*,mysql-connector-java-8.0.30.jar 
说明
  • mysql-connector-java-8.0.30.jar包含了MySQL JDBC Driver,请根据实际地址填写MySQL Driver的版本以及路径。
  • /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/*中包含JDBC2 DataSource类型。如果您EMR集群使用的是Spark2,则应修改上面命令中的spark3应该换成spark2
建表和读取数据示例如下。
//建表
create table test1(id int)
using jdbc2
options(
  url="jdbc:mysql://mysql_url/test_db?user=root&password=root",
  dbtable="test1",
  driver="com.mysql.jdbc.Driver");

//读取MySQL
select * from test1;

//写入MySQL
insert into test1 values(1);

相关文档