RDD，全稱Resilient Distributed
Datasets（彈性分散式資料集），是Spark最為核心的概念，是Spark對資料的抽象。RDD是分散式的元素集合，每個RDD只支援讀操作，且每個RDD都被分為多個分割槽儲存到叢集的不同節點上。除此之外，RDD還允許使用者顯示的指定資料儲存到記憶體和磁碟中，掌握了RDD程式設計是SPARK開發的第一步。

1.RDD讀取資料

檔案讀取

scala> val lines = sc.textFile("README.md")
scala> lines.collect()

並行化讀取

scala> var
 
 lines = sc.parallelize(List("i love you"))
scala> lines.collect()

2.RDD操作之transformation

RDD的操作可分兩種：
1 轉化操作(transformation) : 由一個RDD生成一個新的RDD
2 行動操作(action) : 對RDD中的元素進行計算，並把結果返回
參考api: org.apache.spark.rdd.RDD
3.下劃線用法

//以下兩句等同
_.split(" ")
line => line.split(" ")
//等同
_ + _
(left, right
 
) => left + right

2.1.map()

返回：一個新的RDD

scala> val rdd = sc.parallelize(List(1,2,3,3))
scala> val rrr=rdd.map(x => x +1 )
scala> rrr.collect()
        Array[Int] = Array(2, 3, 4, 4)
scala> rrr.foreach(println)
2
4
4
3

2.2.flatmap()

返回：迭代器的所有元素組成一個新的RDD

val rdd1 = sc.parallelize(List
 
(1,2,3,3))
scala> rdd1.map(x=>x+1).collect
       Array[Int] = Array(2, 3, 4, 4)
scala> rdd1.flatMap(x=>x.to(3)).collect
       Array[Int] = Array(1, 2, 3, 2, 3, 3, 3)

從上面可以看出flatMap先map然後再flat

2.3.filter()

對每個元素進行篩選，返回符合條件的元素組成的一個新RDD

scala> val rdd = sc.parallelize(List(1,2,3,3))
scala> rdd.filter(x => x != 1).collect()
     Array[Int] = Array(2, 3, 3)

兩變數過濾操作

scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))

對value做控制，key不加限制條件

scala> val result = rdd.filter{case(x,y)=>y%3==0}
scala> result.foreach(println)
(3,6)
scala> val result = rdd.filter{case(x,y)=>y<=4}
scala> result.foreach(println)
(1,2)
(3,4)

對key做控制，value不控制

scala> val result = rdd.filter{case(x,y)=>x<3}
scala> result.foreach(println)
(1,2)

2.4 distinct()

distinct() : 去掉重複元素

scala> rdd.distinct().collect()
      Array[Int] = Array(1, 2, 3)

2.5 sample(withReplacement,fration,[seed])

sample():對RDD取樣

• 第一個引數如果為true,可能會有重複的元素，如果為false，不會有重複的元素；
• 第二個引數取值為[0,1]，最後的資料個數大約等於第二個引數乘總數；
• 第三個引數為隨機因子。

 scala> rdd.sample(false,0.5).collect()
     Array[Int] = Array(3, 3)

scala> rdd.sample(false,0.5).collect()
     Array[Int] = Array(1, 2)

scala> rdd.sample(false,0.5,10).collect()
    Array[Int] = Array(2, 3)

2.6排序

rdd.sortByKey()：返回一個根據鍵排序的RDD

資料排序，可以通過接受ascending的引數表示我們是否想要結果按升序排序（預設是true）

scala> val result = rdd.sortByKey().collect()
          Array[(Int, Int)] = Array((1,2), (3,4), (3,6))

2.7reduce

rdd.reduceByKey(func)：合併具有相同key的value值

scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> val result = rdd.reduceByKey((x,y)=>x+y)
scala> result.foreach(println)
(1,2)
(3,10)

2.8group

rdd.groupByKey()：對具有相同鍵的進行分組 [資料分組]

scala> val result = rdd.groupByKey()
scala> result.foreach(println)
       (3,CompactBuffer(4, 6))
       (1,CompactBuffer(2))

2.9 mapValues

rdd.mapValues(func)：對pairRDD中的每個值應用func 鍵不改變

scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> val result = rdd.mapValues(x=>x+1)
scala> result.foreach(println)
(1,3)
(3,5)
(3,7)

3.針對兩個pair RDD 的轉化操作

3.1 rdd.subtractByKey( other )

刪除掉RDD中與other RDD中鍵相同的元素

scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)))
scala> val other = sc.parallelize(List((3,9)))
scala> val result = rdd.subtractByKey(other)
scala> result.foreach(println)
(1,2)

3.2 join內連線

rdd,other是3.1的變數

scala> val result = rdd.join(other)
scala> result.foreach(println)
(3,(4,9))
(3,(6,9))

3.3 右外連線

rdd,other是3.1的變數

scala> val result = rdd.rightOuterJoin(other)
scala> result.foreach(println)
     (3,(Some(4),9))
     (3,(Some(6),9))

3.4 左外連線

scala> val result = rdd.leftOuterJoin(other)
scala> result.foreach(println)
   (3,(4,Some(9)))
   (3,(6,Some(9)))
   (1,(2,None))

3.5按相同鍵分組

scala> val result = rdd.cogroup(other)
scala> result.foreach(println)
(1,(CompactBuffer(2),CompactBuffer()))
(3,(CompactBuffer(4, 6),CompactBuffer(9)))

3.6 使用reduceByKey()和mapValues()計算每個鍵對應累加值

x._1表示取元組的第一個值

scala> val rdd = sc.parallelize(List(Tuple2("panda",0),Tuple2("pink",3),Tuple2("pirate",3),Tuple2("panda",1),Tuple2("pink",4)))
scala> val result = rdd.mapValues(x=>(x,1)).reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))
scala> result.foreach(println)
(pirate,(3,1))
(panda,(1,2))
(pink,(7,2))

3.7計單詞(wordcount)

用map reduce實現：

scala> val rdd = sc.parallelize(List("i am thinkgamer, i love cyan"))
scala> val words = rdd.flatMap(line => line.split(" "))
scala> val result = words.map(x=>(x,1)).reduceByKey((x,y) => x+y)
scala> result.foreach(println)
(cyan,1)
(love,1)
(thinkgamer,,1)
(am,1)
(i,2)

用countByValue實現：

scala> val rdd = sc.parallelize(List("i am thinkgamer, i love cyan"))
scala> val result = rdd.flatMap(x=>x.split(" ")).countByValue()
scala> result.foreach(println)
(am,1)
(thinkgamer,,1)
(i,2)
(love,1)
(cyan,1)

4.RDD操作之action

下面例子以 val rdd = sc.parallelize(List(1,2,3,3))為輸入

4.1 collect()返回所有元素

scala> rdd.collect()
       Array[Int] = Array(1, 2, 3, 3)

4.2 count() ：返回元素個數

scala> rdd.count()
         Long = 4

4.3 countByValue() ： 各個元素出現的次數

scala> rdd.countByValue()
     scala.collection.Map[Int,Long] = Map(1 -> 1, 2 -> 1, 3 -> 2)

4.4 take(num) ： 返回num個元素

scala> rdd.take(2)
       Array[Int] = Array(1, 2)

4.5 top(num) ： 返回前num個元素

scala> rdd.top(2)
         Array[Int] = Array(3, 3)

4.6 foreach(func)：對每個元素使用func

scala> rdd.foreach(x => println(x*2))