IT评测·应用市场-qidao123.com

标题: Spark中RDD的Transformation算子 [打印本页]
作者: 写过一篇    时间: 2023-7-2 08:48
标题: Spark中RDD的Transformation算子
RDD的Transformation算子

map

map算子的功能为做映射，即将原来的RDD中对应的每一个元素，应用外部传入的函数进行运算，返回一个新的RDD
  1. val rdd1: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 2)
  2. val rdd2: RDD[Int] = rdd1.map(_ * 2)
复制代码

flatMap

flatMap算子的功能为扁平化映射，即将原来RDD中对应的每一个元素应用外部的运算逻辑进行运算，然后再将返回的数据进行压平，类似先map，然后再flatten的操作，最后返回一个新的RDD
  1. val arr = Array(
  2.   "spark hive flink",
  3.   "hive hive flink",
  4.   "hive spark flink",
  5.   "hive spark flink"
  6. )
  7. val rdd1: RDD[String] = sc.makeRDD(arr, 2)
  8. val rdd2: RDD[String] = rdd1.flatMap(_.split(" "))
复制代码

filter

filter的功能为过滤，即将原来RDD中对应的每一个元素，应用外部传入的过滤逻辑，然后返回一个新的的RDD
  1. val rdd1: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 2)
  2. val rdd2: RDD[Int] = rdd1.filter(_ % 2 == 0)
复制代码

mapPartitions

将数据以分区为的形式返回进行map操作，一个分区对应一个迭代器，该方法和map方法类似，只不过该方法的参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器，如果在映射的过程中需要频繁创建额外的对象，使用mapPartitions要比map高效的过。
  1. val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5), 2)
  2. var r1: RDD[Int] = rdd1.mapPartitions(it => it.map(x => x * 10))
复制代码
map和mapPartitions的区别，mapPartitions一定会比map效率更高吗？
不一定：如果对RDD中的数据进行简单的映射操作，例如变大写，对数据进行简单的运算，map和mapPartitions的效果是一样的，但是如果是使用到了外部共享的对象或数据库连接，mapPartitions效率会更高一些。
原因：map出入的函数是一条一条的进行处理，如果使用数据库连接，会每来一条数据创建一个连接，导致性能过低，而mapPartitions传入的函数参数是迭代器，是以分区为单位进行操作，可以事先创建好一个连接，反复使用，操作一个分区中的多条数据。
特别提醒：如果使用mapPartitions方法不当，即将迭代器中的数据toList，就是将数据都放到内存中，可能会出现内存溢出的情况。
mapPartitionsWithIndex

类似于mapPartitions, 不过函数要输入两个参数，第一个参数为分区的索引，第二个是对应分区的迭代器。函数的返回的是一个经过该函数转换的迭代器。
  1. val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
  2. val rdd2 = rdd1.mapPartitionsWithIndex((index, it) => {
  3.   it.map(e => s"partition: $index, val: $e")
  4. })
复制代码
keys

RDD中的数据为对偶元组类型，调用keys方法后返回一个新的的RDD，该RDD的对应的数据为原来对偶元组的全部key，该方法有隐式转换
  1. val lst = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. val keyRDD: RDD[String] = wordAndOne.keys
复制代码
values

RDD中的数据为对偶元组类型，调用values方法后返回一个新的的RDD，该RDD的对应的数据为原来对偶元组的全部values
  1. val lst = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. val valueRDD: RDD[Int] = wordAndOne.values
复制代码
mapValues

RDD中的数据为对偶元组类型，将value应用传入的函数进行运算后再与key组合成元组返回一个新的RDD
  1. val lst = List(("spark", 5), ("hive", 3), ("hbase", 4), ("flink", 8))
  2. val rdd1: RDD[(String, Int)] = sc.parallelize(lst, 2)
  3. //将每一个元素的次数乘以10再可跟key组合在一起
  4. //val rdd2 = rdd1.map(t => (t._1, t._2 * 10))
  5. val rdd2 = rdd1.mapValues(_ * 10)
复制代码
flatMapValues

RDD中的数据为对偶元组类型，将value应用传入的函数进行flatMap打平后再与key组合成元组返回一个新的RDD
  1. val lst = List(("spark", "1,2,3"), ("hive", "4,5"), ("hbase", "6"), ("flink", "7,8"))
  2. val rdd1: RDD[(String, String)] = sc.parallelize(lst, 2)
  3. //将value打平，再将打平后的每一个元素与key组合("spark", "1,2,3") =>（"spark",1）,（"spark",2）,（"spark",3）
  4. val rdd2: RDD[(String, Int)] = rdd1.flatMapValues(_.split(",").map(_.toInt))
  5. //    val rdd2 = rdd1.flatMap(t => {
  6. //      t._2.split(",").map(e => (t._1, e.toInt))
  7. //    })
