Spark SQL入門到實戰之（6）Spark Sql 連接mysql

 

1、基本概念和用法（摘自spark官方文檔中文版）

Spark SQL 還有一個能夠使用 JDBC 從其他數據庫讀取數據的數據源。當使用 JDBC 訪問其它數據庫時，應該首選 JdbcRDD。這是因為結果是以數據框（DataFrame）返回的，且這樣 Spark SQL操作輕松或便於連接其它數據源。因為這種 JDBC 數據源不需要用戶提供 ClassTag，所以它也更適合使用 Java 或 Python 操作。（注意，這與允許其它應用使用 Spark SQL 執行查詢操作的 Spark SQL JDBC 服務器是不同的）。

使用 JDBC 訪問特定數據庫時，需要在 spark classpath 上添加對應的 JDBC 驅動配置。例如，為了從 Spark Shell 連接 postgres，你需要運行如下命令 :

bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar

通過調用數據源API，遠程數據庫的表可以被加載為DataFrame 或Spark SQL臨時表。支持的參數有 :

屬性名	含義
url	要連接的 JDBC URL。
dbtable	要讀取的 JDBC 表。 注意，一個 SQL 查詢的 From 分語句中的任何有效表都能被使用。例如，既可以是完整表名，也可以是括號括起來的子查詢語句。
driver	用於連接 URL 的 JDBC 驅動的類名。
partitionColumn, lowerBound, upperBound, numPartitions	這幾個選項，若有一個被配置，則必須全部配置。它們描述了當從多個 worker 中並行的讀取表時，如何對它分區。partitionColumn 必須時所查詢表的一個數值字段。注意，lowerBound 和 upperBound 都只是用於決定分區跨度的，而不是過濾表中的行。因此，表中的所有行將被分區並返回。
fetchSize	JDBC fetch size，決定每次讀取多少行數據。 默認將它設為較小值（如，Oracle上設為 10）有助於 JDBC 驅動上的性能優化。

2、scala代碼實現連接mysql

2.1 添加mysql 依賴

在sbt 配置文件里添加：

"mysql" % "mysql-connector-java" % "6.0.6"

 

然后執行：

sbt eclipse

 

2.2 建表並初始化數據

DROP TABLE IF EXISTS `USER_T`;  
CREATE TABLE `USER_T` (  
  `ID` INT(11) NOT NULL,  
  `USER_NAME` VARCHAR(40) NOT NULL,  
  PRIMARY KEY (`ID`)  
) ENGINE=INNODB  DEFAULT CHARSET=UTF8;  

INSERT  INTO `USER_T`(`ID`,`USER_NAME`) VALUES (1,'測試1');
INSERT  INTO `USER_T`(`ID`,`USER_NAME`) VALUES (2,'測試2');

image

2.3 代碼

2.3.1 查詢

package com.dkl.leanring.spark.sql

import org.apache.spark.sql.SparkSession

/**
 * spark查詢mysql測試
 */
object MysqlQueryDemo {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("MysqlQueryDemo").master("local").getOrCreate()
    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://192.168.44.128:3306/hive?useUnicode=true&characterEncoding=utf-8")
      .option("dbtable", "USER_T")
      .option("user", "root")
      .option("password", "Root-123456")
      .load()
    jdbcDF.show()
  }
}

 

image

2.3.2 插入數據

新建USER_T.csv,造幾條數據如圖：
（需將csv的編碼格式轉為utf-8,否則spark讀取中文亂碼，轉碼方法見：https://jingyan.baidu.com/article/fea4511a092e53f7bb912528.html）

 

image

 

package com.dkl.leanring.spark.sql

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode
import java.util.Properties

/**

 * 從USER_T.csv讀取數據並插入的mysql表中
 */
object MysqlInsertDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("MysqlInsertDemo").master("local").getOrCreate()
    val df = spark.read.option("header", "true").csv("src/main/resources/scala/USER_T.csv")
    df.show()
    val url = "jdbc:mysql://192.168.44.128:3306/hive?useUnicode=true&characterEncoding=utf-8"
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "Root-123456")
    df.write.mode(SaveMode.Append).jdbc(url, "USER_T", prop)
  }
}

 

image

再查詢一次，就會發現表里多了幾條數據

 

image