SparkCore（11）：RDD概念和创建RDD两种方法，以及RDD的Partitions以及并行度理解

本文主要是介绍SparkCore（11）：RDD概念和创建RDD两种方法，以及RDD的Partitions以及并行度理解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、RDD概念

1.概念

Resilient Distributed Datasets弹性分布式数据集，默认情况下：每一个block对应一个分区，一个分区会开启一个task来处理。

（a）Resilient：可以存在给定不同数目的分区、数据缓存的时候可以缓存一部分数据也可以缓存全部数据
（b）Distributed：分区可以分布到不同的executor执行(也就是不同的worker/NM上执行)
（c）Datasets：内部存储是数据

2.特性

（1）是一系列的分片，分区
（2）每个分片有一个方法来做计算
（3）rdd会有依赖其他rdd的操作，可以通过wordCountRDD.toDebugString来查看
（4）（可选项）如果rdd是二元组，就会存在分区器（默认是hashpartition）
（5）（可选项）最佳位置。数据在哪台机器上，任务就启在哪个机器上，数据在本地上，不用走网络。不过数据进行最后汇总的时候就要走网络。（hdfs file的block块）

二、RDD创建方法

1.外部数据

val path = "hdfs://192.168.31.3:8020/page_views.data"
val originalRdd: RDD[String] = sc.textFile(path)

2.内存中数据：基于序列化进行创建

scala> val seq = List(1,2,3,4,5,6,7)
seq: List[Int] = List(1, 2, 3, 4, 5, 6, 7)scala> val rdd2 = sc.parallelize(seq)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:29

三、关键：Partitions以及并行度

1.RDD的partitions数目

（1）读取数据阶段，对于textFile来说，没有在方法中的指定分区数，则默认为min(defaultParallelism,2)，而defaultParallelism对应的就是spark.default.parallelism。如果是从hdfs上面读取文件，其分区数为文件block数(128MB/block)
（2）在Map阶段partition数目保持不变。
（3）在Reduce阶段，RDD的聚合会触发shuffle操作，聚合后的RDD的partition数目跟具体操作有关，例如repartition操作会聚合成指定分区数，还有一些算子是可配置的。

2.并行度

定义：一个job一次所能执行的task数目，即一个job对应的总的core资源个数

执行一个job的task的并行数 = job的Executor数目 * 每个Executor的core个数。
例如提交scalaProjectMaven.jar的spark任务

date=`date +"%Y%m%d%H%M"`
/opt/modules/spark-2.1.0-bin-2.7.3/bin/spark-submit \
--master yarn \
--deploy-mode client \
--name spark_shell_${date} \
--class wordcount \
--driver-memory   8G \
--driver-cores 4 \    
--executor-memory 4G \
--executor-cores 4 \
--num-executors 3 \
--conf spark.app.coalesce=1 \
/opt/datas/scalaProjectMaven.jar

提交job的并行数=3*4=12，即每一个批次执行12个task，对应12个partitions。

3.partitions和并行度关系

一个partition对应一个要做的task，一个executor的core执行一个task。
Tasks（一个RDD的总task数）=该RDD的partitions
Doing（并行执行任务数）= job的Executor数目 * 每个Executor核心数
总共执行批次=Tasks/Doing （如果不是整除，则加1）

例如：Tasks=50，Doing=30，则执行两次，第一次执行30个task，第二次执行20个task。