复制代码
uion

将两个类型一样的RDD合并到一起，返回一个新的RDD，新的RDD的分区数量是原来两个RDD的分区数量之和
  1. //两个RDD进行union，对应的数据类型必须一样
  2. //Union不会去重
  3. val rdd1 = sc.parallelize(List(1,2,3,4), 2)
  4. val rdd2 = sc.parallelize(List(5, 6, 7, 8, 9,10), 3)
  5. val rdd3 = rdd1.union(rdd2)
  6. println(rdd3.partitions.length)
复制代码

reduceByKey

将数据按照相同的key进行聚合，特点是先在每个分区中进行局部分组聚合，然后将每个分区聚合的结果从上游拉取到下游再进行全局分组聚合
  1. val lst = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. val reduced: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
复制代码

combineByKey
  1. val lst = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. //调用combineByKey传入三个函数
  10. //val reduced = wordAndOne.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
  11. val f1 = (x: Int) => {
  12.   val stage = TaskContext.get().stageId()
  13.   val partition = TaskContext.getPartitionId()
  14.   println(s"f1 function invoked in state: $stage, partition: $partition")
  15.   x
  16. }
  17. //在每个分区内，将key相同的value进行局部聚合操作
  18. val f2 = (a: Int, b: Int) => {
  19.   val stage = TaskContext.get().stageId()
  20.   val partition = TaskContext.getPartitionId()
  21.   println(s"f2 function invoked in state: $stage, partition: $partition")
  22.   a + b
  23. }
  24. //第三个函数是在下游完成的
  25. val f3 = (m: Int, n: Int) => {
  26.   val stage = TaskContext.get().stageId()
  27.   val partition = TaskContext.getPartitionId()
  28.   println(s"f3 function invoked in state: $stage, partition: $partition")
  29.   m + n
  30. }
  31. val reduced = wordAndOne.combineByKey(f1, f2, f3)
复制代码
combineByKey要传入三个函数：
第一个函数：在上游执行，该key在当前分区第一次出现时，对value处理的运算逻辑
第二个函数：在上游执行，当该key在当前分区再次出现时，将以前相同key的value进行运算的逻辑
第三个函数：在下游执行，将来自不同分区，相同key的数据通过网络拉取过来，然后进行全局聚合的逻辑
groupByKey

按照key进行分组，底层使用的是ShuffledRDD，mapSideCombine = false，传入的三个函数只有前两个被调用了，并且是在下游执行的
  1. val lst = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. //按照key进行分组
  10. val grouped: RDD[(String, Iterable[Int])] = wordAndOne.groupByKey()
复制代码

foldByKey

与reduceByKey类似，只不过是可以指定初始值，每个分区应用一次初始值，先在每个进行局部聚合，然后再全局聚合，局部聚合的逻辑与全局聚合的逻辑相同。
  1. val lst: Seq[(String, Int)] = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  10. //与reduceByKey类似，只不过是可以指定初始值，每个分区应用一次初始值
  11. val reduced: RDD[(String, Int)] = wordAndOne.foldByKey(0)(_ + _)
复制代码
aggregateByKey

与reduceByKey类似，并且可以指定初始值，每个分区应用一次初始值，传入两个函数，分别是局部聚合的计算逻辑、全局聚合的逻辑。
  1. val lst: Seq[(String, Int)] = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. //在第一个括号中传入初始化，第二个括号中传入两个函数，分别是局部聚合的逻辑和全局聚合的逻辑
  10. val reduced: RDD[(String, Int)] = wordAndOne.aggregateByKey(0)(_ + _, _ + _)
复制代码
ShuffledRDD

reduceByKey、combineByKey、aggregateByKey、foldByKey底层都是使用的ShuffledRDD，并且mapSideCombine = true
  1. val f1 = (x: Int) => {
  2.   val stage = TaskContext.get().stageId()
  3.   val partition = TaskContext.getPartitionId()
  4.   println(s"f1 function invoked in state: $stage, partition: $partition")
  5.   x
  6. }
  7. //在每个分区内，将key相同的value进行局部聚合操作
  8. val f2 = (a: Int, b: Int) => {
  9.   val stage = TaskContext.get().stageId()
  10.   val partition = TaskContext.getPartitionId()
  11.   println(s"f2 function invoked in state: $stage, partition: $partition")
  12.   a + b
  13. }
  14. //第三个函数是在下游完成的
  15. val f3 = (m: Int, n: Int) => {
  16.   val stage = TaskContext.get().stageId()
  17.   val partition = TaskContext.getPartitionId()
  18.   println(s"f3 function invoked in state: $stage, partition: $partition")
  19.   m + n
  20. }
  21. //指定分区器为HashPartitioner
  22. val partitioner = new HashPartitioner(wordAndOne.partitions.length)
  23. val shuffledRDD = new ShuffledRDD[String, Int, Int](wordAndOne, partitioner)
  24. //设置聚合亲器并关联三个函数
  25. val aggregator = new Aggregator[String, Int, Int](f1, f2, f3)
  26. shuffledRDD.setAggregator(aggregator) //设置聚合器
  27. shuffledRDD.setMapSideCombine(true) //设置map端聚合
