mapPartitions/mapPartitionsWithIndex
这 两个transform中:mapPartitions与map的区别是map中是对每个partition中的iterator执行map操作,对 map过程中的每一条record进行传入的function的处理,而mapPartitions是把partition中整个iterator传给 function进行处理.如果是map操作,你并不能知道这个iterator什么时候结束,但mapPartitions时给你的是一个 iterator,所以你的函数中知道这个iterator什么时候会结束.而mapPartitionsWithIndex的函数是在 mapPartitions的基础上,多了一个传入参数,这个传入参数就是对应的partition的index.
mapPartitions的函数定义:
def mapPartitions[U: ClassTag](
这个函数的定义部分为一个参数,是对应partition的数据的iterator,
f: Iterator[T] => Iterator[U],
这个参数false表示在生成的MapPartitionsRDD中不包含partitioner算子.
preservesPartitioning: Boolean = false): RDD[U] = withScope {
val cleanedF = sc.clean(f)
直接生成一个MapPartitionsRDD,在生成这个rdd时,通过传入的function,这个function与map对应的function不同的是,function直接拿到一个iterator进行操作.
new MapPartitionsRDD(
this,
(context: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(iter),
preservesPartitioning)
}
mapPartitionsWithIndex的函数定义:
这个函数的处理方法可以看到与mapPartitions基本上相同,不同的地方是f(function)的定义部分,此部分多出一个int类型的参数,这个参数是对应的iterator所在的partition的index.
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U] = withScope {
val cleanedF = sc.clean(f)
new MapPartitionsRDD(
this,
(context: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(index, iter),
preservesPartitioning)
}
相关推荐
spark-core_2.11-2.0.0.jar比spark-core_2.11-1.5.2.jar少了org.apache.spark.Logging.class,故此把缺少的class放到spark-core_2.11-1.5.2.logging.jar里面
spark-sql_2.11-2.4.0-cdh6.1.1.jar
mongodb-spark官方连接器,运行spark-submit --packages org.mongodb.spark:mongo-spark-connector_2.11:1.1.0可以自动下载,国内网络不容易下载成功,解压后保存到~/.ivy2目录下即可。
spark-core_2.10-1.1.0
spark_2_7_7
spark-launcher_2.11-2.1.3-SNAPSHOT.jar
spark-streaming_2.11-2.1.3-SNAPSHOT.jar
比maven仓库下载快很多
spark_2_6_3_online.rar openfire的客户端
spark-core_2.11-1.6.0.jar spark-core_2.11-1.6.0.jar
mongodb spark连接器,适用版本spark2.1.X ,Scala2.11.X, java 6 or later,mongodb 2.6 or later,请根据上面的版本选择,不然会报各种错误
spark 读取excel 的jar 包
spark_2_5_8_online 内网聊天工具 可以是内网供作的人们方便的进行沟通
spark-tags_2.11-2.1.3-SNAPSHOT.jar
Spark_for_PythonSpark_for_PythonSpark_for_PythonSpark_for_PythonSpark_for_PythonSpark_for_PythonSpark_for_Python
spark-sql_2.11-2.1.3-SNAPSHOT_bak.jar
博客https://blog.csdn.net/lsshlsw/article/details/82670508 spark_prometheus_metrics.json
解决Spark升级到2.1后测试streaming-kafkaIDEA报java.lang.NoClassDefFoundError: org/apache/spark/Logging错误,将这个jar包放到spark的lib目录中,将这个jar包导进IDEA中的依赖,文档中除资源地址外,还有详细图解...
spark-yarn_2.11-2.1.3-SNAPSHOT.jar
spark-mllib_2.11-2.1.3-SNAPSHOT.jars