2020寒假学习笔记08

1 create database sparktest;
2 use sparktest;
3 create table employee(id int(4),name char(20),gender char(4),age int(4));
4 insert into employee values(1,‘Alice‘,‘F‘,22);
5 insert into employee values(2,‘John‘,‘M‘,25);

 1 import java.util.Properties
 2 import org.apache.spark.sql.types._
 3 import org.apache.spark.sql.Row
 4 object TestMySQL
 5 {
 6     def main(args:Array[String])
 7     {
 8         val employeeRDD=spark.sparkContext.parallelize(Array("3 Mary F 26","4 Tom M 23")).map(_.split(" "))
 9         val schema=StructType(List(StructField("id",IntegerType,true),StructField("name",StringType,true),StructField("gender",StringType,true),StructField("age",IntegerType,true)))
10         val rowRDD=employeeRDD.map(p=>Row(p(0).toInt,p(1).trim,p(2).trim,p(3).toInt))
11         val employeeDF=spark.createDataFrame(rowRDD,schema)
12         val prop=new Properties()
13         prop.put("user","root")
14         prop.put("password","yr123456")
15         prop.put("driver","com.mysql.jdbc.Driver")
16         employeeDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest","sparktest.employee",prop)
17         val jdbcDF=spark.read.format("jdbc").option("driver","com.mysql.jdbc.Driver").option("dbtable","employee").option("user","root").option("password","yr123456").load()
18         jdbcDF.agg("age"->"max","age"->"sum")
19     }
20 }