免费观看又色又爽又黄的小说免费_美女福利视频国产片_亚洲欧美精品_美国一级大黄大色毛片

sparkrdd轉(zhuǎn)dataframe寫(xiě)入mysql的實(shí)例講解

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

創(chuàng)新互聯(lián)致力于互聯(lián)網(wǎng)品牌建設(shè)與網(wǎng)絡(luò)營(yíng)銷，包括成都網(wǎng)站制作、網(wǎng)站建設(shè)、SEO優(yōu)化、網(wǎng)絡(luò)推廣、整站優(yōu)化營(yíng)銷策劃推廣、電子商務(wù)、移動(dòng)互聯(lián)網(wǎng)營(yíng)銷等。創(chuàng)新互聯(lián)為不同類型的客戶提供良好的互聯(lián)網(wǎng)應(yīng)用定制及解決方案，創(chuàng)新互聯(lián)核心團(tuán)隊(duì)10多年專注互聯(lián)網(wǎng)開(kāi)發(fā)，積累了豐富的網(wǎng)站經(jīng)驗(yàn)，為廣大企業(yè)客戶提供一站式企業(yè)網(wǎng)站建設(shè)服務(wù)，在網(wǎng)站建設(shè)行業(yè)內(nèi)樹(shù)立了良好口碑。

下面的示例為讀取本地文件成rdd并隱式轉(zhuǎn)換成dataframe對(duì)數(shù)據(jù)進(jìn)行查詢，最后以追加的形式寫(xiě)入MySQL表的過(guò)程,scala代碼示例如下

import java.sql.Timestamp
import 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]")
// ----------------------
 //參數(shù) spark.sql.autoBroadcastJoinThreshold 設(shè)置某個(gè)表是否應(yīng)該做broadcast，默認(rèn)10M，設(shè)置為-1表示禁用
 //spark.sql.codegen 是否預(yù)編譯sql成java字節(jié)碼，長(zhǎng)時(shí)間或頻繁的sql有優(yōu)化效果
 // spark.sql.inMemoryColumnarStorage.batchSize 一次處理的row數(shù)量，小心oom
 //spark.sql.inMemoryColumnarStorage.compressed 設(shè)置內(nèi)存中的列存儲(chǔ)是否需要壓縮
// ----------------------
 conf.set("spark.sql.shuffle.partitions","20") //默認(rèn)partition是200個(gè)
 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) { //做一個(gè)簡(jiǎn)單的過(guò)濾
  ac.add(1)
  false
  } else true)
  .map(line => memberbase(line(0).toLong, line(1),Timestamp.valueOf(line(2)), line(3).toInt))
 // 方法一、利用隱式轉(zhuǎn)換
 import sqc.implicits._
 val dftemp = log.toDF() // 轉(zhuǎn)換
 /*
  方法二、利用createDataFrame方法,內(nèi)部利用反射獲取字段及其類型
  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")
 // 調(diào)用DataFrameWriter將數(shù)據(jù)寫(xiě)入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中的示例數(shù)據(jù)如下，數(shù)據(jù)來(lái)自hive，字段信息分別為 分區(qū)號(hào)、用戶id、注冊(cè)時(shí)間、第三方號(hào)

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

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

spark rdd轉(zhuǎn)dataframe 寫(xiě)入mysql的實(shí)例講解

更多明細(xì)可以查看官方文檔 Spark SQL and DataFrame Guide

以上這篇spark rdd轉(zhuǎn)dataframe 寫(xiě)入mysql的實(shí)例講解就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持創(chuàng)新互聯(lián)。

網(wǎng)頁(yè)標(biāo)題：sparkrdd轉(zhuǎn)dataframe寫(xiě)入mysql的實(shí)例講解
本文網(wǎng)址：http://m.newbst.com/article0/iigcoo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供定制開(kāi)發(fā)、商城網(wǎng)站、微信公眾號(hào)、建站公司、域名注冊(cè)、自適應(yīng)網(wǎng)站

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)

成都網(wǎng)頁(yè)設(shè)計(jì)公司