Spark SQL 編程初級實踐2- 編程實現利用 DataFrame 讀寫 MySQL 的數據


編程實現利用 DataFrame 讀寫 MySQL 的數據
(1)在 MySQL 數據庫中新建數據庫 sparktest,再創建表 employee,包含如表 6-2 所示的兩行數據。
表 6-2 employee 表原有數據
(2)配置 Spark 通過 JDBC 連接數據庫 MySQL,編程實現利用 DataFrame 插入如表 6-3 所示的兩行數據到 MySQL 中,最后打印出 age 的最大值和 age 的總和。
表 6-3 employee 表新增數據

 

 

(1)登錄mysql:

 (2)新建數據庫sparktest,新建數據表employee,並輸入數據:

 

 

(3)啟動spark-shell,並指定mysql連接驅動jar包(如果你前面已經采用下面方式啟動了spark-shell,就不需要重復啟動了)

cd /usr/local/spark
./bin/spark-shell \
--jars /usr/local/spark/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar \
--driver-class-path /usr/local/spark/mysql-connector-java-5.1.40-bin.jar

 (4)輸入程序:

import java.util.Properties
import org.apache.spark.sql.{SQLContext, Row}
import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}
val sqlContext = new SQLContext(sc)
//下面我們設置兩條數據表示兩個職工信息
val studentRDD = sc.parallelize(Array("3 Mary F 26","4 Tom M 23")).map(_.split(" "))
//下面要設置模式信息
val schema = StructType(List(StructField("id", IntegerType, true),StructField("name", StringType, true),StructField("gender", StringType, true),StructField("age", StringType, true)))
//下面創建Row對象,每個Row對象都是rowRDD中的一行
val rowRDD = studentRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).trim, p(3).trim))
//建立起Row對象和模式之間的對應關系,也就是把數據和模式對應起來
val studentDataFrame = sqlContext.createDataFrame(rowRDD, schema)
//下面創建一個prop變量用來保存JDBC連接參數
val prop = new Properties()
prop.put("user", "root") //表示用戶名是root
prop.put("password", "123456") //表示密碼是hadoop
prop.put("driver","com.mysql.jdbc.Driver") //表示驅動程序是com.mysql.jdbc.Driver
//下面就可以連接數據庫,采用append模式,表示追加記錄到數據庫sparktest的employee表中
studentDataFrame.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest", "sparktest.employee", prop)

查看更新后的數據表:

 

 

 

 

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM