2023學(xué)年完整公開(kāi)課版RDD編程

RDD編程RDD概念RDD概念DatasourceDatasourceRDDRDDRDDRDDRDDOutput創(chuàng)建創(chuàng)建轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換動(dòng)作RDD特性（1）高效的容錯(cuò)性（2）中間結(jié)果持久化到內(nèi)存，數(shù)據(jù)在內(nèi)存中的多個(gè)RDD操作之間進(jìn)行傳遞，避免了不必要的讀寫(xiě)磁盤(pán)開(kāi)銷（3）存放的數(shù)據(jù)可以是Java對(duì)象，避免了不必要的對(duì)象序列化和反序列化RDD運(yùn)行原理RDD1RDD2窄依賴RDD3窄依賴RDD7RDD5RDD6RDD4RDD運(yùn)行原理寬依賴寬依賴RDD8RDD9RDD運(yùn)行過(guò)程Rdd1.join(rdd2).groupBy().filter(…)RDDobjectsDAGScheduler集群資源管理器TaskSchedulerThreadsBlockmanageWorkerDAG

任務(wù)集任務(wù)PairRDD鍵值對(duì)RDD（PairRDD）是指每個(gè)RDD元素都是（key，value）鍵值對(duì)類型，應(yīng)用場(chǎng)景很多。創(chuàng)建PairRDD（1）從文件系統(tǒng)中加載數(shù)據(jù)生成RDD，然后使用map()函數(shù)轉(zhuǎn)換為PairRDD

scala>vallines=sc.textFile("file:///usr/local/spark/mycode/rdd/word.txt")scala>valpairRDD=lines.flatMap(line=>line.split("")).map(word=>(word,1))scala>pairRDD.foreach(println)(Hadoop,1)(Spark,1)(Hive,1)(Spark,1)創(chuàng)建PairRDD（2）通過(guò)并行集合創(chuàng)建PairRDD

scala>vallist=List("Hadoop","Spark","Hive","Spark")list:List[String]=List(Hadoop,Spark,Hive,Spark)scala>valrdd=sc.parallelize(list)rdd:org.apache.spark.rdd.RDD[String]=ParallelCollectionRDD[11]scala>valpairRDD=rdd.map(word=>(word,1))pairRDD:org.apache.spark.rdd.RDD[(String,Int)]scala>pairRDD.foreach(println)(Hadoop,1)(Spark,1)(Hive,1)(Spark,1)PairRDD轉(zhuǎn)換操作reduceByKey(func)

scala>pairRDD.reduceByKey((a,b)=>a+b).foreach(println)(Spark,2)(Hive,1)(Hadoop,1)PairRDD轉(zhuǎn)換操作groupByKey(func)scala>scala>pairRDD.groupByKey().foreach(println)(Spark,CompactBuffer(1,1))(Hive,CompactBuffer(1))(Hadoop,CompactBuffer(1))PairRDD轉(zhuǎn)換操作pairRDD.keys

scala>pairRDD.keys.foreach(println)HadoopSparkHiveSparkPairRDD轉(zhuǎn)換操作pairRDD.values

scala>pairRDD.values.foreach(println)1111PairRDD轉(zhuǎn)換操作sortByKey()

scala>pairRDD.sortByKey().foreach(println)(Hadoop,1)(Hive,1)(Spark,1)(Spark,1)小例子scala>valrdd=sc.parallelize(Array(("spark",2),("hadoop",6),("hadoop",4),("spark",6)))scala>rdd.mapValues(x=>(x,1)).reduceByKey((x,y)=>(x._1+y._1,x._2+y._2)).mapValues(x=>(x._1/x._2)).collect()res22:Array[(String,Int)]=Array((spark,4),(hadoop,5))給定一組鍵值對(duì)("spark",2),("hadoop",6),("hadoop",4),("spark",6)，鍵值對(duì)的key表示圖書(shū)名稱，value表示某天圖書(shū)銷量，請(qǐng)計(jì)算每個(gè)鍵對(duì)應(yīng)的平均值，也就是計(jì)算每種圖書(shū)的每天平均銷量。RDD操作RDDRDDRDDRDDRDDOutput創(chuàng)建創(chuàng)建轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換動(dòng)作DatasourceDatasourceRDD綜合案例1）

