初識spark,編程核心rdd算子類別詳解javascala版本實現(xiàn)

Scala語言特點，ScalaJava并發(fā)和分布式特質(zhì)，特征(類似java中interfaces和 5).模式匹配（類似javaswitch）ScalaJavaScalastaticobjectmain方法就是在放object類型中；forScalaif語句、多層嵌yield語法產(chǎn)生新的集合；Scala定義函數(shù)使用defScala返回值類型是函數(shù)后ScalavalvarScala語法上支持很多省略的寫法，如調(diào)用無參數(shù)時，（）可以省略，當調(diào).以及()List(1,2,3).size()寫成Scala不支持break、continueScalatrait、Actorjavascala的關鍵字的差異：implicitrequiressealedsealedScalaActorActorActorModel是消息傳遞模型,基本特征就是消息傳遞importimportclassmyActorextendsdefreceive{casex:String=>println("getString="+casex:Int=>println("getcase_=>println("get}}}}objectLesson_Actordefmain(args:Array[String]):Unit=valactor=newactor!"iloveyou}}casecaseclassclassActor1extendsdefcasemsg:Message=>println("isavamsg!="+msg.actor!"imsg.actor!"iloveyoutoo}casemsg:String=>case_=>println("default}}}}classActor2(actor:Actor)extendsActor{actor!Message(this,"iloveyou!")defcasemsg:String=>if(msg.equals("iloveyoutoo!")){actor!"couldwehaveadate}}case_=>println("default}}}}objectLesson_Actor2defmain(args:Array[String]):Unit={valactor1=newActor1()valactor2=newActor2(actor1)}}SparkMR答：1.Spark可以基于內(nèi)存處理數(shù)據(jù)，MRSparkDAGMRmapreduce，SparkSpark是粗粒度資源申請，MRSparksparkConfSparkRDD,RDDspark解釋Spark1).RDDpartition組成的2).partition上的3).RDD之間有一系列的依賴關系4).k,vRDDSparkaction類算子cache,persist,checkpoint區(qū)別，各自cacheRDDpersist可以指定持久化的級別,最常用的是MEMORY_ONLYMEMORY_AND_DISKcachepersist都是懶執(zhí)行,action類算子來觸發(fā)執(zhí)行cachepersist算子的返回值可以賦值給一個變量,在其他job中直接使用這個變量就是持久化的數(shù)據(jù)了,partitioncachepersistactioncheckpoint將RDD持久化到磁盤,還可以切斷RDD之間的依賴關系 數(shù)據(jù)checkpoint執(zhí)行原理RDDjob執(zhí)行完之后,finalRDD當回溯到某一個RDD調(diào)用了checkpoint方對當前的RDD做一個標spark框架會自動啟動一個新的job,RDD數(shù)據(jù),HDFSSparkStandalone-client,clusterYarn-client,clusterclient模式提交任務后,driverdrivermasterapplication資源申請成功,drivertaskworkerworkertaskdriverdrivertaskworkerworkerdriverresourcemanager收到請求,隨機選擇一臺nodemanager啟動applicationmaster,這里的nodemanagerstandaloneworker節(jié)點applicationmaster啟動后,會向resourcemanager請求一批container資源,用于啟動executor5)resourcemanagernodemanager返回給applicationmaster,executor6)applicationmasternodemanagerexecutor7)executor啟動后,會反向給driver,driver發(fā)送task到executor,執(zhí)行情況和結果返回給driver端rs收到請求后,nodemanageram啟動,amRS,containerRSnodemanagerobjectSpark02defmain(args:Array[String]):Unit=valspark=newsql.SparkSession.Builder().master("local").appName("test").getOrCreate()valsc=spark.sparkContext * valjoinRDD:RDD[(String,(Int,Int))]= * valresult:RDD[(String,(Int,Option[Int]))]= * valresult:RDD[(String,(Option[Int],Int))]= * valresult:RDD[(String,(Option[Int],Option[Int]))]= valnamePartition:Int= valscorePartition:Int= valpartitions:Int= * valresult:RDD[(String,Int)]= valpartitions:Int= * valrdd1:RDD[String]= valrdd2:RDD[String]= valresult:RDD[String]= * valresult:RDD[String]= valresult2:RDD[String]= * valrdd:RDD[String]= valresult:RDD[String]= * valresult:RDD[(String,(Iterable[Int],Iterable[Int]))]= valrdd1:RDD[String]= valrdd2:RDD[Int]= valresult:RDD[(String,Int)]= valresult2:RDD[(String,Long)]= defmain(args:Array[String]):Unit={valconf=newSparkConf()valsc=newvallines:RDD[String]=valwords:RDD[String]=lines.flatMap(line=>{line.split("")})valsampleResult:RDD[String]=words.sample(true,0.1)valsamplePairResult:RDD[(String,Int)]=sampleResult.map(word=>{newTuple2(word,1)}) valsampleSortResult:(String,Int)=sampleReduceResult.sortBy(tp=>{tp._2},false).()valsampleWord:String=sampleSortResult._1valpairResult:RDD[(String,Int)]=filterWords.map(word=>{newTuple2(word,1)})valreduceResult:RD