SparkSql与Mysql交互, SparkSql如何读取Mysql中的数据, SparkSql如何将数据写入Mysql 26

古城微笑少年丶 2023-06-05 08:16 35阅读 0赞

# 前言 #

SparkSql可以通过**JDBC**从关系型数据库中读取数据,创建**DataFrame**. 然后通过一系列操作后再把数据写回关系型数据库中.

# 1. Spark从Mysql中读取数据 #

## 1.1 通过IDEA中代码加载 ##

*  要设置好**properties**属性,也就是登录Mysql的用户名和密码
 *  通过**spark.read.jdbc**读取数据.

package com.acece.sparkSql
    import java.util.Properties
    import org.apache.spark.sql.{ DataFrame, SparkSession}
    /** * todo:Sparksql从mysql中加载数据 */
    object DataFromMysql { 
      def main(args: Array[String]): Unit = { 
          //todo:1、创建sparkSession对象
          val spark: SparkSession = SparkSession.builder()
            .appName("DataFromMysql")
            .master("local[2]")
            .getOrCreate()
        //todo:2、创建Properties对象，设置连接mysql的用户名和密码
        val properties: Properties =new Properties()
        properties.setProperty("user","root")
        properties.setProperty("password","123456")
        //todo:3、读取mysql中的数据
        val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://192.168.200.100:3306/spark","iplocation",properties)
        //todo:4、显示mysql中表的数据
        mysqlDF.show()
        spark.stop()
      }
    }

*  查看执行结果  
    ![在这里插入图片描述][20191008174943265.png]
    
    ## 1.2 通过Spark-shell读取Mysql中数据 ##
    
     *  启动**Spark-shell**, **必须指定Mysql的驱动包**

spark-shell \
    --master spark://hdp-node-01:7077 \
    --executor-memory 1g \
    --total-executor-cores  2 \
    --jars /opt/bigdata/hive/lib/mysql-connector-java-5.1.35.jar \
    --driver-class-path /opt/bigdata/hive/lib/mysql-connector-java-5.1.35.jar

*  从Mysql中加载数据

val mysqlDF = spark.read \
    				   .format("jdbc") \
    				   .options(Map("url" -> \
    			"jdbc:mysql://192.168.200.150:3306/spark", "driver" -> \
    			"com.mysql.jdbc.Driver", "dbtable" -> "iplocation","user" -> \
    			"root", "password" -> "123456")).load()

*  执行查询  
    ![在这里插入图片描述][20191008175452615.png]

# 2. SparkSql将数据写入到MySQL中 #

**使用代码的方式,通过IDEAb编码**

*  首先要读取数据,比如"\\person.txt"
 *  然后解析数据, 生成DataFrame,最后注册成表
 *  创建**properties**属性, 连接数据库的用户名和密码
 *  通过**spark.write.jdbc**写入到数据库中.

package com.acece.sparkSql
    import java.util.Properties
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{ DataFrame, Dataset, SaveMode, SparkSession}
    /** * todo:sparksql写入数据到mysql中 */
      
    //todo:创建样例类Student
    case class Student(id:Int,name:String,age:Int)
    
    object SparkSqlToMysql { 
      def main(args: Array[String]): Unit = { 
        //todo:1、创建sparkSession对象
          val spark: SparkSession = SparkSession.builder()
            .appName("SparkSqlToMysql")
            .getOrCreate()
        //todo:2、读取数据
          val data: RDD[String] = spark.sparkContext.textFile(args(0))
        //todo:3、切分每一行,
        val arrRDD: RDD[Array[String]] = data.map(_.split(" "))
        //todo:4、RDD关联Student
        val studentRDD: RDD[Student] = arrRDD.map(x=>Student(x(0).toInt,x(1),x(2).toInt))
        //todo:导入隐式转换
        import spark.implicits._
        //todo:5、将RDD转换成DataFrame
        val studentDF: DataFrame = studentRDD.toDF()
        //todo:6、将DataFrame注册成表
        studentDF.createOrReplaceTempView("student")
        //todo:7、操作student表 ,按照年龄进行降序排列
        val resultDF: DataFrame = spark.sql("select * from student order by age desc")
    
        //todo:8、把结果保存在mysql表中
          //todo:创建Properties对象，配置连接mysql的用户名和密码
          val prop =new Properties()
          prop.setProperty("user","root")
          prop.setProperty("password","123456")
    
      resultDF.write.jdbc("jdbc:mysql://192.168.200.150:3306/spark","student",prop)
    
        //todo:写入mysql时，可以配置插入mode，overwrite覆盖，append追加，ignore忽略，error默认表存在报错
        //resultDF.write.mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.200.150:3306/spark","student",prop)
        spark.stop()
      }
    }

*  用Maven打包  
    通过IDEA工具打包即可
 *  将Jar包提交到spark集群

spark-submit \
    --class itcast.sql.SparkSqlToMysql \
    --master spark://hdp-node-01:7077 \
    --executor-memory 1g \
    --total-executor-cores 2 \
    --jars /opt/bigdata/hive/lib/mysql-connector-java-5.1.35.jar  \
    --driver-class-path /opt/bigdata/hive/lib/mysql-connector-java-5.1.35.jar \
    /root/original-spark-2.0.2.jar  /person.txt

*  执行结果如下  
    ![在这里插入图片描述][20191008180132252.png]
 *  查看mysql中的数据  
    ![在这里插入图片描述][2019100818014539.png]

[20191008174943265.png]: /images/20230601/542503783cbd41848bb8eb7fc6fb8ec1.png
[20191008175452615.png]: /images/20230601/7fe729ab60604b328240345d3668c5ae.png
[20191008180132252.png]: /images/20230601/760f20467a4540038ce4a578d788b540.png
[2019100818014539.png]: /images/20230601/31158771e3a9472b97e951be9cc809ec.png