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

package com.zy.sparksqlimport java.util.Propertiesimport 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 = {//創建sparkSessionval sparkSession: SparkSession = SparkSession.builder().appName("SparkSqlToMysql").master("local").getOrCreate()//讀取數據val sc: SparkContext = sparkSession.sparkContextval fileRDD: RDD[String] = sc.textFile("D:\\person.txt")//切分val lineRDD: RDD[Array[String]] = fileRDD.map(_.split(","))//關聯 通過StructType指定schema將rdd轉換成DataFrameval 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創建DataFrameval 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.sparksqlimport java.util.Propertiesimport org.apache.spark.sql.{DataFrame, SparkSession}/*** 從mysql中讀取數據到DataFrame中*/ object DataFromMysql {def main(args: Array[String]): Unit = {//創建sparkSessionval 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()} }

?

轉載于:https://www.cnblogs.com/blazeZzz/p/9851154.html

總結

以上是生活随笔為你收集整理的Spark之 使用SparkSql操作mysql和DataFrame的Scala实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。