2020年寒假学习进度第九天

今天主要解决了实验五未解决的问题，前天在做实验五spark连接mysql时出现找不到jar包的问题，看了网上的讲解，调整了一下路径，有人说jar包要放在同名的一个文件夹里，实际上jar包应该直接放在spark安装目录的jars目录下就可，这样就可以连接上。

3. 编程实现利用 DataFrame 读写 MySQL 的数据

(1) 在 MySQL 数据库中新建数据库 sparktest，再建表 employee，包含下列两行数据；

                                                                                                                           表 1 employee 表原有数据

 

(2) 配置Spark通过JDBC连接数据库MySQL，编程实现利用DataFrame插入下列数据到MySQL，

最后打印出 age 的最大值和 age 的总和。

                                                                                                                             表 2 employee 表新增数据

带jar包启动shell编程：

bin/spark-shell--jars/usr/local/spark/jars/mysql-connector-java-5.1.40-bin.jar

 

代码：

import java.util.Properties

 import org.apache.spark.sql.types._

import org.apache.spark.sql.Row

 import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.DataFrameReader

object TestMySQL {

def main(args: Array[String]) {

val spark = SparkSession.builder().appName("RddToDFrame").master("local").getOrCreate()

import spark.implicits._

val employeeRDD = spark.sparkContext.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", IntegerType, true)))

val rowRDD = employeeRDD.map(p => Row(p(0).toInt,p(1).trim,p(2).trim,p(3).toInt)) val employeeDF = spark.createDataFrame(rowRDD, schema)

val prop = new Properties() prop.put("user", "root") prop.put("password", "0") prop.put("driver","com.mysql.jdbc.Driver") employeeDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest","sparktest.employee", prop)

Val jdbcDF = spark.read.format("jdbc").option("url","jdbc:mysql://localhost:3306/sparktest").option("driver","com.mysql.jdbc.Driver").option("dbtable","employee").option("user","root").option("password", "0").load()

jdbcDF.agg("age" -> "max", "age" -> "sum").show()

 print("ok") } }