1）map（func）：返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成.

2）mapPartitions(func)：类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]。假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。

3）reduceByKey（func，[numTask]）：在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。

4）aggregateByKey (zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U: 在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的value传递给combine函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给combine函数，以此类推），将key与计算结果作为一个新的kv对输出。

 5）combineByKey(createCombiner: V=>C, mergeValue: (C, V) =>C, mergeCombiners: (C, C) =>C):

对相同K，把V合并成一个集合。

1.createCombiner: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就和之前的某个元素的键相同。如果这是一个新的元素,combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值

2.mergeValue: 如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并

3.mergeCombiners: 由于每个分区都是独立处理的， 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器， 就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。

spark常用算子有两种：

• transformation：RDD中所有转换算子都是延迟加载，从一个RDD到另一个RDD转换没有立即转换，仅记录数据的逻辑操作，只有要求结果还回到Driver时的动作时才会真正运行。

• action：触发action时才会真正的执行action操作动作
  transformation常用算子类型如下：
  1.textFile (path: String) : RDD[String] 读取hdfs的文本数据，还回String元素的RDD，文本一行即为RDD一行；
  val lineRdd: RDD[String] = sc.textFile(file_path, numPartitions)
  2. mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U],preservesPartitioning: Boolean = false): RDD[U]：该算子与map算子类似，每个分区对RDD先map再聚集在到一起，map元素时可共享分区内资源，当需要额外对象数据时mapPartitions算子比map效率高。
  mapPartitionsWithIndex [U: ClassTag](f: (Int, Iterator[T]) => Iterator[U],preservesPartitioning: Boolean = false): RDD[U] f函数参数包含分区编号和该分区对应的数据集合两个参数，在转换的时候可以把分区index数据加上；参数preservesPartitioning表示是否保留父RDD的partitioner分区信息。

• 3.filterByRange(lower: K, upper: K): RDD[P]：以RDD中元素key的范围做过滤，包含lower和upper上下边界
  val rdd = sc.parallelize(List((2, 21), (9, 2), (5, 3), (6, 3), (3, 21), (10, 21)), 2)
  rdd.filterByRange(3, 9)
  4.flatMapValues[U](f: V => TraversableOnce[U]): RDD[(K, U)]：对元组的value进行业务逻辑操作还回集合，并分别与key进行组合

  
  5.combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C,...): RDD[(K, C)]：该算子是比较底层的算子，groupByKey和reduceByKey是基于此实现；在shuffle前各个partition内先以key做local聚集，会还回每个分区内key对应的C中间值，在shuffle后再合并各个key对应的C。有三个关键的函数，首先分区内独立处理数据，在分区内遍历所有元素把相同key聚合到一起，若key首次出现createCombiner函数把元素V转为C类型还回，若分区内key已经存在mergeValue函数会把对应V与对应的C做合并；分区内处理完成后，若key对应2个以上分区mergeCombiners函数把key对应的各个分区结果C放在一起合并。
  reduceByKey (func: (V, V) => V, numPartitions: Int): RDD[(K, V)] 相同的key对value做聚合，先分区内再整体做聚合，还回与value相同的数据类型；
  foldByKey(zeroValue: V,...)(func: (V, V) => V): RDD[(K, V)]：该算子通过调用combineByKey算子实现，先在各个partition内以key做聚集，分区内首次出现key对应的value通过调用createCombiner对V进行V=>V+zeroValue操作，然后再按key通过func函数对分区内V与余下V进行合并调用mergeValue函数，各个分区的结果也按key聚合通过func函数完成合并调用mergeCombiners函数。

  6.aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) => U,combOp: (U, U) => U): RDD[(K, U)]：算子把相同的key聚集在一起，聚集流程与aggregate算子类似；seqOp函数在分区内先以相同的key做聚集，zeroValue与分区内第一个元素做聚集再依次与剩余元素聚集；在combOp函数中初始值zeroValue不参与聚集，各个分区聚集结果一起聚集还回结果。

  7.coalesce(numPartitions: Int, shuffle: Boolean = false): RDD[T]：对RDD重新分区，默认不shuffle；
  repartition(numPartitions: Int): RDD[T]：重新分区会shuffle
  rdd.partitions.length 分区数量
  9.keyBy[K](f: T => K): RDD[(K, T)]：对RDD元素新加入一个key，旧的元素元素做为value组合成新的二元组元素
  keys: RDD[K]：获取RDD所有的key组合成新的RDD
  values: RDD[V]：获取所有value组合新的RDD

1. val rdd = sc.parallelize(List("abc", "abcd", "ab", "bcd", "bc", "bcde"), 2)

2. // keyBy

3. val rdd2 = rdd.keyBy(_.size)

4. println(rdd2.collect().toBuffer)

5. // keys

6. val keys = rdd2.keys

7. println(keys.collect().toBuffer)

8. // values

9. val values = rdd2.values

10. println(values.collect().toBuffer)

11. /**

12. ArrayBuffer((3,abc), (4,abcd), (2,ab), (3,bcd), (2,bc), (4,bcde))

13. ArrayBuffer(3, 4, 2, 3, 2, 4)

14. ArrayBuffer(abc, abcd, ab, bcd, bc, bcde)

15. */

16. 
    

17. /**

18.    * 分区内初始值zeroValue与每个元素依次聚集

19.    *

20.    * @param zeroValue

21.    * @param value

22.    * @return

23.    */

