初识spark,编程核心rdd算子类别详解javascala版本实现

Scala语言特点，ScalaJava并发和分布式特质，特征(类似java中interfaces和 5).模式匹配（类似javaswitch）ScalaJavaScalastaticobjectmain方法就是在放object类型中；forScalaif语句、多层嵌yield语法产生新的集合；Scala定义函数使用defScala返回值类型是函数后ScalavalvarScala语法上支持很多省略的写法，如调用无参数时，（）可以省略，当调.以及()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可以基于内存处理数据，MRSparkDAGMRmapreduce，SparkSpark是粗粒度资源申请，MRSparksparkConfSparkRDD,RDDspark解释Spark1).RDDpartition组成的2).partition上的3).RDD之间有一系列的依赖关系4).k,vRDDSparkaction类算子cache,persist,checkpoint区别，各自cacheRDDpersist可以指定持久化的级别,最常用的是MEMORY_ONLYMEMORY_AND_DISKcachepersist都是懒执行,action类算子来触发执行cachepersist算子的返回值可以赋值给一个变量,在其他job中直接使用这个变量就是持久化的数据了,partitioncachepersistactioncheckpoint将RDD持久化到磁盘,还可以切断RDD之间的依赖关系 数据checkpoint执行原理RDDjob执行完之后,finalRDD当回溯到某一个RDD调用了checkpoint方对当前的RDD做一个标spark框架会自动启动一个新的job,RDD数据,HDFSSparkStandalone-client,clusterYarn-client,clusterclient模式提交任务后,driverdrivermasterapplication资源申请成功,drivertaskworkerworkertaskdriverdrivertaskworkerworkerdriverresourcemanager收到请求,随机选择一台nodemanager启动applicationmaster,这里的nodemanagerstandaloneworker节点applicationmaster启动后,会向resourcemanager请求一批container资源,用于启动executor5)resourcemanagernodemanager返回给applicationmaster,executor6)applicationmasternodemanagerexecutor7)executor启动后,会反向给driver,driver发送task到executor,执行情况和结果返回给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