文本文件讀取，示例：18001,Spark,10018001,Python,5018001,Hadoop,3018001,Software,9418002,DataBase,1818002,Python,8218002,Hadoop,7618003,Hadoop,92RDD綜合案例2）選擇RDD或者PairRDD操作（1） 總共有多少學(xué)生；（2）共開(kāi)設(shè)來(lái)多少門課程；（3）Spark課程共有多少人選修；（4）某位同學(xué)的總成績(jī)平均分是多少。RDD綜合案例（1） 總共有多少學(xué)生；vallines=sc.textFile("file:///usr/local/spark/students.txt")18001,Spark,10018001,Python,5018001,Hadoop,3018001,Software,9418002,DataBase,1818002,Python,8218002,Hadoop,7618003,Hadoop,92。。。。。。。RDD綜合案例（1） 總共有多少學(xué)生；vallines=sc.textFile("file:///usr/local/spark/students.txt")valstudents=lines.map(row=>row.split(",")(0))180011800118001180011800218002

。。。。。。。RDD綜合案例（1） 總共有多少學(xué)生；vallines=sc.textFile("file:///usr/local/spark/students.txt")valstudents=lines.map(row=>row.split(",")(0))valstudents_num=students.distinct()//去重操作180011800218003RDD綜合案例（1） 總共有多少學(xué)生；vallines=sc.textFile("file:///usr/local/spark/students.txt")valstudents=lines.map(row=>row.split(",")(0))valstudents_num=students.distinct()//去重操作students_num.count//取得總數(shù)RDD綜合案例（2）共開(kāi)設(shè)來(lái)多少門課程；vallines=sc.textFile("file:///usr/local/spark/students.txt")18001,Spark,10018001,Python,5018001,Hadoop,3018001,Software,9418002,DataBase,1818002,Python,8218002,Hadoop,7618003,Hadoop,92。。。。。。。RDD綜合案例（2）共開(kāi)設(shè)來(lái)多少門課程；vallines=sc.textFile("file:///usr/local/spark/students.txt")valcourses=lines.map(row=>row.split(",")(1))SparkPythonHadoopSoftwarePythonHadoop。。。。。。。RDD綜合案例（2）共開(kāi)設(shè)來(lái)多少門課程；vallines=sc.textFile("file:///usr/local/spark/students.txt")valcourses=lines.map(row=>row.split(",")(1))valcourses_num=num.distinct()//去重操作SparkPythonHadoopSoftwareRDD綜合案例（2）共開(kāi)設(shè)來(lái)多少門課程；vallines=sc.textFile("file:///usr/local/spark/students.txt")valcourses=lines.map(row=>row.split(",")(1))valcourses_num=num.distinct()//去重操作courses_num.count//取得總數(shù)RDD綜合案例（3）Spark課程共有多少人選修；vallines=sc.textFile("file:///usr/local/spark/students.txt")18001,Spark,10018001,Python,5018001,Hadoop,3018001,Software,9418002,DataBase,1818002,Python,8218002,Hadoop,7618003,Hadoop,92。。。。。。。RDD綜合案例（3）Spark課程共有多少人選修；vallines=sc.textFile("file:///usr/local/spark/students.txt")valSpark=lines.fliter(row=>row.split(",")(1)==“Spark”)Spark.countRDD綜合案例（4）某位同學(xué)的總成績(jī)平均分是多少。vallines=sc.textFile("file:///usr/local/spark/students.txt")valrdd1=lines.map(row=>(row.split(",")(0),row.split(",")(2).toInt))(18001,100)(18001,50)(18001,30)(18001,94)。。。。。。。RDD綜合案例（4）某位同學(xué)的總成績(jī)平均分是多少。vallines=sc.textFile("file:///usr/local/spark/students.txt")valrdd1=lines.map(row=>(row.split(",")(0),row.split(",")(2).toInt))valrdd2=rdd1.mapValues(x=>(x,1))(18001,100,1)(18001,50,1)(18001,30,1)(18001,94,1)。。。。。。。RDD綜合案例（4）某位同學(xué)的總成績(jī)平均分是多少。vallines=sc.textFile("file:///usr/local/spark/students.txt")valrdd1=lines.map(row=>(row.split(",")(0),row.split(",")(2).toInt))valrdd2=rdd1.mapValues(x=>(x,1))valres=rdd2.reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))(18001,274,4)(18002,……)RDD綜合案例（4）某位同學(xué)的總成績(jī)平均分是多少。vallines=sc.textFile("file:///usr/local/spark/students.txt")valrdd1=lines.map(row=>(row.split(",")(0),row.split(",")(2).toInt))valrdd2=rdd1.mapValues(x=>(x,1))valres=rdd2.reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))res.mapValues(x=>x._1/x._2).collect()RDD操作——?jiǎng)?chuàng)建RDD（1）從本地文件系統(tǒng)中加載數(shù)據(jù)

