Spark算子总结2-Spark-About云-梭伦科技

linux_oracle 发表于 2020-12-4 15:54:07

Spark算子总结2

二. TransformationTransformation 变换/转换算子：这种变换并不触发提交作业，完成作业中间过程处理。Transformation 操作是延迟计算的，也就是说从一个RDD 转换生成另一个 RDD 的转换操作不是马上执行，需要等到有 Action 操作的时候才会真正触发运算。2.1 　RDD基本操作转换2.1.1　map算子将一个RDD中的每个数据项，通过map中的函数映射变为一个新的元素。输入分区与输出分区一对一，即：有多少个输入分区，就有多少个输出分区。
2.1.2　flatmap算子<div style="text-indent: 0em;">1. // 使用flatMap算子</div><div style="text-indent: 0em;">2. scala> var flatmapresult = data.flatMap(line => line.split("\\s+"))</div><div style="text-indent: 0em;">3. flatmapresult: org.apache.spark.rdd.RDD = MapPartitionsRDD at flatMap at :23</div><div style="text-indent: 0em;">4. //运算flagMap算子结果</div><div style="text-indent: 0em;">5. scala> flatmapresult.collect</div><div style="text-indent: 0em;">6. res1: Array = Array(hello, world, hello, spark, hello, hive)</div>使用flatMap时候需要注意：flatMap会将字符串看成是一个字符数组看下面的例子：<span style="font-family: 微软雅黑; color: rgb(255, 192, 0); letter-spacing: 0pt; font-size: 9pt; background-image: initial; background-position: initial; background-size: initial; background-repeat: initial; background-attachment: initial; background-origin: initial; background-clip: initial;">案例区别12. res32: Array = Array(HELLO WORLD, HELLO SPARK, HELLO HIVE, HI SPARK)
3.
4. scala> data.flatMap(_.toUpperCase).collect
5. res33: Array = Array(H, E, L, L, O, , W, O, R, L, D, H, E, L, L, O, , S, P, A, R, K, H, E, L, L, O, , H, I, V, E, H, I, , S, P, A, R, K)

再看：<span style="font-family: 微软雅黑; color: rgb(255, 192, 0); letter-spacing: 0pt; font-size: 9pt; background-image: initial; background-position: initial; background-size: initial; background-repeat: initial; background-attachment: initial; background-origin: initial; background-clip: initial;">案例区别22. res34: Array] = Array(Array(hello, world), Array(hello, spark), Array(hello, hive), Array(hi, spark))
3.
4. scala> data.flatMap(x => x.split("\\s+")).collect
5. res35: Array = Array(hello, world, hello, spark, hello, hive, hi, spark)

总：flatMap只会将String扁平化成字符数组，并不会把Array也扁平化成字符数组，
在案例区别1中，flatmap返回的是String，案例区别2中flatmap返回的是Array

