[大数据之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

//使用mapPartitions
scala> 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 = 3

scala> 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 <console>:27

scala> def sum(a:Int,b:Int):Int = { a+b }
sum: (a: Int, b: Int)Int

scala> data.aggregateByKey(0)(sum,sum).collect
res42: 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)Int

scala> data.aggregateByKey(0)(max,sum).collect
res44: 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 <console>:27

//对数据集按照key进行默认排序
scala> data.sortByKey().collect
res23: Array[(String, Int)] = Array((A,2), (A,1), (B,2), (B,1), (C,1))

//升序排序
scala> data.sortByKey(true).collect
res24: Array[(String, Int)] = Array((A,2), (A,1), (B,2), (B,1), (C,1))

//降序排序
scala> data.sortByKey(false).collect
res25: 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 <console>:27

//测试脚本
scala> data.pipe("head -n 1").collect
res26: Array[String] = Array(1, 4, 7)

scala> data.pipe("tail -n 1").collect
res27: Array[String] = Array(3, 6, 9)

scala> data.pipe("tail -n 2").collect
res28: 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 <console>:27

//查看分区的大小
scala> data.partitions.size
res3: Int = 3

//不使用shuffle重新分区
scala> var result = data.coalesce(2,false)
result: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[19] at coalesce at <console>:29

scala> result.partitions.length
res12: Int = 2

scala> result.toDebugString
res13: String = 
(2) CoalescedRDD[19] at coalesce at <console>:29 []
 |  ParallelCollectionRDD[9] at parallelize at <console>:27 []

//使用shuffle重新分区
scala> var result = data.coalesce(2,true)
result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[23] at coalesce at <console>:29

scala> result.partitions.length
res14: Int = 2

scala> result.toDebugString
res15: String = 
(2) MapPartitionsRDD[23] at coalesce at <console>:29 []
 |  CoalescedRDD[22] at coalesce at <console>:29 []
 |  ShuffledRDD[21] at coalesce at <console>:29 []
 +-(3) MapPartitionsRDD[20] at coalesce at <console>:29 []
    |  ParallelCollectionRDD[9] at parallelize at <console>: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 <console>:27

//查看分区的大小
scala> data.partitions.size
res3: Int = 3

scala> var result = data.repartition(2)
result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[27] at repartition at <console>:29

scala> result.partitions.length
res16: Int = 2

scala> result.toDebugString
res17: String = 
(2) MapPartitionsRDD[27] at repartition at <console>:29 []
 |  CoalescedRDD[26] at repartition at <console>:29 []
 |  ShuffledRDD[25] at repartition at <console>:29 []
 +-(3) MapPartitionsRDD[24] at repartition at <console>:29 []
    |  ParallelCollectionRDD[9] at parallelize at <console>: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 <console>:27

scala> data.repartitionAndSortWithinPartitions(new org.apache.spark.HashPartitioner(2)).collect
res52: 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)).collect
res53: 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)).collect
res54: Array[(Int, Int)] = Array((3,5), (1,2), (1,1), (1,4), (2,3), (2,1))

参考

spark 官方文档

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏魂祭心

原 ObjectToJson实现

2865
来自专栏静默虚空的博客

Java 序列化

被序列化的类必须属于 Enum、Array 和 Serializable 类型其中的任何一种。

1070
来自专栏积累沉淀

Spark 算子

RDD算子分类,大致可以分为两类,即: Transformation:转换算子,这类转换并不触发提交作业,完成作业中间过程处理。 Action:行动算子,这类算...

2435
来自专栏博客园

Object.assign()

https://developer.mozilla.org/zh-CN/docs/Web/JavaScript/Reference/Global_Objects...

2102
来自专栏xingoo, 一个梦想做发明家的程序员

[大数据之Spark]——Actions算子操作入门实例

Actions reduce(func) Aggregate the elements of the dataset using a function fun...

2356
来自专栏指尖下的Android

二进制的运算

在计算机中存储字节是定长的,即我们8、16、32位等等,6的二进制位为110,但如果在8位计算机中是00000110,高位补零

2273
来自专栏mathor

ObjectIo

 T类的成员中有一个transient修饰的变量,transient意思是透明的,打印时会打印其所属变量类型的默认值,double的默认值是0.0,如果给boo...

1002
来自专栏软件开发

JavaSE学习总结(八)—— 异常处理(Exception)

一、理解异常及异常处理的概念 异常就是在程序的运行过程中所发生的不正常的事件,它会中断正在运行的程序。 异常不是错误 程序中关键的位置有异常处理,提高程序的稳定...

2329
来自专栏CDA数据分析师

CDA数据分析师学习之路第3期 | Spark RDD的转换操作举例

Spark RDD中的操作非常丰富,有80多种针对数据的操作。其中最重要的是Transformation(转换操作)和Action(执行操作)两类。其中转换操作...

21010
来自专栏机器学习入门

LWC 63:748. Shortest Completing Word

LWC 63:748. Shortest Completing Word 传送门:748. Shortest Completing Word Problem: ...

2089

扫码关注云+社区

领取腾讯云代金券