通過讀取文件轉(zhuǎn)換成DataFrame數(shù)據(jù)寫入到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}/*** 通過讀取文件轉(zhuǎn)換成DataFrame數(shù)據(jù)寫入到mysql中*/ object SparkSqlToMysql {def main(args: Array[String]): Unit = {//創(chuàng)建sparkSessionval sparkSession: SparkSession = SparkSession.builder().appName("SparkSqlToMysql").master("local").getOrCreate()//讀取數(shù)據(jù)val sc: SparkContext = sparkSession.sparkContextval fileRDD: RDD[String] = sc.textFile("D:\\person.txt")//切分val lineRDD: RDD[Array[String]] = fileRDD.map(_.split(","))//關(guān)聯(lián) 通過StructType指定schema將rdd轉(zhuǎn)換成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)//根據(jù)rdd和schema創(chuàng)建DataFrameval personDF: DataFrame = sparkSession.createDataFrame(rowRDD, schema)//將df注冊成表personDF.createOrReplaceTempView("person")//操作表val resultDF: DataFrame = sparkSession.sql("select * from person order by age desc")//將數(shù)據(jù)存到mysql中//創(chuàng)建properties對象 設(shè)置連接mysql的信息val prop: Properties = new Properties()prop.setProperty("user", "root")prop.setProperty("password", "root")/** mode方法可以指定數(shù)據(jù)插入模式* overwrite:覆蓋，覆蓋表中已經(jīng)存在的數(shù)據(jù)，如果表不存在它會事先幫你創(chuàng)建* append:追加，向表中追加數(shù)據(jù)，如果表不存在它會事先幫你創(chuàng)建* ignore:忽略，表示如果表事先存在，就不進(jìn)行任何操作* error :如果表存在就報錯，它是默認(rèn)選項*/resultDF.write.mode("error").jdbc("jdbc:mysql://192.168.44.31:3306/spark", "person", prop)sparkSession.stop()} }

?

從mysql中讀取數(shù)據(jù)到DataFrame中

package com.zy.sparksqlimport java.util.Propertiesimport org.apache.spark.sql.{DataFrame, SparkSession}/*** 從mysql中讀取數(shù)據(jù)到DataFrame中*/ object DataFromMysql {def main(args: Array[String]): Unit = {//創(chuàng)建sparkSessionval sparkSession: SparkSession = SparkSession.builder().appName("DataFromMysql").master("local").getOrCreate()//創(chuàng)建properties對象 設(shè)置連接mysql的信息val prop: Properties = new Properties()prop.setProperty("user", "root")prop.setProperty("password", "root")//讀取mysql數(shù)據(jù)val mysqlDF: DataFrame = sparkSession.read.jdbc("jdbc:mysql://192.168.44.31:3306/spark", "person", prop)mysqlDF.show()sparkSession.stop()} }

?

轉(zhuǎn)載于:https://www.cnblogs.com/blazeZzz/p/9851154.html

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。