[大数据之Spark]——Transformations转换入门经典实例

Spark相比于Mapreduce的一大优势就是提供了很多的方法，可以直接使用；另一个优势就是执行速度快，这要得益于DAG的调度，想要理解这个调度规则，还要理解函数之间的依赖关系。

本篇就着重描述下Spark提供的Transformations方法.

依赖关系

宽依赖和窄依赖

窄依赖(narrow dependencies)

窄依赖是指父RDD仅仅被一个子RDD所使用，子RDD的每个分区依赖于常数个父分区（O(1),与数据规模无关）。

输入输出一对一的算子，且结果RDD的分区结构不变。主要是map/flatmap

输入输出一对一的算子，但结果RDD的分区结构发生了变化，如union/coalesce

从输入中选择部分元素的算子，如filter、distinct、substract、sample

宽依赖(wide dependencies)

宽依赖是指父RDD被多个子分区使用，子RDD的每个分区依赖于所有的父RDD分区（O(n),与数据规模有关）

对单个RDD基于key进行重组和reduce，如groupByKey，reduceByKey

对两个RDD基于key进行join和重组，如join(父RDD不是hash-partitioned )

需要进行分区，如partitionBy

Transformations转换方法实例

map(func)

map用于遍历rdd中的每个元素，可以针对每个元素做操作处理:

scala> var data = sc.parallelize(1 to 9,3)//内容为 Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)scala> data.map(x=>x*2).collect()//输出内容 Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)

filter(func)

filter用于过滤元素信息，仅仅返回满足过滤条件的元素

scala> var data = sc.parallelize(1 to 9,3)//内容为 Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)scala> data.filter(x=> x%2==0).collect()//输出内容 Array[Int] = Array(2, 4, 6, 8)

flatMap(func)

flatMap与map相比，不同的是可以输出多个结果，比如

scala> var data = sc.parallelize(1 to 4,1)//输出内容为 Array[Int] = Array(1, 2, 3, 4)scala> data.flatMap(x=> 1 to x).collect()//输出内容为 Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4)

mapPartitions(func)

mapPartitions与map类似，只不过每个元素都是一个分区的迭代器，因此内部可以针对分区为单位进行处理。

比如，针对每个分区做和

//首先创建三个分区scala> var data = sc.parallelize(1 to 9,3)//输出为 Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)//查看分区的个数scala> data.partitions.size//输出为 Int = 3//使用mapPartitionsscala> var result = data.mapPartitions{ x=> { | var res = List[Int]() | var i = 0 | while(x.hasNext){ | i+=x.next() | } | res.::(i).iterator | }}scala> result.collect//输出为 Array[Int] = Array(6, 15, 24)

mapPartitionsWithIndex(func)

这个方法与上面的mapPartitions相同，只不过多提供了一个Index参数。

//首先创建三个分区scala> var data = sc.parallelize(1 to 9,3)//输出为 Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)//查看分区的个数scala> data.partitions.size//输出为 Int = 3scala> var result = data.mapPartitionsWithIndex{ | (x,iter) => { | var result = List[String]() | var i = 0 | while(iter.hasNext){ | i += iter.next() | } | result.::( x + "|" +i).iterator | }} result.collect//输出结果为 Array[String] = Array(0|6, 1|15, 2|24)

sample(withReplacement, fraction, seed)

这个方法可以用于对数据进行采样，比如从1000个数据里面随机5个数据。

第一个参数withReplacement代表是否进行替换，如果选true，上面的例子中，会出现重复的数据

第二个参数fraction 表示随机的比例

第三个参数seed 表示随机的种子

//创建数据var data = sc.parallelize(1 to 1000,1)//采用固定的种子seed随机data.sample(false,0.005,0).collect//输出为 Array[Int] = Array(53, 423, 433, 523, 956, 990)//采用随机种子data.sample(false,0.005,scala.util.Random.nextInt(1000)).collect//输出为 Array[Int] = Array(136, 158)

union(otherDataset)

union方法可以合并两个数据集，但是不会去重，仅仅合并而已。

//创建第一个数据集scala> var data1 = sc.parallelize(1 to 5,1)//创建第二个数据集scala> var data2 = sc.parallelize(3 to 7,1)//取并集scala> data1.union(data2).collect//输出为 Array[Int] = Array(1, 2, 3, 4, 5, 3, 4, 5, 6, 7)

intersection(otherDataset)

这个方法用于取两个数据集的交集

//创建第一个数据集scala> var data1 = sc.parallelize(1 to 5,1)//创建第二个数据集scala> var data2 = sc.parallelize(3 to 7,1)//取交集scala> data1.intersection(data2).collect//输出为 Array[Int] = Array(4, 3, 5)

distinct([numTasks]))

这个方法用于对本身的数据集进行去重处理。

//创建数据集scala> var data = sc.parallelize(List(1,1,1,2,2,3,4),1)//执行去重scala> data.distinct.collect//输出为 Array[Int] = Array(4, 1, 3, 2)//如果是键值对的数据，kv都相同，才算是相同的元素scala> var data = sc.parallelize(List(("A",1),("A",1),("A",2),("B",1)))//执行去重scala> data.distinct.collect//输出为 Array[(String, Int)] = Array((A,1), (B,1), (A,2))

groupByKey([numTasks])

这个方法属于宽依赖的方法，针对所有的kv进行分组，可以把相同的k的聚合起来。如果要想计算sum等操作，最好使用reduceByKey或者combineByKey

//创建数据集scala> var data = sc.parallelize(List(("A",1),("A",1),("A",2),("B",1)))//分组输出scala> data.groupByKey.collect//输出为 Array[(String, Iterable[Int])] = Array((B,CompactBuffer(1)), (A,CompactBuffer(1, 1, 2)))

reduceByKey(func, [numTasks])

这个方法用于根据key作分组计算，但是它跟reduce不同，它还是属于transfomation的方法。

//创建数据集scala> var data = sc.parallelize(List(("A",1),("A",1),("A",2),("B",1)))scala> data.reduceByKey((x,y) => x+y).collect//输出为 Array[(String, Int)] = Array((B,1), (A,4))

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

aggregateByKey比较复杂，我也不是很熟练，不过试验了下，大概的意思是针对分区内部使用seqOp方法，针对最后的结果使用combOp方法。

比如，想要统计分区内的最大值，然后再全部统计加和：

scala> var data = sc.parallelize(List((1,1),(1,2),(1,3),(2,4)),2)data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[54] at parallelize at :27scala> def sum(a:Int,b:Int):Int = { a+b }sum: (a: Int, b: Int)Intscala> data.aggregateByKey(0)(sum,sum).collectres42: Array[(Int, Int)] = Array((2,4), (1,6))scala> def max(a:Int,b:Int):Int = { math.max(a,b) }max: (a: Int, b: Int)Intscala> data.aggregateByKey(0)(max,sum).collectres44: Array[(Int, Int)] = Array((2,4), (1,5))

sortByKey([ascending], [numTasks])

sortByKey用于针对Key做排序，默认是按照升序排序。

//创建数据集scala> var data = sc.parallelize(List(("A",2),("B",2),("A",1),("B",1),("C",1)))data: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[30] at parallelize at :27//对数据集按照key进行默认排序scala> data.sortByKey().collectres23: Array[(String, Int)] = Array((A,2), (A,1), (B,2), (B,1), (C,1))//升序排序scala> data.sortByKey(true).collectres24: Array[(String, Int)] = Array((A,2), (A,1), (B,2), (B,1), (C,1))//降序排序scala> data.sortByKey(false).collectres25: Array[(String, Int)] = Array((C,1), (B,2), (B,1), (A,2), (A,1))

join(otherDataset, [numTasks])

join方法为(K,V)和(K,W)的数据集调用，返回相同的K,所组成的数据集。相当于sql中的按照key做连接。

有点类似于 select a.value,b.value from a inner join b on a.key = b.key;

举个例子

//创建第一个数据集scala> var data1 = sc.parallelize(List(("A",1),("B",2),("C",3)))//创建第二个数据集scala> var data2 = sc.parallelize(List(("A",4)))//创建第三个数据集scala> var data3 = sc.parallelize(List(("A",4),("A",5)))data1.join(data2).collect//输出为 Array[(String, (Int, Int))] = Array((A,(1,4)))data1.join(data3).collect//输出为 Array[(String, (Int, Int))] = Array((A,(1,4)), (A,(1,5)))

cogroup(otherDataset, [numTasks])

在类型为（K,V)和（K,W)的数据集上调用，返回一个 (K, (Seq[V], Seq[W]))元组的数据集。