24. def seqOp(zeroValue: ArrayBuffer[Int], value: Int): ArrayBuffer[Int] = {

25. println("zeroValue:" + zeroValue + "\tvalue:" + value)

26. zeroValue += value

27. }

28. 
    

29. /**

30.    * 各个分区聚集后的结果再依次聚集

31.    *

32.    * @param a

33.    * @param b

34.    * @return

35.    */

36. def combOp(a: ArrayBuffer[Int], b: ArrayBuffer[Int]): ArrayBuffer[Int] = {

37. println("a:" + a + "\tb:" + b)

38. a ++ b

39. }

40. val rdd = sc.parallelize(List(("a", 21), ("b1", 2), ("a", 22), ("b2", 1), ("a", 23), ("c1", 1), ("a", 24),("c2", 1)), 2)

41. val rdd2 = rdd.aggregateByKey(ArrayBuffer[Int](88))(seqOp, combOp)

42. println(rdd2.collect().toBuffer)

43. 
    

44. /**

45. zeroValue:ArrayBuffer(88) value:21

46. zeroValue:ArrayBuffer(88) value:2

47. zeroValue:ArrayBuffer(88, 21) value:22

48. zeroValue:ArrayBuffer(88) value:1

49. zeroValue:ArrayBuffer(88) value:23

50. zeroValue:ArrayBuffer(88) value:1

51. zeroValue:ArrayBuffer(88, 23) value:24

52. zeroValue:ArrayBuffer(88) value:1

53. a:ArrayBuffer(88, 21, 22) b:ArrayBuffer(88, 23, 24)

54. ArrayBuffer((b2,ArrayBuffer(88, 1)), (c1,ArrayBuffer(88, 1)), (a,ArrayBuffer(88, 21, 22, 88, 23, 24)), (c2,ArrayBuffer(88, 1)), (b1,ArrayBuffer(88, 2)))

55. 说明：zeroValue在分区内先按相同key与每个元素依次聚集，各个分区结果再依次聚集zeroValue不参与，故2个分区结果有2个88

56. */

57. /**

58.    * 分区内key首次出现第一个元素转为C

59.    *

60.    * @param value

61.    * @return

62.    */

63. def createCombiner(value: Int): List[Int] = {

64. println("create value:" + value)

65. List(value)

66. }

67. 
    

68. /**

69.    * 分区内key再次出现把V与对应C做合并

70.    *

71.    * @param list

72.    * @param value

73.    * @return

74.    */

75. def mergeValue(list: List[Int], value: Int): List[Int] = {

76. println("merge value:" + value)

77. list :+ (value)

78. }

79. 
    

80. /** *

81.    * 若key对应2+个分区，则合并key对应的各个分区聚集结果C

82.    *

83.    * @param a

84.    * @param b

85.    * @return

86.    */

87. def mergeCombiners(a: List[Int], b: List[Int]): List[Int] = {

88. println("a:" + a.toBuffer + "\tb:" + b.toBuffer)

89. a ++ b

90. }

91. 
    

92. // rdd data

93. val rdd = sc.parallelize(List(("a", 21), ("b1", 2), ("a", 22), ("b2", 1), ("a", 23), ("c1", 1), ("a", 24), ("c2", 1)), 2)

94. // combineByKey

95. val rdd2 = rdd.combineByKey(createCombiner, mergeValue, mergeCombiners)

96. println("combineByKey result:" + rdd2.collect().toBuffer)

97. // reduceByKey

98. val rdd3 = rdd.reduceByKey((pre: Int, after: Int) => (pre + after))

99. println("reduceByKey result:"+rdd3.collect().toBuffer)

100. // reduceByKey

101. val rdd4= rdd.foldByKey(100)(_+_)

102. println("foldByKey result:"+rdd4.collect().toBuffer)

103. /**

104. create value:21

105. create value:1

106. merge value:22

107. create value:1

108. create value:23

109. create value:1

110. merge value:24

111. create value:1

112. a:ArrayBuffer(121, 22) b:ArrayBuffer(123, 24)

113. combineByKey result:ArrayBuffer((b2,List(101)), (c1,List(101)), (a,List(121, 22, 123, 24)), (c2,List(101)), (b1,List(101)))

114. reduceByKey result:ArrayBuffer((b2,1), (c1,1), (a,90), (c2,1), (b1,1))

115. foldByKey result:ArrayBuffer((b2,101), (c1,101), (a,290), (c2,101), (b1,101))

116. */

117. val rdd = sc.parallelize(List((2, "a b c"), (5, "q w e"), (2, "x y z"), (6, "t y")), 2)

118. rdd.flatMapValues(_.split(" ")).collect()

119. /**

120. Array((2,a), (2,b), (2,c), (5,q), (5,w), (5,e), (2,x), (2,y), (2,z), (6,t), (6,y))

121. */

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

2. // mapPartitions算子

3. rdd.mapPartitions {

4. val value = { //map外连接资源 }

5. iterator => iterator.map(_ * value)

6. }

7. // mapPartitionsWithIndex算子

8. val partitionIndex = (index: Int, iter: Iterator[Int]) => {

9. iter.toList.map(item => "index:" + index + ": value: " + item).iterator

10. }

11. rdd.mapPartitionsWithIndex(partitionIndex, true).foreach(println(_))

12. /**

13. index:0: value: 1

14. index:0: value: 2

15. index:1: value: 3

16. index:1: value: 4

17. index:2: value: 5

18. index:2: value: 6

19. */