（2）從分布式文件系統(tǒng)HDFS中加載數(shù)據(jù)scala>vallines=sc.textFile("file:///usr/local/spark/mycode/rdd/word.txt")scala>vallines=sc.textFile("hdfs://localhost:9000/user/hadoop/word.txt")RDD操作——?jiǎng)?chuàng)建RDD（3）調(diào)用SparkContext的parallelize方法，在Driver中一個(gè)已經(jīng)存在的集合（數(shù)組）上創(chuàng)建。

scala>valarray=Array(1,2,3,4,5)array:Array[Int]=Array(1,2,3,4,5)scala>valrdd=sc.parallelize(array)rdd:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[13]atparallelizeat<console>:29RDD操作——?jiǎng)?chuàng)建RDD調(diào)用SparkContext的parallelize方法，也可以從列表中創(chuàng)建scala>vallist=List(1,2,3,4,5)list:List[Int]=List(1,2,3,4,5)scala>valrdd=sc.parallelize(list)rdd:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[14]atparallelizeat<console>:29RDD操作——轉(zhuǎn)換操作lines.filter(line=>line.contains(“spark”))RDD(lines)rdd.filter()IlikehadoopSparkisfastIlikeSparkSparkisfastIlikeSparkRDDRDD操作——轉(zhuǎn)換操作lines.map(line=>line.split(“”))RDD(lines)rdd.map()IlikehadoopSparkisfastIlikeSparkRDDArray(“I”,”like”,”hadoop”)Array(“Spark”,”is”,”fast”)Array(“I”,”like”,”spark”)RDD操作——轉(zhuǎn)換操作lines.flatmap(line=>line.split(“”))RDD(lines)rdd.flatmap()IlikehadoopSparkisfastIlikeSparkRDD“I””like”“hadoop”“Spark””is”“fast”“I””like”“Spark”RDD操作——轉(zhuǎn)換操作rdd.reduceByKey((a,b)=>a+b)RDD(“Iike”,1)(“hadoop”,1)(“Spark”,1)(“fast”,1)(”like”,1)(“Spark”,1)rdd.reduceByKey()(“Iike”,2)(“hadoop”,1)(“Spark”,2)(“fast”,1)RDD中間結(jié)果key,Value-list例如：(spark，（1,1））RDD操作——轉(zhuǎn)換操作rdd.reduceByKey((a,b)=>a+b)RDD(“Iike”,1)(“hadoop”,1)(“Spark”,1)(“fast”,1)(”like”,1)(“Spark”,1)rdd.reduceByKey()(“Iike”,2)(“hadoop”,1)(“Spark”,2)(“fast”,1)RDDspark，（1,1））RDD操作——?jiǎng)幼鞑僮鱟ount()