mapPartitions

mapPartitions与map转换操作类似，只不过映射函数的输入参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器
def mapPartitions[U](f: Iterator[Double] => Iterator[U],preservesPartitioning: Boolean)(implicit evidence$6: scala.reflect.ClassTag[U]): org.apache.spark.rdd.RDD[U]

传入的参数是一个迭代器
perservesPartitioning指明mapPartitions是否保留父RDD的partitions的分区信息

scala版本

将每个分区内元素乘2

// 分为三个区val rdd1: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6),3)// mapPartitions需要传递val mapPartitionsRDD: RDD[Int] = rdd1.mapPartitions((f: Iterator[Int]) => {println(f)f.map(_ * 2)})mapPartitionsRDD.collect().foreach(println)

java版本

JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6));
JavaRDD<Integer> mapPartitionsRDD = rdd1.mapPartitions(new FlatMapFunction<Iterator<Integer>, Integer>() {@Overridepublic Iterator<Integer> call(Iterator<Integer> i) throws Exception {List<Integer> list = new ArrayList<>();while (i.hasNext()) {Integer x = i.next();list.add(x);}return list.iterator();}});
List<Integer> collect = mapPartitionsRDD.collect();
for (Integer i : collect) {System.out.println(i);
}

mapPartitionsWithIndex

mapPartitionWithIndex和mapPartitions功能类似，只是输入参数时多了一个分区的ID
def mapPartitionsWithIndex[U](f: (Int, Iterator[Double]) => Iterator[U],preservesPartitioning: Boolean)(implicit evidence$9: scala.reflect.ClassTag[U]): org.apache.spark.rdd.RDD[U]

scala版本

val rdd1: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6),3)
val f =(x:Int,y:Iterator[Int])=>{println(x+"号分区")y.map(_*2)
}
val  mapPartitionsWithIndexRDD: RDD[Int] = rdd1.mapPartitionsWithIndex(f)
mapPartitionsWithIndexRDD.collect.foreach(println)

java版本

JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6),3);
Function2<Integer, Iterator<Integer>, Iterator<String>> function2 = new Function2<Integer, Iterator<Integer>, Iterator<String>>() {@Overridepublic Iterator<String> call(Integer v1, Iterator<Integer> v2) throws Exception {System.out.println(v1 + "号分区");List<String> list = new ArrayList<>();while (v2.hasNext()) {String s = "hello" + v2.next();list.add(s);}return list.iterator();}
};
// 第二个参数用于指明是否保留父RDD的partitions的分区信息
JavaRDD<String> mapPartitionsWithIndexRDD = rdd1.mapPartitionsWithIndex(function2, false);
List<String> collect = mapPartitionsWithIndexRDD.collect();
for (String s : collect) {System.out.println(s);
}

Spark RDD算子（八）mapPartitions， mapPartitionsWithIndex相关推荐

Spark学习之Spark RDD算子
个人主页zicesun.com 这里,从源码的角度总结一下Spark RDD算子的用法. 单值型Transformation算子 map /*** Return a new RDD by applyi ...
spark RDD算子大全
目录 map()&&flatMap() map()&&mapPartitions() mapPartitionsWithIndex() filter() take()& ...
Spark RDD算子（transformation + action）
概念 RDD(Resilient Distributed Dataset)叫做弹性分布式数据集,是Spark中最基本的数据抽象,它代表一个不可变.可分区.里面的元素可并行计算的集合.RDD具有数据流模 ...
Spark RDD算子介绍
Spark学习笔记总结 01. Spark基础 1. 介绍 Spark可以用于批处理.交互式查询(Spark SQL).实时流处理(Spark Streaming).机器学习(Spark MLlib) ...
Spark—RDD算子使用IDEA-Scala操作练习：请根据磁盘文件“数据集”data01.txt，该数据集包含了某大学计算机系的成绩，计算下列问题。
一.数据源文件下载 https://download.csdn.net/download/weixin_45947938/66589736 二.问题描述请根据给定的实验数据,在idea中通过Scal ...
Spark RDD算子(四) mapToPair、flatMapToPair
目录一.mapToPair 二.flatMapToPair 一.mapToPair 将每一行的第一个单词作为键,1 作为value创建pairRDD scala版本 scala没有mapToPair ...
spark常用RDD算子汇总（java和scala版本）
github: https://github.com/zhaikaishun/spark_tutorial spark RDD的算子挺多,有时候如何灵活的使用,该如何用一下子想不起来,这一段时间将s ...
spark封神之路(7)-RDD算子详解第一部分
1 算子简介算子是一个函数空间到函数空间上的[映射]O:X→X.广义上的算子可以推广到任何空间,如[内积空间]等. RDD上的方法称为算子在 RDD 上支持 2 种操作: transformati ...
大数据——Spark RDD常用算子总结
Spark的核心是建立在同一的抽象弹性分布式数据集(Resilient Distributed Datasets,RDD)之上的,这使得Spark的各个组件可以无缝的进行集成,能够在同一个应用程序中完 ...

Spark RDD算子（八）mapPartitions， mapPartitionsWithIndex

Spark RDD算子（八）

mapPartitions

scala版本

java版本

mapPartitionsWithIndex

scala版本

java版本

Spark RDD算子（八）mapPartitions， mapPartitionsWithIndex相关推荐

最新文章

热门文章