大數(shù)據(jù)-6編程實例

Spark編程實主要11Scala2RDD2RDD3RDD主要11Scala2RDD2RDD3Spark程序設計—?JavaJVM 次語?靜態(tài)?與Java結合完Spark程序設計—varx:Int=varx7類型推斷valyhi只讀的?函數(shù)：defsquare(x:Int):Int=defsquare(x:Int)={x*x}defannounce(text:String){}

?Java等價代intx=finalstringy=intsquare(intx){returnx*x;}voidannounce(Stringtext){}Spark程序設計—?泛型vararr=newvallst=List(1,2,arr(5)=7

?Java等價int[]arr=newint[8];List<Integer>lst=arr[5]=用函數(shù)式編程的方式處理集varlist=List(1,2,list.foreach(x=>println(x))//打印list.map(x=>x+2)//=>List(3,4,list.filter(x=>x%2==1)//=>List(1,list.filter(_%2==list.reduce((x,y)=>x+y)// +(xIntx2//閉包標準xx2類型推_2每個元素作用x=>{//x+}defaddTwo(xInt):Int=x+2主要11Scala2RDD2RDD3

RDDexistingcollectioninyourdrivertrringtdatasetsharedanydatasourceofferinga創(chuàng)建conf，封裝了spark配置信valconf=newSparkConf().setAppName(appName)conf.set(“”,“MyFirstProgram”)創(chuàng)建SparkContextvalsc=newSpark程序設計—創(chuàng)建RDDScalaSpark程序設計—創(chuàng)建RDD：本地文件sc.textFile(“file.txt”)//將本地文本文件加載成RDD”)//sequenceFile文件sc.sequenceFile[String,Int](“hdfs://nn:9000/path/file”)?Transformation：將一個RDD通過一種規(guī)則 ?Action：返回結果或者保存結果，只有action//創(chuàng)建

RDD//將RDD傳入函數(shù)，生成新的valsquares=nums.map(x=>x*x)//{1,4,valevensquares.filter(_2==0)//{4}//將一個元 成多個，生成新的nums.flatMap(x=>1tox)//=>{1,1,2,1,2,valpets=sc.parallelize(List((“cat”,1),(“dog”,1),(“cat”,2)))pets.reduceByKey(_+_)//=>{(cat,3),(dog,1)}pets.groupByKey()//=>{(cat,Seq(1,2)),(dog,Seq(1)}pets.sortByKey()//=>{(cat,1),(cat,2),(dog,自動 端進行本 Spark程序設計—RDD//創(chuàng)建新的//將RDD保存為本地集合（返回到driver端nums.collect()//=>Array(1,2,計算元素總合并集合元nums.reduce(_+_)//=>//將RDD寫到HDFS?所有key/valueRDD操作符均包含一個可選參示reducetask并行üwords.reduceByKey(_+_,üvisits.join(pageViews,5)?用戶也可以通過修改spark.default.parallelism認并 的外部變量將被自動傳遞到集üvalpages=üvalquery=?注意?sample():從數(shù)據(jù)集中?union():合并兩個或多個 ?pipe():傳入一個外部Spark編程相關支持，參考 ü類似于MapReduce中的counter，將數(shù)據(jù)從一個節(jié)點發(fā)送到其他各點上去ü通常用 ，調試，記錄符合某類特征的數(shù)據(jù)數(shù)目?Accumulatorimportvaltotal_counter=sc.accumulator(0L,valcounter0=sc.accumulator(0L,"counter0")valcounter1=sc.accumulator(0L,valcount=sc.parallelize(1ton,slices).map{i=>total_counter+=1valx=random*2-1valy=random*2–1if(x*x+y*y<1){counter1+=}elsecounter0+=}if(x*x+y*y<1)1else}.reduce(_+

定義兩個累加累加器加? 高效分發(fā)大對象，比如字典（map），集合（set）等，每個executor一，而不是每個task一份 包括HttpBroadcast和TorrentBroadcast兩?HttpBroadcast與

e

Worker

valdata=List(1,2,3,4,5,6)valbdata=sc.broadcast(data)valrdd=sc.parallelize(1to6,2)

將List 取廣播的valdata=valdata=List(1,2,3,4,5,valrdd=sc.parallelize(1to6,RDDüSpark提供了多種緩存級別，以便于用戶根據(jù)實際需求進行

RDDü對于丟失部分數(shù)據(jù)分區(qū)只需根據(jù)它的lineage就可重新計算出 兩者的區(qū)別是什RDD?SparkRDDü允許將RDD緩存到內存中或磁盤上，以便于重Spark默認的緩存級別是cache和persist/**PersistthisRDDwiththedefaultstoragelevel(`MEMORY_ONLY`).*/defpersist():this.type=persist(StorageLevel.MEMORY_ONLY)/**PersistthisRDDwiththedefaultstoragelevel(`MEMORY_ONLY`).*/defcache():this.type=persist()如果想使用其 級別，可以調用RDD主要11Scala2RDD2RDD3實例1：分布式估算?假設正方形邊長為x?當x->∞時

實例1：分布式估算defmain