Spark編程實現利用 DataFrame 讀寫 MySQL 的數據

2 ．編程實現將 RDD 轉換為 DataFrame
源文件內容如下（包含 id,name,age）：

請將數據復制保存到 Linux 系統中，命名為 employee.txt，實現從 RDD 轉換得到
DataFrame，並按“id:1,name:Ella,age:36”的格式打印出 DataFrame 的所有數據。請寫出程序代
碼。

3. 編程實現利用 DataFrame 讀寫 MySQL 的數據
（1）在 MySQL 數據庫中新建數據庫 sparktest，再創建表 employee，包含如表 6-2 所示的
兩行數據。

（2）配置 Spark 通過 JDBC 連接數據庫 MySQL，編程實現利用 DataFrame 插入如表 6-3 所
示的兩行數據到 MySQL 中，最后打印出 age 的最大值和 age 的總和。

 

 

mysql -uroot -p

create database sparktest; 
use sparktest;
create table employee(id int(4),name char(50), gender char(20), age int(10)); 
insert into employee values(1,'Alice','F',22);
insert into employee values(2,'John','M',25);
select * from employee;

 

ctrl+z退出

cd /usr/local/spark

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

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)))
val rowRDD = studentRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).trim, p(3).trim))
val studentDataFrame = sqlContext.createDataFrame(rowRDD, schema)
val prop = new Properties()
prop.put("user", "root")
prop.put("password", "123456")
prop.put("driver","com.mysql.jdbc.Driver")
studentDataFrame.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest?characterEncoding=utf-8&useSSL=false", "sparktest.employee", prop)