Spark學習筆記總結

01. Spark基礎

1. 介紹

Spark可以用於批處理、互動式查詢（Spark SQL）、實時流處理（Spark Streaming）、機器學習（Spark MLlib）和圖計算（GraphX）。
Spark是MapReduce的替代方案，而且相容HDFS、Hive，可融入Hadoop的生態系統，以彌補MapReduce的不足。

2. Spark-Shell

1. spark-shell是Spark自帶的互動式Shell程式，使用者可以在該命令列下用scala編寫spark程式。
2. 直接啟動spark-shell，實質是spark的local模式，在master:8080中並未顯示客戶端連線。
3. 叢集模式：
   /usr/local/spark/bin/spark-shell \
   --master spark://172.23.27.19:7077 \
   --executor-memory 2g \
   --total-executor-cores 2
4. spark-shell中編寫wordcount
   sc.textFile("hdfs://172.23.27.19:9000/wrd/wc/srcdata/").flatMap(.split(" ")).map((,1)).reduceByKey(+).sortBy(_._2,false).collect

3. RDD介紹與屬性

1. 介紹

RDD（Resilient Distributed Dataset）叫做分散式資料集，是Spark中最基本的資料抽象，它代表一個不可變（建立了內容不可變）、可分割槽、裡面的元素可平行計算的集合。

2. 屬性：

1. 由多個分割槽組成。對於RDD來說，每個分片都會被一個計算任務處理，並決定平行計算的粒度。
2. 一個計算函式用於每個分割槽。Spark中RDD的計算是以分片為單位的，每個RDD都會實現compute函式以達到這個目的。
3. RDD之間的依賴關係。RDD的每次轉換都會生成一個新的RDD，所以RDD之間就會形成類似於流水線一樣的前後依賴關係。資料丟失時，根據依賴重新計算丟失的分割槽而不是整個分割槽。
4. 一個Partitioner，即RDD的分片函式。預設是HashPartition
5. 分割槽資料的最佳位置去計算。就是將計算任務分配到其所要處理資料塊的儲存位置。資料本地化。

3. 建立方式：

1. 可通過並行化scala集合建立RDD
   val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))
2. 通過HDFS支援的檔案系統建立，RDD裡沒有真的資料，只是記錄了元資料
   val rdd2 = sc.textFile("hdfs://172.23.27.19:9000/wrd/wc/srcdata/")

檢視該rdd的分割槽數量
rdd1.partitions.length

3. 基礎的transformation和action

RDD中兩種運算元：
transformation轉換，是延遲載入的

常用的transformation：
（1）map、flatMap、filter
（2）intersection求交集、union求並集：注意型別要一致
distinct:去重
（3）join：型別為(K,V)和(K,W)的RDD上呼叫，返回一個相同key對應的所有元素對在一起的(K,(V,W))的RDD
（4）groupByKey:在一個(K,V)的RDD上呼叫，返回一個(K, Iterator[V])的RDD
但是效率reduceByKey較高，因為有一個本地combiner的過程。
（5）cartesian笛卡爾積

常用的action
（1）collect()、count()
（2）reduce：通過func函式聚集RDD中的所有元素
（3）take(n):取前n個；top(2)：排序取前兩個
（4）takeOrdered(n)，排完序後取前n個

4. 較難的transformation和action

參考《http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html》

（1）mapPartitions(func)和
mapPartitions(func)：
獨立地在RDD的每一個分片上執行，但是返回值；foreachPartition(func)也常用，不需要返回值

mapPartitionsWithIndex(func)：
可以看到分割槽的編號，以及該分割槽資料。
類似於mapPartitions，但func帶有一個整數引數表示分片的索引值，func的函式型別必須是
(Int, Interator[T]) => Iterator[U]

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
val func = (index: Int, iter: Iterator[(Int)]) => {iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator}
rdd1.mapPartitionsWithIndex(func).collect

（2）aggregate
action操作,
第一個引數是初始值,
第二個引數:是2個函式[每個函式都是2個引數(第一個引數:先對個個分割槽進行的操作, 第二個:對個個分割槽合併後的結果再進行合併), 輸出一個引數]

例子：

rdd1.aggregate(0)(_+_, _+_)
//前一個是對每一個分割槽進行的操作，第二個是對各分割槽結果進行的結果

rdd1.aggregate(5)(math.max(_, _), _ + _)
//結果：5 + (5+9) = 19

val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
//結果：24或者42

val rdd4 = sc.parallelize(List("12","23","345",""),2)
rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
//結果01或者10

（3）aggregateByKey
將key值相同的，先區域性操作，再整體操作。。和reduceByKey內部實現差不多

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)
pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect
//結果：Array((dog,12), (cat,17), (mouse,6))

PS：
和reduceByKey(+)呼叫的都是同一個方法，只是aggregateByKey要底層一些，可以先區域性再整體操作。

（4）combineByKey
和reduceByKey是相同的效果，是reduceByKey的底層。
第一個引數x:原封不動取出來, 第二個引數:是函式, 區域性運算, 第三個:是函式, 對區域性運算後的結果再做運算
每個分割槽中每個key中value中的第一個值,

val rdd1 = sc.textFile("hdfs://master:9000/wordcount/input/").flatMap(_.split(" ")).map((_, 1))
val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd2.collect

第一個引數的含義：
每個分割槽中相同的key中value中的第一個值
如：
(hello,1)(hello,1)(good,1)-->(hello(1,1),good(1))-->x就相當於hello的第一個1, good中的1

val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd3.collect
//每個會多加3個10

val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
val rdd6 = rdd5.zip(rdd4)
val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)
//將key相同的資料，放入一個集合中

（5）collectAsMap
Action
Map(b -> 2, a -> 1)//將Array的元祖轉換成Map，以後可以通過key取值

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap
//可以下一步使用

（6）countByKey
根據key計算key的數量
Action

val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
rdd1.countByKey
rdd1.countByValue//將("a", 1)當做一個元素，統計其出現的次數

（7）flatMapValues
對每一個value進行操作後壓平