复制代码
如果设置了setMapSideCombine(true)，那么聚合器中的三个函数都会执行，前两个在上游执行，第三个在下游执行
如果设置了setMapSideCombine(false)，那么聚合器中的三个函数只会执行前两个，并且这两个函数都是在下游执行
distinct

distinct是对RDD中的元素进行取重，底层使用的是reduceByKey实现的，先局部去重，然后再全局去重
  1. val arr = Array(
  2.   "spark", "hive", "spark", "flink",
  3.   "spark", "hive", "hive", "flink",
  4.   "flink", "flink", "flink", "spark"
  5. )
  6. val rdd1: RDD[String] = sc.parallelize(arr, 3)
  7. //去重
  8. val rdd2: RDD[String] = rdd1.distinct()
  9. distinct的底层实现如下：
  10. Scala
  11. val rdd11: RDD[(String, Null)] = rdd1.map((_, null))
  12. val rdd12: RDD[String] = rdd11.reduceByKey((a, _) => a).keys
复制代码
partitionBy

按照指的的分区器进行分区，底层使用的是ShuffledRDD
  1. val lst: Seq[(String, Int)] = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. val partitioner = new HashPartitioner(wordAndOne.partitions.length)
  10. //按照指定的分区进行分区
  11. val partitioned: RDD[(String, Int)] = wordAndOne.partitionBy(partitioner)
复制代码
repartitionAndSortWithinPartitions

按照值的分区器进行分区，并且将数据按照指的的排序规则在分区内排序，底层使用的是ShuffledRDD，设置了指定的分区器和排序规则
  1. val lst: Seq[(String, Int)] = List(
  2.   ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
  3.   ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
  4.   ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
  5.   ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
  6. )
  7. //通过并行化的方式创建RDD，分区数量为4
  8. val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)
  9. val partitioner = new HashPartitioner(wordAndOne.partitions.length)
  10. //按照指定的分区进行分区，并且将数据按照指定的排序规则在分区内排序
  11. val partitioned = wordAndOne.repartitionAndSortWithinPartitions(partitioner)
  12. repartitionAndSortWithinPartitions的底层实现：
  13. Scala
  14. new ShuffledRDD[K, V, V](self, partitioner).setKeyOrdering(ordering)
复制代码
sortBy
  1. val lines: RDD[String] = sc.textFile("hdfs://node-1.51doit.cn:9000/words")
  2. //切分压平
  3. val words: RDD[String] = lines.flatMap(_.split(" "))
  4. //将单词和1组合
  5. val wordAndOne: RDD[(String, Int)] = words.map((_, 1))
  6. //分组聚合
  7. val reduced: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
  8. //按照单词出现的次数，从高到低进行排序
  9. val sorted: RDD[(String, Int)] = reduced.sortBy(_._2, false)
复制代码
sortByKey

按照指的的key排序规则进行全局排序
  1. val lines: RDD[String] = sc.textFile("hdfs://node-1.51doit.cn:9000/words")
  2. //切分压平
  3. val words: RDD[String] = lines.flatMap(_.split(" "))
  4. //将单词和1组合
  5. val wordAndOne: RDD[(String, Int)] = words.map((_, 1))
  6. //分组聚合
  7. val reduced: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
  8. //按照单词出现的次数，从高到低进行排序
  9. //val sorted: RDD[(String, Int)] = reduced.sortBy(_._2, false)
  10. //val keyed: RDD[(Int, (String, Int))] = reduced.keyBy(_._2).sortByKey()
  11. val sorted = reduced.map(t => (t._2, t)).sortByKey(false)
复制代码
sortBy、sortByKey是Transformation，但是为什么会生成job？
因为sortBy、sortByKey需要实现全局排序，使用的是RangePartitioner，在构建RangePartitioner时，会对数据进行采样，所有会触发Action，根据采样的结果来构建RangePartitioner。
RangePartitioner可以保证数据按照一定的范围全局有序，同时在shuffle的同时，有设置了setKeyOrdering，这样就又可以保证数据在每个分区内有序了！
reparation