2.1.3　mapPartitions算子
def mapPartitions(f: (Iterator) => Iterator, preservesPartitioning: Boolean = false)(implicit arg0: ClassTag): RDD该函数和map函数类似，只不过映射函数的参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器。如果在映射的过程中需要频繁创建额外的对象，使用mapPartitions要比map高效。比如，将RDD中的所有数据通过JDBC连接写入数据库，如果使用map函数，可能要为每一个元素都创建一个connection，这样开销很大，如果使用mapPartitions，那么只需要针对每一个分区建立一个connection。
<div style="text-indent: 0em;">1. var rdd1 = sc.makeRDD(1 to 5,2)</div><div style="text-indent: 0em;">2. //rdd1有两个分区</div><div style="text-indent: 0em;">3. scala> var rdd3 = rdd1.mapPartitions{ x => {</div><div style="text-indent: 0em;">4. | var result = List()</div><div style="text-indent: 0em;">5. | var i = 0</div><div style="text-indent: 0em;">6. | while(x.hasNext){</div><div style="text-indent: 0em;">7. | i += x.next()</div><div style="text-indent: 0em;">8. | }</div><div style="text-indent: 0em;">9. | result.::(i).iterator</div><div style="text-indent: 0em;">10. | }}</div><div style="text-indent: 0em;">11. rdd3: org.apache.spark.rdd.RDD = MapPartitionsRDD[84] at mapPartitions at :23</div><div style="text-indent: 0em;">12. </div><div style="text-indent: 0em;">13. //rdd3将rdd1中每个分区中的数值累加</div><div style="text-indent: 0em;">14. scala> rdd3.collect</div><div style="text-indent: 0em;">15. res65: Array = Array(3, 12)</div><div style="text-indent: 0em;">16. scala> rdd3.partitions.size</div><div style="text-indent: 0em;">17. res66: Int = 2</div>
2.1.4　distinct算子对RDD中的元素进行去重操作。<div style="text-indent: 0em;">1. scala> data.flatMap(line => line.split("\\s+")).collect</div><div style="text-indent: 0em;">2. res61: Array = Array(hello, world, hello, spark, hello, hive, hi, spark)</div><div style="text-indent: 0em;">3. </div><div style="text-indent: 0em;">4. scala> data.flatMap(line => line.split("\\s+")).distinct.collect</div><div style="text-indent: 0em;">5. res62: Array = Array(hive, hello, world, spark, hi)</div>2.1.5　coalesce算子def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering = null): RDD
该函数用于将RDD进行重分区，使用HashPartitioner。
第一个参数为重分区的数目，第二个为是否进行shuffle，默认为false;<div style="text-indent: 0em;">1. scala> var data = sc.textFile("/tmp/lxw1234/1.txt")</div><div style="text-indent: 0em;">2. data: org.apache.spark.rdd.RDD = MapPartitionsRDD[53] at textFile at :21</div><div style="text-indent: 0em;">3. scala> data.collect</div><div style="text-indent: 0em;">4. res37: Array = Array(hello world, hello spark, hello hive, hi spark)</div><div style="text-indent: 0em;">5. scala> data.partitions.size</div><div style="text-indent: 0em;">6. res38: Int = 2 //RDD data默认有两个分区</div><div style="text-indent: 0em;">7. scala> var rdd1 = data.coalesce(1)</div><div style="text-indent: 0em;">8. rdd1: org.apache.spark.rdd.RDD = CoalescedRDD[2] at coalesce at :23</div><div style="text-indent: 0em;">9. scala> rdd1.partitions.size</div><div style="text-indent: 0em;">10. res1: Int = 1 //rdd1的分区数为1</div><div style="text-indent: 0em;">11. scala> var rdd1 = data.coalesce(4)</div><div style="text-indent: 0em;">12. rdd1: org.apache.spark.rdd.RDD = CoalescedRDD[3] at coalesce at :23</div><div style="text-indent: 0em;">13. scala> rdd1.partitions.size</div><div style="text-indent: 0em;">14. res2: Int = 2 //如果重分区的数目大于原来的分区数，那么必须指定shuffle参数为true，//否则，分区数不便</div><div style="text-indent: 0em;">15. scala> var rdd1 = data.coalesce(4,true)</div><div style="text-indent: 0em;">16. rdd1: org.apache.spark.rdd.RDD = MapPartitionsRDD[7] at coalesce at :23</div><div style="text-indent: 0em;">17. scala> rdd1.partitions.size</div><div style="text-indent: 0em;">18. res3: Int = 4</div>2.1.6　union算子该函数比较简单，就是将两个RDD进行合并，不去重
2.1.7　zip算子def zip(other: RDD)(implicit arg0: ClassTag): RDD[(T, U)]
zip函数用于将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常。
<div style="text-indent: 0em;">1. scala> var rdd1 = sc.makeRDD(1 to 5,2)</div><div style="text-indent: 0em;">2. rdd1: org.apache.spark.rdd.RDD = ParallelCollectionRDD[1] at makeRDD at :21</div><div style="text-indent: 0em;">3. </div><div style="text-indent: 0em;">4. scala> var rdd2 = sc.makeRDD(Seq("A","B","C","D","E"),2)</div><div style="text-indent: 0em;">5. rdd2: org.apache.spark.rdd.RDD = ParallelCollectionRDD[2] at makeRDD at :21</div><div style="text-indent: 0em;">6. </div><div style="text-indent: 0em;">7. scala> rdd1.zip(rdd2).collect</div><div style="text-indent: 0em;">8. res0: Array[(Int, String)] = Array((1,A), (2,B), (3,C), (4,D), (5,E)) </div><div style="text-indent: 0em;">9. </div><div style="text-indent: 0em;">10. scala> rdd2.zip(rdd1).collect</div><div style="text-indent: 0em;">11. res1: Array[(String, Int)] = Array((A,1), (B,2), (C,3), (D,4), (E,5))</div><div style="text-indent: 0em;">12. </div><div style="text-indent: 0em;">13. scala> var rdd3 = sc.makeRDD(Seq("A","B","C","D","E"),3)</div><div style="text-indent: 0em;">14. rdd3: org.apache.spark.rdd.RDD = ParallelCollectionRDD[5] at makeRDD at :21</div><div style="text-indent: 0em;">15. </div><div style="text-indent: 0em;">16. scala> rdd1.zip(rdd3).collect</div><div style="text-indent: 0em;">17. java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions</div><div style="text-indent: 0em;">18. //如果两个RDD分区数不同，则抛出异常</div>

页: [1]

About云-梭伦科技's Archiver

Spark算子总结2