Spark基础之算子

1、join操作

join操作是在特征提取过程中非常常见的一种需求，从多个不同文件完成了特征提取之后，通过join合并为一个完整的特征，可以方便进行接下来的模型训练、预测等其它操作。首先我们准备两份简单的数据，gender和age，其中两张表中的前3条有相同的姓名，后两条不同。

2、union

就是将两个RDD进行合并，不去重。

3、map

map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

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)

4、parallelize

创建出一个可以被并行操作的分布式数据集。

1 2	data = [1, 2, 3, 4, 5] distData = sc.parallelize(data)

一旦分布式数据集（distData）被创建好，它们将可以被并行操作。例如，我们可以调用distData.reduce(lambda a, b: a + b)来将数组的元素相加。

5、reduceByKey

reduceByKey将RDD中所有K,V对中，K值相同的V进行合并，而这个合并，仅仅根据用户传入的函数来进行。参数是Value.

scala> byKey.foreach(println)
(key3,(4,1))
(key1,(1,1))
(key1,(1,1))
(key2,(1,1))
(key2,(4,1))

scala> val reducedByKey = byKey.reduceByKey((x1, x2) =>(x1._1 + x2._1,x1._2 + x2._2))
scala> reducedByKey.collect.foreach(println)
(key1,(2,2))                                                                    
(key2,(5,2))
(key3,(4,1))

6、filter

过滤数据

1	val rdd = sc.parallelize(Seq(("a",1), ("a",2), ("b",2), ("b",3), ("c",1)))

整数类型数据的过滤

rdd.filter(_._2==2)

7、groupBy

groupBy(function) function返回key，传入的RDD的各个元素根据这个key进行分组

val a = sc.parallelize(1 to 9, 3)
a.groupBy(x => { if (x % 2 == 0) "even" else "odd" }).collect//分成两组
Array(
(even,ArrayBuffer(2, 4, 6, 8)),
(odd,ArrayBuffer(1, 3, 5, 7, 9))
)

8、groupByKey( )

val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "spider", "eagle"), 2)
val b = a.keyBy(_.length)//给value加上key，key为对应string的长度
b.groupByKey.collect
//结果 Array((4,ArrayBuffer(lion)), (6,ArrayBuffer(spider)), (3,ArrayBuffer(dog, cat)), (5,ArrayBuffer(tiger, eagle)))

9、keyBy

为各个元素，按指定的函数生成key，形成key-value的RDD。

scala> val a = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
scala> val b = a.keyBy(_.length)
scala> b.collect
res80: Array[(Int, String)] = 
Array((3,dog), (6,salmon), (6,salmon), (3,rat), (8,elephant))

10、lookup

从key-value型的RDD中，筛选出指定的key集合。返回的是Scala的sequence。

val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
val b = a.map(x => (x.length, x))
b.lookup(5)
res0: Seq[String] =
WrappedArray(tiger, eagle)