Spark之 使用SparkSql操作mysql和DataFrame的Scala實現

通過讀取文件轉換成DataFrame數據寫入到mysql中

package com.zy.sparksql

import java.util.Properties

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructType}

/**
  * 通過讀取文件轉換成DataFrame數據寫入到mysql中
  */
object SparkSqlToMysql {
  def main(args: Array[String]): Unit = {
    //創建sparkSession
    val sparkSession: SparkSession = SparkSession.builder().appName("SparkSqlToMysql").master("local").getOrCreate()
    //讀取數據
    val sc: SparkContext = sparkSession.sparkContext
    val fileRDD: RDD[String] = sc.textFile("D:\\person.txt")
    //切分
    val lineRDD: RDD[Array[String]] = fileRDD.map(_.split(","))

    //關聯  通過StructType指定schema將rdd轉換成DataFrame
    val rowRDD: RDD[Row] = lineRDD.map(x => Row(x(0).toInt, x(1), x(2).toInt))
    val schema = (new StructType).add("id", IntegerType, true).add("name", StringType, true).add("age", IntegerType, true)
    //根據rdd和schema創建DataFrame
    val personDF: DataFrame = sparkSession.createDataFrame(rowRDD, schema)

    //將df注冊成表
    personDF.createOrReplaceTempView("person")

    //操作表
    val resultDF: DataFrame = sparkSession.sql("select * from person order by age desc")

    //將數據存到mysql中
    //創建properties對象 設置連接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "root")

    /** mode方法可以指定數據插入模式
      * overwrite:覆蓋，覆蓋表中已經存在的數據，如果表不存在它會事先幫你創建
      * append:追加，向表中追加數據，如果表不存在它會事先幫你創建
      * ignore:忽略，表示如果表事先存在，就不進行任何操作
      * error :如果表存在就報錯，它是默認選項
      */
    resultDF.write.mode("error").jdbc("jdbc:mysql://192.168.44.31:3306/spark", "person", prop)

    sparkSession.stop()
  }
}

 

從mysql中讀取數據到DataFrame中

package com.zy.sparksql

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 從mysql中讀取數據到DataFrame中
  */
object DataFromMysql {
  def main(args: Array[String]): Unit = {
    //創建sparkSession
    val sparkSession: SparkSession = SparkSession.builder().appName("DataFromMysql").master("local").getOrCreate()
    //創建properties對象 設置連接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "root")

    //讀取mysql數據
    val mysqlDF: DataFrame = sparkSession.read.jdbc("jdbc:mysql://192.168.44.31:3306/spark", "person", prop)
    mysqlDF.show()

    sparkSession.stop()
  }
}