RDD转换算子本编主要基于B站尚硅谷的视频及文档做出的一些改写和添加自己的理解而编写
B站尚硅谷Spark视频讲解链接: 点击跳转.
RDD的行动算子点击跳转
- RDD算子的简绍
- 转换算子
- map算子
- mapPartitions
- map与mapPartition的区别
- mapPartitionsWithIndex
- flatMap
- glom
- groupBy
- filter
- sample
- distinct
- coalesce
- sortBy
- intersection、subtract、union和拉链
- partitionBy
- reduceByKey
- groupByKey
- reduceByKey和groupByKey的区别
- aggregateByKey
- foldByKey
- combineByKey
- sortByKey
- join
- leftOuterJoin和rightOuterJoin
- cogroup
RDD算子可以分为俩类，分别是转换算子和行动算子
- 转换算子：它是功能的补充，将旧的RDD包装成新的RDD
- 行动算子：触发任务的调度和任务的执行
rdd的数据执行顺序
-
rdd的计算一个分区的数据是一个一个执行的
-
只有前面一个数据全部逻辑执行完毕之后，才会执行下一个数据
-
同一分区内执行数据是有序的
-
不同分区数据执行是无序的
如果没有行动算子那么任务就不会执行
转换算子
由于在练习时可能需要多次生成数据，需要使用到SparkContext,为了使代码看起来更加的简洁，我们先生成SparkContext
val spark = new SparkConf().setMaster("local[6]").setAppName("Operator")
val sc = new SparkContext(spark)
map算子
➢ 函数签名：def map[U: ClassTag](f: T => U): RDD[U]
➢ 函数说明：将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。
@Test
def mapTest(): Unit = {
//要求：将[1,2,3,4] --> [2,4,6,8]
val rdd = sc.makeRDD(List(1,2,3,4))
rdd.map(_*2)
.collect()
.foreach(println)
}
}
结果展示
注意:rdd.map(_*2)是简写，完整的应该写成rdd.map(item => item*2),在spark中，如果只要一个元素就可以使用_来代替
❖ 小功能：从服务器日志数据 apache.log 中获取用户请求 URL 资源路径
数据展示(取出红框的那一列)
@Test
def mapLogTest(): Unit = {
//读取文件数据
val rdd = sc.textFile("data/apache.log")
//数据都是以空格来进行分割的，我们只需要取最后一个即可
rdd.map(item => item.split(" ")(6))
.collect()
.foreach(println)
}
部分结果展示
跳转顶部
mapPartitions
➢ 函数签名
def mapPartitions[U: ClassTag]( f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
➢ 函数说明:将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处
理，哪怕是过滤数据。
@Test
def mapPartitionTest(): Unit = {
val rdd = sc.parallelize(Seq(1,2,3,4),2)
rdd.mapPartitions(_.map(_*2))
.collect()
.foreach(println)
}
结果展示
注意:可以以分区为单位进行数据转换 *** 作，但是会将整个分区的数据加载到内存进行引用，处理完数据是不会释放的，在数据量较大，内存较小的情况下不建议使用
❖ 小功能：获取每个数据分区的最大值
val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9), 2)
//因为mapPartition输入的是一个迭代器所以也必须返回一个迭代器
rdd.mapPartitions(
iter => {
List(iter.max).iterator
}
).collect()
.foreach(println)
}
结果展示
跳转顶部
map与mapPartition的区别
➢ 数据处理角度：Map 算子是分区内一个数据一个数据的执行，类似于串行 *** 作。而 mapPartitions 算子是以分区为单位进行批处理 *** 作。
➢ 功能的角度：Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，所以可以增加或减少数据
➢ 性能的角度：Map 算子因为类似于串行 *** 作，所以性能比较低，而是 mapPartitions 算子类似于批处理，所以性能较高。但是 mapPartitions 算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用 map *** 作。
mapPartitionsWithIndex
➢ 函数签名
def mapPartitionsWithIndex[U: ClassTag]( f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
➢ 函数说明:将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处
理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。
@Test
def mapPartitionsWithIndex(): Unit = {
val rdd = sc.parallelize(Seq(1, 2, 3, 4), 2)
rdd.mapPartitionsWithIndex(
(index, iter) =>
if (index == 1) {
iter
}else{
Nil.iterator
}
).collect()
.foreach(println)
}
结果展示
❖ 小功能：查看每个数据所在的分区
val rdd = sc.parallelize(Seq(1, 2, 3, 4), 2)
rdd.mapPartitionsWithIndex(
(index,iter) =>
iter.map((index,_))
).collect()
.foreach(println)
}
结果展示
跳转顶部
flatMap
➢ 函数签名:def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
➢ 函数说明:将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射
@Test
def flatMap():Unit = {
val rdd = sc.makeRDD(List("hello hadoop","hi spark"))
rdd.flatMap(iter => iter.split(" "))
.collect()
.foreach(println)
}
结果展示
❖ 小功能：将List(List(1,2),3,List(4,5))进行扁平化 *** 作
val rdd1 = sc.makeRDD(List(List(1,2),3,List(4,5)))
rdd1.flatMap(
data =>
data match {
case list: List[_] => list
case dat => List(dat)
}
).collect()
.foreach(println)
}
结果展示
跳转顶部
glom
➢ 函数签名:def glom(): RDD[Array[T]]
➢ 函数说明:将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变
def glom():Unit = {
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
//int => Array
val glomRDD: RDD[Array[Int]] = rdd.glom()
glomRDD.collect().foreach(data=>println(data.mkString(",")))
}
结果展示
❖ 小功能：计算所有分区最大值求和（分区内取最大值，分区间最大值求和）
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val maxRDD: Array[Int] = rdd.glom().map(_.max).collect()
println(maxRDD.sum)
结果展示
跳转顶部
groupBy
➢ 函数签名:def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]
➢ 函数说明:将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的 *** 作称之为 shuffle。极限情况下，数据可能被分在同一个分区中,一个组的数据在一个分区中，但是并不是说一个分区中只有一个组
@Test
def group(): Unit = {
val rdd1 = sc.makeRDD(List(1,2,3,4),2)
rdd1.groupBy(_%2).collect().foreach(println)
}
结果展示
❖ 小功能：将 List(“Hello”, “Spark”, “Scala”, “Hadoop”)根据单词首写字母进行分组。
@Test
def group(): Unit = {
val rdd = sc.makeRDD(List("Hello","Spark","Hadoop","Scala"),2)
//根据首字母分组
rdd.groupBy(_.charAt(0))
.collect()
.foreach(println)
}
结果展示
❖ 小功能：从服务器日志数据 apache.log 中获取每个时间段访问量。
数据展示(只需要取时间即可，日期暂时不考虑)
val rdd2 = sc.textFile("data/apache.log")
val timeRDD = rdd2.map(
line => {
val datas = line.split(" ")
val time = datas(3)
val sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
val date = sdf.parse(time)
val sdf1 = new SimpleDateFormat("HH")
val hour = sdf1.format(date)
(hour, 1)
}
).groupBy(_._1)
timeRDD.map {
case (hour, iter) => {
(hour, iter.size)
}
}.collect()
.foreach(println)
}
结果展示
❖ 小功能：WordCount。
数据展示
def main(args: Array[String]): Unit = {
//1.创建SparkContext
val conf = new SparkConf().setMaster("local[6]").setAppName("wordCount")
val sc = new SparkContext(conf)
//2.加载文件
val rdd1: RDD[String] = sc.textFile("data/wordcount.txt")
//3.处理
//拆分
val rdd2: RDD[String] = rdd1.flatMap(item => item.split(" "))
//指定词频
val rdd3: RDD[(String, Int)] = rdd2.map(item => (item, 1))
//聚合
val rdd4: RDD[(String, Int)] = rdd3.reduceByKey((curr, agg) => curr + agg)
//4.得到结果
val result: Array[(String, Int)] = rdd4.collect()
result.foreach(item => println(item))
}
结果展示
跳转顶部
filter
➢ 函数签名:def filter(f: T => Boolean): RDD[T]
➢ 函数说明:将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。
@Test
def filter():Unit = {
val rdd = sc.makeRDD(List(1,2,3,4,5))
//将技术留下来
rdd.filter(_%2==1)
.collect()
.foreach(println)
}
结果展示
❖ 小功能：从服务器日志数据 apache.log 中获取 2015 年 5 月 17 日的请求路径
数据展示
val rdd2 = sc.textFile("data/apache.log")
rdd2.map(
line => {
val time = line.split(" ")(3)
val sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
val sdf1 = new SimpleDateFormat("yyyy/MM/dd")
val date = sdf1.format(sdf.parse(time))
(date,line.split(" ")(6))
}
).filter(_._1.equals("2015/05/17"))
.collect()
.foreach(println)
部分结果展示
跳转顶部
sample
➢ 函数签名
def sample( withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T])
➢ 函数说明:根据指定的规则从数据集中抽取数据
@Test
def sample():Unit = {
val rdd = sc.makeRDD(List(1,2,3,4,5,6,7,8,9,10))
rdd.sample(false,0.4,1).collect().foreach(println)
}
结果展示
注意
- 第一个参数表示抽取数据后是否将数据放回（true表示放回，false表示丢弃）
- 第二个参数
当是抽取不放回时：参数表示每条数据可能被抽取的概率,不是抽取百分之多少
当是抽取放回时：参数表示每条数据可能被抽取的次数（可能是0，1，2，3……） - 第三个参数表示抽取数据时随机算法的种子:种子一样，不管抽取多少次数都一样，伪随机
思考：sample的用处？
- 在数据倾斜的时候使用，数据在经过shuffle阶段后，可能会导致数据倾斜，此时我们可以使用sample在数据大的分区进行抽取，查看是否是某一数据过多，可以提前进行处理
跳转顶部
distinct
➢ 函数签名
def distinct()(implicit ord: Ordering[T] = null): RDD[T] def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
➢ 函数说明:将数据集中重复的数据去重
@Test
def distinct():Unit = {
val rdd = sc.makeRDD(List(1,2,3,4,2,4,5,63,5,7,8))
rdd.distinct().collect().foreach(println)
}
结果展示
跳转顶部
coalesce
➢ 函数签名
def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty) (implicit ord: Ordering[T] = null) : RDD[T]
➢ 函数说明:根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率当 spark 程序中,存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少分区的个数，减小任务调度成本
@Test
def coalesce(): Unit = {
val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 6, 7), 3)
val newRDD = rdd.coalesce(2)
newRDD.saveAsTextFile("output")
}
结果展示
注意
- coalesce是由俩个参数的，一个就是缩减后的分区数
- 还有一个参数就是是否进行shuffle阶段
1️⃣第二个参数如果是false（就是默认的不进行shuffle）：这样可能会导致数据倾斜，spark一般是不会将数据进行分区的，也就是说，若原本有三个分区，缩减成两个后，可能是第一二两个分区的数据合并，也可能是二三分区数据合并
2️⃣第二个参数是true，也就是进行shuffle阶段，会将数据打乱重新进行分区 - 其实coalesce也是可以进行分区的增加到，但是增加时必须要进行shuffle *** 作
跳转顶部
sortBy
➢ 函数签名
def sortBy[K]( f: (T) => K, ascending: Boolean = true, numPartitions: Int = this.partitions.length) (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
➢ 函数说明:该 *** 作用于排序数据。在排序之前，可以将数据通过 f 函数进行处理，之后按照 f 函数处理的结果进行排序，默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程
@Test
def sortBy():Unit ={
val rdd = sc.makeRDD(List(("1",1),("3",4),("2",3)),2)
.sortBy(_._1,true)
.collect()
.foreach(println)
}
结果展示
注意
- sortby分区数量是不会发生变化的，但是因为要排序，会经过shuffle *** 作
- 默认是升序
跳转顶部
intersection、subtract、union和拉链
我们首先创建两个RDD
val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
intersection
➢ 函数签名：def intersection(other: RDD[T]): RDD[T]
➢ 函数说明：对源 RDD 和参数 RDD 求交集后返回一个新的 RDD,即俩者共有的部分
rdd1.intersection(rdd2).collect().foreach(println)
结果展示
问：如果两个RDD的类型不一致会怎么样？
- 答：类型不一致在编译时就会报错（不仅仅是在交集中，并集和差集的数据类型也要求一致，但是拉链不要求数据源一致）
union
➢ 函数签名：def union(other: RDD[T]): RDD[T]
➢ 函数说明：对源 RDD 和参数 RDD 求并集后返回一个新的 RDD，不管是否重复（就算rdd1[1,1,1],rdd2[1,1,1]最后的新的rdd[1,1,1,1,1,1]）
rdd1.union(rdd2).collect().foreach(println)
结果展示
subtract
➢ 函数签名：def subtract(other: RDD[T]): RDD[T]
➢ 函数说明：以一个 RDD 元素为主，去除两个 RDD 中重复元素，将其他元素保留下来。求差集
rdd1.subtract(rdd2).collect().foreach(println)
结果展示
拉链
➢ 函数签名：def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]
➢ 函数说明：将两个 RDD 中的元素，以键值对的形式进行合并。其中，键值对中的 Key 为第 1 个 RDD中的元素，Value 为第 2 个 RDD 中的相同位置的元素。
rdd1.zip(rdd2).collect().foreach(println)
结果展示
注意
- 两个数据源的分区数要求一致
- 两个数据源的元素数要求一致
跳转顶部
partitionBy
➢ 函数签名:def partitionBy(partitioner: Partitioner): RDD[(K, V)]
➢ 函数说明:将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitioner
@Test
def partitionBy(): Unit = {
val rdd = sc.makeRDD(List(1, 2, 3, 4),3)
//将数据类型转换成KV类型
val mapRDD: RDD[(Int, Int)] = rdd.map((_, 1))
//隐式转换
//只有kv类型的数据才能使用partitionBy,
mapRDD.partitionBy(new HashPartitioner(2)).saveAsTextFile("output")
}
结果展示
跳转顶部
reduceByKey
➢ 函数签名
def reduceByKey(func: (V, V) => V): RDD[(K, V)] def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
➢ 函数说明:可以将数据按照相同的 Key 对 Value 进行聚合
@Test
def reduceByKey(): Unit = {
val rdd = sc.makeRDD(List(("a",1),("b",4),("c",2),("a",3),("c",5)))
rdd.reduceByKey(_+_)
.collect()
.foreach(println(_))
}
结果展示
跳转顶部
groupByKey
➢ 函数签名
def groupByKey(): RDD[(K, Iterable[V])] def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])] def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
➢ 函数说明:将数据源的数据根据 key 对 value 进行分组
@Test
def groupByKey(): Unit = {
val rdd = sc.makeRDD(List(("a",1),("b",4),("c",2),("a",3),("c",5)))
rdd.groupByKey()
.collect()
.foreach(println(_))
}
结果展示
跳转顶部
reduceByKey和groupByKey的区别groupByKey
- 会导致数据打乱重组，存在shuffle *** 作
- shuffle *** 作必须落盘处理（落盘就是下图中中间的那段文件 *** 作，先汇总再由新的RDD读取），不能再内存中等待数据，会导致内存溢出
reduceByKey
- reduceByKey支持分区内预聚合功能，可以有效地减少shuffle落盘的数据量
从 shuffle 的角度：reduceByKey 和 groupByKey 都存在 shuffle 的 *** 作，但是 reduceByKey可以在 shuffle 前对分区内相同 key 的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而 groupByKey 只是进行分组，不存在数据量减少的问题，reduceByKey 性能比较高。
从功能的角度：reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组，不能聚合，所以在分组聚合的场合下，推荐使用 reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey
跳转顶部
➢ 函数签名
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]
➢ 函数说明:将数据根据不同的规则进行分区内计算和分区间计算
求每个分区内相同key的最大值，然后分区间相加
@Test
def aggregateByKey(): Unit = {
val rdd = sc.makeRDD(List(("b", 2), ("a", 1), ("a", 4), ("a", 2), ("b", 3), ("b", 5)), 2)
rdd.aggregateByKey(0)(
(x, y) => math.max(x, y),
(x, y) => x + y
).collect()
.foreach(println(_))
}
结果展示
注意
aggregateByKey存在柯里化，有两个参数列表
俩俩相加：第一次是输入的初始值和第一个元素进行运算，然后是第一个（或者是初始值与第一个运算后的结果）与第二个运算
第一个参数列表：需要传递一个参数，表示初始值，主要用于辅助计算
- 如：全部数据乘以0.8，这里就可以写0.8，
- 若求最大值，但是第一个数字该如何比较？可以在这里填写一个足够小的数字，辅助计算
第二个参数列表：
- 第一个参数表示分区内计算规则
- 第二个参数表示分区间计算规则
图解如下
跳转顶部
foldByKey
➢ 函数签名:def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
➢ 函数说明:当分区内计算规则和分区间计算规则相同时，aggregateByKey 就可以简化为 foldByKey
按照key求和
@Test
def foldByKey():Unit = {
val rdd = sc.makeRDD(List(("b", 2), ("a", 1), ("a", 4), ("a", 2), ("b", 3), ("b", 5)),2)
rdd.foldByKey(0)(_+_)
.collect()
.foreach(println(_))
}
结果展示
跳转顶部
combineByKey
➢ 函数签名
def combineByKey[C]( createCombiner: V => C, mergevalue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
➢ 函数说明:最通用的对 key-value 型 rdd 进行聚集 *** 作的聚集函数（aggregation function）。类似于aggregate()，combineByKey()允许用户返回值的类型与输入不一致。
小练习：将数据 List((“a”, 88), (“b”, 95), (“a”, 91), (“b”, 93), (“a”, 95), (“b”, 98))求每个 key 的平
均值
@Test
def combineByKey(): Unit = {
val rdd = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)))
rdd.combineByKey(
createCombiner = (_, 1), //curr(88,1) next(91)
mergevalue = (curr: (Int, Int), next:(Int))=>(curr._1 + next,curr._2 + 1),
mergeCombiners = (curr:(Int,Int),agg:(Int,Int))=>(curr._1 + agg._1,curr._2 + agg._2)
).map(item =>(item._1,item._2._1/item._2._2))
.collect()
.foreach(println(_))
}
结果展示
注意
- 两两运行处理
- 方法需要三个参数
- 第一个参数表示：将相同key的第一个数据进行结构转换，实现 *** 作,且第一步只作用于分组后的第一条数据
- 第二个参数表示：分区内的计算规则，这时是已经经过上一步处理好的数据和下一个没有处理的数据进行处理
- 第三个参数表示：分区间的计算规则
跳转顶部
sortByKey
➢ 函数签名
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length) : RDD[(K, V)]
➢ 函数说明:在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口(特质)，返回一个按照 key 进行排序的
@Test
def sortByKey():Unit = {
val rdd1 = sc.makeRDD(List((1,2),(2,2),(3,2),(5,2),(6,2),(4,2)))
val rdd2 = sc.makeRDD(List(("a",1),("f",1),("b",1),("e",1),("c",1)))
rdd1.sortByKey(false).collect().foreach(println(_))
rdd2.sortByKey(true).collect().foreach(println(_))
}
结果展示
跳转顶部
join
➢ 函数签名：def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
➢ 函数说明：在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD
@Test
def join(): Unit = {
val rdd1 = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
val rdd2 = sc.makeRDD(List(("a", 4), ("b", 5), ("d", 6)))
rdd1.join(rdd2)
.collect()
.foreach(println(_))
}
结果展示
注意
- 两个不同数据源的数据，相同key的value会连接在一起，形成元组
- 如果key不相同那么数据不会出现在结果中
- 多个key相同的会依次匹配
跳转顶部
leftOuterJoin和rightOuterJoin
➢ 函数签名:def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
➢ 函数说明:类似于 SQL 语句的左外连接
@Test
def leftOuterJoin(): Unit = {
val rdd1 = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 4)))
val rdd2 = sc.makeRDD(List(("a", 4), ("b", 5), ("d", 6)))
rdd1.leftOuterJoin(rdd2).collect().foreach(println(_))
}
结果展示
rightOuterJoin
rdd1.rightOuterJoin(rdd2).collect().foreach(println(_))
结果展示
跳转顶部
cogroup
➢ 函数签名:def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
➢ 函数说明:在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable,Iterable))类型的 RDD
@Test
def cogroup():Unit = {
val rdd1 = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 4)))
val rdd2 = sc.makeRDD(List(("a", 4), ("b", 5), ("d", 6)))
rdd1.cogroup(rdd2).collect().foreach(println(_))
}
结果展示
跳转顶部
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）