一、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的partitionsDoing（并行执行任务数）= job的Executor数目 * 每个Executor核心数总共执行批次=Tasks/Doing  （如果不是整除，则加1）

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

4.参考

（1）https://blog.csdn.net/yu0_zhang0/article/details/80454517