reparation的功能是重新分区，一定会shuffle，即将数据打散。reparation的功能是改变分区数量（可以增大、减少、不变）可以将数据相对均匀的重新分区，可以改善数据倾斜的问题
  1. val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3)
  2. //repartition方法一定shuffle
  3. //不论将分区数量变多、变少、或不变，都shuffle
  4. val rdd2 = rdd1.repartition(3)
复制代码

reparation的底层调用的是coalesce，shuffle = true
  1. coalesce(numPartitions, shuffle = true)
复制代码
coalesce

coalesce可以shuffle，也可以不shuffle，如果将分区数量减少，并且shuffle = false，就是将分区进行合并
  1. val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3)
  2. //shuffle = true
  3. val rdd2 = rdd1.coalesce(3, true)
  4. //与repartition(3)功能一样
复制代码
  1. val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 4)
  2. //shuffle = false
  3. val rdd2 = rdd1.coalesce(2, false)
复制代码

cogroup

协同分组，即将多个RDD中对应的数据，使用相同的分区器（HashPartitioner），将来自多个RDD中的key相同的数据通过网络传入到同一台机器的同一个分区中(与groupByKey、groupBy区别是，groupByKey、groupBy只能对一个RDD进行分组)
注意:调用cogroup方法，两个RDD中对应的数据都必须是对偶元组类型，并且key类型一定相同
  1. //通过并行化的方式创建一个RDD
  2. val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2), ("jerry", 4)), 3)
  3. //通过并行化的方式再创建一个RDD
  4. val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2), ("jerry", 4)), 2)
  5. //将两个RDD都进行分组
  6. val grouped: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)
复制代码

join

两个RDD进行join，相当于SQL中的内关联join
两个RDD为什么要进行jion？想要的数据来自于两个数据集，并且两个数据集的数据存在相同的条件，必须关联起来才能得到想要的全部数据
  1. //通过并行化的方式创建一个RDD
  2. val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)), 2)
  3. //通过并行化的方式再创建一个RDD
  4. val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2), ("jerry", 4)), 2)
  5. val rdd3: RDD[(String, (Int, Double))] = rdd1.join(rdd2)
复制代码

leftOuterJoin

左外连接，相当于SQL中的左外关联
  1. //通过并行化的方式创建一个RDD
  2. val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)), 2)
  3. //通过并行化的方式再创建一个RDD
  4. val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2), ("jerry", 4)), 2)
  5. val rdd3: RDD[(String, (Int, Option[Int]))] = rdd1.leftOuterJoin(rdd2)
复制代码

rightOuterJoin

右外连接，相当于SQL中的右外关联
  1. //通过并行化的方式创建一个RDD
  2. val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)), 2)
  3. //通过并行化的方式再创建一个RDD
  4. val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2), ("jerry", 4)), 2)
  5. val rdd3: RDD[(String, (Option[Int], Int))] = rdd1.rightOuterJoin(rdd2)
复制代码

fullOuterJoin

全连接，相当于SQL中的全关联
  1. //通过并行化的方式创建一个RDD
  2. val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)), 2)
  3. //通过并行化的方式再创建一个RDD
  4. val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2), ("jerry", 4)), 2)
  5. val rdd3: RDD[(String, (Option[Int], Option[Int]))] = rdd1.fullOuterJoin(rdd2)
复制代码

intersection

求交集，底层使用的是cogroup实现的
  1. val rdd1 = sc.parallelize(List(1,2,3,4,4,6), 2)
  2. val rdd2 = sc.parallelize(List(3,4,5,6,7,8), 2)
  3. //求交集
  4. val rdd3: RDD[Int] = rdd1.intersection(rdd2)
  6. //使用cogroup实现intersection的功能
  7. val rdd11 = rdd1.map((_, null))
  8. val rdd22 = rdd2.map((_, null))
  9. val rdd33: RDD[(Int, (Iterable[Null], Iterable[Null]))] = rdd11.cogroup(rdd22)
  10. val rdd44: RDD[Int] = rdd33.filter { case (_, (it1, it2)) => it1.nonEmpty && it2.nonEmpty }.keys
复制代码
subtract

求两个RDD的差集，将第一个RDD中的数据，如果在第二个RDD中出现了，就从第一个RDD中移除
  1. val rdd1 = sc.parallelize(List("A", "B", "C", "D", "E"))
  2. val rdd2 = sc.parallelize(List("A", "B"))
  4. val rdd3: RDD[String] = rdd1.subtract(rdd2)
  5. //返回 C D E
复制代码
cartesian

笛卡尔积
  1. val rdd1 = sc.parallelize(List("tom", "jerry"), 2)
  2. val rdd2 = sc.parallelize(List("tom", "kitty", "shuke"), 3)
  3. val rdd3 = rdd1.cartesian(rdd2)
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/) Powered by Discuz! X3.4