//创建第一个数据集scala> var data1 = sc.parallelize(List(("A",1),("B",2),("C",3)))//创建第二个数据集scala> var data2 = sc.parallelize(List(("A",4)))//创建第三个数据集scala> var data3 = sc.parallelize(List(("A",4),("A",5)))scala> data1.cogroup(data2).collect//输出为 Array[(String, (Iterable[Int], Iterable[Int]))] = Array((B,(CompactBuffer(2),CompactBuffer())), (A,(CompactBuffer(1),CompactBuffer(4))), (C,(CompactBuffer(3),CompactBuffer())))scala> data1.cogroup(data3).collect//输出为 Array[(String, (Iterable[Int], Iterable[Int]))] = Array((B,(CompactBuffer(2),CompactBuffer())), (A,(CompactBuffer(1),CompactBuffer(4, 5))), (C,(CompactBuffer(3),CompactBuffer())))

cartesian(otherDataset)

这个方法用于计算两个(K,V)数据集之间的笛卡尔积

//创建第一个数据集scala> var a = sc.parallelize(List(1,2))//创建第二个数据集scala> var b = sc.parallelize(List("A","B"))//计算笛卡尔积scala> a.cartesian(b).collect//输出结果 res2: Array[(Int, String)] = Array((1,A), (1,B), (2,A), (2,B))

pipe(command, [envVars])

pipe方法用于针对每个分区的RDD执行一个shell脚本命令，可以使perl或者bash。分区的元素将会被当做输入，脚本的输出则被当做返回的RDD值。

//创建数据集scala> var data = sc.parallelize(1 to 9,3)data: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[34] at parallelize at :27//测试脚本scala> data.pipe("head -n 1").collectres26: Array[String] = Array(1, 4, 7)scala> data.pipe("tail -n 1").collectres27: Array[String] = Array(3, 6, 9)scala> data.pipe("tail -n 2").collectres28: Array[String] = Array(2, 3, 5, 6, 8, 9)

coalesce(numPartitions)

这个方法用于对RDD进行重新分区，第一个参数是分区的数量，第二个参数是是否进行shuffle

//创建数据集scala> var data = sc.parallelize(1 to 9,3)data: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at :27//查看分区的大小scala> data.partitions.sizeres3: Int = 3//不使用shuffle重新分区scala> var result = data.coalesce(2,false)result: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[19] at coalesce at :29scala> result.partitions.lengthres12: Int = 2scala> result.toDebugStringres13: String = (2) CoalescedRDD[19] at coalesce at :29 [] | ParallelCollectionRDD[9] at parallelize at :27 []//使用shuffle重新分区scala> var result = data.coalesce(2,true)result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[23] at coalesce at :29scala> result.partitions.lengthres14: Int = 2scala> result.toDebugStringres15: String = (2) MapPartitionsRDD[23] at coalesce at :29 [] | CoalescedRDD[22] at coalesce at :29 [] | ShuffledRDD[21] at coalesce at :29 [] +-(3) MapPartitionsRDD[20] at coalesce at :29 [] | ParallelCollectionRDD[9] at parallelize at :27 []

repartition(numPartitions)

这个方法作用于coalesce一样，重新对RDD进行分区，相当于shuffle版的calesce

//创建数据集scala> var data = sc.parallelize(1 to 9,3)data: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at :27//查看分区的大小scala> data.partitions.sizeres3: Int = 3scala> var result = data.repartition(2)result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[27] at repartition at :29scala> result.partitions.lengthres16: Int = 2scala> result.toDebugStringres17: String = (2) MapPartitionsRDD[27] at repartition at :29 [] | CoalescedRDD[26] at repartition at :29 [] | ShuffledRDD[25] at repartition at :29 [] +-(3) MapPartitionsRDD[24] at repartition at :29 [] | ParallelCollectionRDD[9] at parallelize at :27 []scala>

repartitionAndSortWithinPartitions(partitioner)

这个方法是在分区中按照key进行排序，这种方式比先分区再sort更高效，因为相当于在shuffle阶段就进行排序。

下面的例子中，由于看不到分区里面的数据。可以通过设置分区个数为1，看到排序的效果。

scala> var data = sc.parallelize(List((1,2),(1,1),(2,3),(2,1),(1,4),(3,5)),2)data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[60] at parallelize at :27scala> data.repartitionAndSortWithinPartitions(new org.apache.spark.HashPartitioner(2)).collectres52: Array[(Int, Int)] = Array((2,3), (2,1), (1,2), (1,1), (1,4), (3,5))scala> data.repartitionAndSortWithinPartitions(new org.apache.spark.HashPartitioner(1)).collectres53: Array[(Int, Int)] = Array((1,2), (1,1), (1,4), (2,3), (2,1), (3,5))scala> data.repartitionAndSortWithinPartitions(new org.apache.spark.HashPartitioner(3)).collectres54: Array[(Int, Int)] = Array((3,5), (1,2), (1,1), (1,4), (2,3), (2,1))

参考

spark 官方文档