1次舌、RDD
RDD（Resilient Distributed Dataset彈性分布式數(shù)據(jù)集）是Spark中抽象的數(shù)據(jù)結(jié)構(gòu)類型馒稍，任何數(shù)據(jù)在Spark中都被表示為RDD荆责。從編程的角度來看睹逃，RDD可以簡單看成是一個數(shù)組嬉橙。和普通數(shù)組的區(qū)別是早直，RDD中的數(shù)據(jù)時分區(qū)存儲的，這樣不同分區(qū)的數(shù)據(jù)就可以分布在不同的機器上市框，同時可以被并行處理霞扬。因此，Spark應(yīng)用程序所做的無非是把需要處理的數(shù)據(jù)轉(zhuǎn)換為RDD枫振，然后對RDD進行一系列的變換和操作喻圃，從而得到結(jié)果。
2粪滤、RDD創(chuàng)建
RDD可以從普通數(shù)組創(chuàng)建出來斧拍，也可以從文件系統(tǒng)或者HDFS中的文件創(chuàng)建出來。
1) 從普通數(shù)組創(chuàng)建RDD额衙，里面包含了1到9這9個數(shù)字饮焦，它們分別在3個分區(qū)中
scala> val a = sc.parallelize(1 to 9, 3)
2)讀取文件README.md來創(chuàng)建RDD，文件中的每一行就是RDD中的一個元素
scala> val b = sc.textFile("README.md")
3窍侧、兩類操作算子
主要分兩類县踢，轉(zhuǎn)換（transformation）和動作（action）。兩類函數(shù)的主要區(qū)別是伟件，transformation接受RDD并返回RDD硼啤，而action接受RDD返回非RDD.
transformation操作是延遲計算的，也就是說從一個RDD生成另一個RDD的轉(zhuǎn)換操作不是馬上執(zhí)行斧账，需要等到有action操作的時候才真正觸發(fā)運算谴返。
action算子會觸發(fā)spark提交作業(yè)job煞肾，并將數(shù)據(jù)輸出spark系統(tǒng)。
4嗓袱、轉(zhuǎn)換算子
更多可以參看?鏈接
1）map
對RDD中的每個元素都執(zhí)行一個指定的函數(shù)來產(chǎn)生一個新的RDD籍救。任何原RDD中的元素在新RDD中都有且只有一個元素與之對應(yīng)。
舉例：
scala> val a = sc.parallelize(1 to 9, 3)
scala> val b = a.map(x => x*2)
scala> a.collect
res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
scala> b.collect
res11: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)
2）flatMap
與map類似渠抹，區(qū)別是原RDD中的元素經(jīng)map處理后只能生成一個元素蝙昙，而原RDD中的元素經(jīng)flatmap處理后可生成多個元素來構(gòu)建新RDD。
舉例：對原RDD中的每個元素x產(chǎn)生y個元素（從1到y(tǒng)梧却，y為元素x的值）
scala> val a = sc.parallelize(1 to 4, 2)
scala> val b = a.flatMap(x => 1 to x)
scala> b.collect
res12: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4,1,2,3,4)
3）mapPartitions
mapPartitions是map的一個變種奇颠。map的輸入函數(shù)是應(yīng)用于RDD中每個元素，而mapPartitions的輸入函數(shù)是應(yīng)用于每個分區(qū)放航，也就是把每個分區(qū)中的內(nèi)容作為整體來處理的烈拒。
它的函數(shù)定義為：
def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
f即為輸入函數(shù)，它處理每個分區(qū)里面的內(nèi)容广鳍。每個分區(qū)中的內(nèi)容將以Iterator[T]傳遞給輸入函數(shù)f荆几，f的輸出結(jié)果是Iterator[U]。最終的RDD由所有分區(qū)經(jīng)過輸入函數(shù)處理后的結(jié)果合并起來的赊时。
舉例：
上述例子中的函數(shù)myfunc是把分區(qū)中一個元素和它的下一個元素組成一個Tuple伴郁。因為分區(qū)中最后一個元素沒有下一個元素了，所以(3,4)和(6,7)不在結(jié)果中蛋叼。
mapPartitions還有些變種，比如mapPartitionsWithContext剂陡，它能把處理過程中的一些狀態(tài)信息傳遞給用戶指定的輸入函數(shù)狈涮。還有mapPartitionsWithIndex，它能把分區(qū)的index傳遞給用戶指定的輸入函數(shù)鸭栖。
4）mapWith
是map的另外一個變種歌馍，map只需要一個輸入函數(shù)，而mapWith有兩個輸入函數(shù)晕鹊。它的定義如下：
def mapWith[A: ClassTag, U: ](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]
第一個函數(shù)constructA是把RDD的partition index（index從0開始）作為輸入松却，輸出為新類型A；
第二個函數(shù)f是把二元組(T, A)作為輸入（其中T為原RDD中的元素溅话，A為第一個函數(shù)的輸出）晓锻，輸出類型為U。
舉例：把partition index 乘以10飞几，然后加上2作為新的RDD的元素砚哆。
val x = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3)
x.mapWith(a => a * 10)((a, b) => (b + 2)).collect
res4: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)
5）flatMapWith
flatMapWith與mapWith很類似，都是接收兩個函數(shù)屑墨，一個函數(shù)把partitionIndex作為輸入躁锁，輸出是一個新類型A纷铣；另外一個函數(shù)是以二元組（T,A）作為輸入，輸出為一個序列战转，這些序列里面的元素組成了新的RDD搜立。它的定義如下：
def flatMapWith[A: ClassTag, U: ClassTag](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => Seq[U]): RDD[U]
舉例：
scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 3)
scala> a.flatMapWith(x => x, true)((x, y) => List(y, x)).collect
res58: Array[Int] = Array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2,8, 2, 9)
6）flatMapValues
flatMapValues類似于mapValues，不同的在于flatMapValues應(yīng)用于元素為KV對的RDD中Value槐秧。每個一元素的Value被輸入函數(shù)映射為一系列的值啄踊，然后這些值再與原RDD中的Key組成一系列新的KV對。
舉例
scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> val b = a.flatMapValues(x=>x.to(5))
scala> b.collect
res3: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))
上述例子中原RDD中每個元素的值被轉(zhuǎn)換為一個序列（從其當前值到5）色鸳，比如第一個KV對(1,2), 其值2被轉(zhuǎn)換為2社痛，3，4命雀，5蒜哀。然后其再與原KV對中Key組成一系列新的KV對(1,2),(1,3),(1,4),(1,5)。
7）union
8）cartesian
9）groupBy
10）filter
當需要比較不同類型數(shù)據(jù)時吏砂，參照 ：更多API
11）sample
12）Cache
將RDD元素從磁盤緩存到內(nèi)存
如果數(shù)據(jù)需要復(fù)用撵儿，可以通過Cache算子，將數(shù)據(jù)緩存到內(nèi)存狐血。
13）persist
14）mapValues
顧名思義就是輸入函數(shù)應(yīng)用于RDD中Kev-Value的Value淀歇，原RDD中的Key保持不變，與新的Value一起組成新的RDD中的元素匈织。因此浪默，該函數(shù)只適用于元素為KV對的RDD。
舉例：
scala> val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", " eagle"), 2)
scala> val b = a.map(x => (x.length, x))
scala> b.mapValues("x" + _ + "x").collect
res5: Array[(Int, String)] = Array((3,xdogx), (5,xtigerx), (4,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))
15）combineByKey
16）reduceByKey
顧名思義缀匕，reduceByKey就是對元素為KV對的RDD中Key相同的元素的Value進行reduce纳决，因此，Key相同的多個元素的值被reduce為一個值乡小，然后與原RDD中的Key組成一個新的KV對阔加。
舉例:
scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> a.reduceByKey((x,y) => x + y).collect
res7: Array[(Int, Int)] = Array((1,2), (3,10))
上述例子中，對Key相同的元素的值求和满钟，因此Key為3的兩個元素被轉(zhuǎn)為了(3,10)胜榔。
17）reduce
reduce將RDD中元素兩兩傳遞給輸入函數(shù)，同時產(chǎn)生一個新的值湃番，新產(chǎn)生的值與RDD中下一個元素再被傳遞給輸入函數(shù)直到最后只有一個值為止夭织。
舉例：對RDD中的元素求和
scala> val c = sc.parallelize(1 to 10)
scala> c.reduce((x, y) => x + y)
res4: Int = 55
18）join
19）zip
20）intersection
5活尊、Action算子
1）foreach
2）saveAsTextFile
3）collect
相當于toArray锌钮，將分布式的RDD返回為一個單機的Scala Array.
4）count
