spark rdd轉dataframe 寫入mysql的實例講解

2024-07-25 19:08:30

字體：大中小

來源：轉載

供稿：網友

dataframe是在spark1.3.0中推出的新的api，這讓spark具備了處理大規模結構化數據的能力，在比原有的RDD轉化方式易用的前提下，據說計算性能更還快了兩倍。spark在離線批處理或者實時計算中都可以將rdd轉成dataframe進而通過簡單的sql命令對數據進行操作，對于熟悉sql的人來說在轉換和過濾過程很方便，甚至可以有更高層次的應用，比如在實時這一塊，傳入kafka的topic名稱和sql語句，后臺讀取自己配置好的內容字段反射成一個class并利用出入的sql對實時數據進行計算，這種情況下不會spark streaming的人也都可以方便的享受到實時計算帶來的好處。　　　

下面的示例為讀取本地文件成rdd并隱式轉換成dataframe對數據進行查詢，最后以追加的形式寫入mysql表的過程,scala代碼示例如下

import java.sql.Timestampimport org.apache.spark.sql.{SaveMode, SQLContext}import org.apache.spark.{SparkContext, SparkConf}object DataFrameSql { case class memberbase(data_date:Long,memberid:String,createtime:Timestamp,sp:Int)extends Serializable{ override def toString: String="%d/t%s/t%s/t%d".format(data_date,memberid,createtime,sp) } def main(args:Array[String]): Unit ={ val conf = new SparkConf() conf.setMaster("local[2]")// ---------------------- //參數 spark.sql.autoBroadcastJoinThreshold 設置某個表是否應該做broadcast，默認10M，設置為-1表示禁用 //spark.sql.codegen 是否預編譯sql成java字節碼，長時間或頻繁的sql有優化效果 // spark.sql.inMemoryColumnarStorage.batchSize 一次處理的row數量，小心oom //spark.sql.inMemoryColumnarStorage.compressed 設置內存中的列存儲是否需要壓縮// ---------------------- conf.set("spark.sql.shuffle.partitions","20") //默認partition是200個 conf.setAppName("dataframe test") val sc = new SparkContext(conf) val sqc = new SQLContext(sc) val ac = sc.accumulator(0,"fail nums") val file = sc.textFile("src//main//resources//000000_0") val log = file.map(lines => lines.split(" ")).filter(line =>  if (line.length != 4) { //做一個簡單的過濾  ac.add(1)  false  } else true)  .map(line => memberbase(line(0).toLong, line(1),Timestamp.valueOf(line(2)), line(3).toInt)) // 方法一、利用隱式轉換 import sqc.implicits._ val dftemp = log.toDF() // 轉換 /*  方法二、利用createDataFrame方法,內部利用反射獲取字段及其類型  val dftemp = sqc.createDataFrame(log)  */ val df = dftemp.registerTempTable("memberbaseinfo") /*val sqlcommand ="select date_format(createtime,'yyyy-MM')as mm,count(1) as nums " +  "from memberbaseinfo group by date_format(createtime,'yyyy-MM') " +  "order by nums desc,mm asc "*/ val sqlcommand="select * from memberbaseinfo" val sel = sqc.sql(sqlcommand) val prop = new java.util.Properties prop.setProperty("user","etl") prop.setProperty("password","xxx") // 調用DataFrameWriter將數據寫入mysql val dataResult = sqc.sql(sqlcommand).write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/test","t_spark_dataframe_test",prop) // 表可以不存在 println(ac.name.get+" "+ac.value) sc.stop() }}

上面代碼textFile中的示例數據如下，數據來自hive，字段信息分別為分區號、用戶id、注冊時間、第三方號

20160309 45386477 2012-06-12 20:13:15 90143820160309 45390977 2012-06-12 22:38:06 90103620160309 45446677 2012-06-14 21:57:39 90143820160309 45464977 2012-06-15 13:42:55 90143820160309 45572377 2012-06-18 14:55:03 90260620160309 45620577 2012-06-20 00:21:09 90260620160309 45628377 2012-06-20 10:48:05 90118120160309 45628877 2012-06-20 11:10:15 90260620160309 45667777 2012-06-21 18:58:34 90252420160309 45680177 2012-06-22 01:49:55 20160309 45687077 2012-06-22 11:23:22 902607

這里注意字段類型映射，即case class類到dataframe映射，從官網的截圖如下

spark,rdd,dataframe,mysql