spark transform算子_Spark高性能Job

知乎导入md文件会失真，无语，将就着看吧！原博客地址为：Spark高性能Job

1.1 Job

遇到一个action算子就会提交一个job，常见的transformation算子以及Action算子：

• Transformation
  • map, mapPartitions, flatMap, filter, union, groupbyKey, repartition, cache
• Action
  • reduce, collect, show, count, foreach, save一系列操作

1.2 Task

​ task是spark最小的执行单元，task的数量就是stage的并行度，分配给不同的executor去执行。RDD在计算的时候，每个分区都会启一个task，这就是我们常说的数据并行！在map阶段，partition分区的数量保持不变，在reduce阶段，RDD聚合会触发shuffle操作。

2 Spark Shuffle

shuffle是spark job中一个比较重要的阶段，发生在map与reduce之间。

2.1 举例分析

对于上述的reduceByKey，涉及到需要将相同的key进行聚合。对于Stage1中的每个分区的数据，其输入可能存在于Stage0中的每个分区，因此需要从上游的每一个分区所在的机器拉取数据，这个过程称为shuffle。

2.2 Shuffle Write

shuffle write操作发生在ShuffleMapTask，Spark中的task分为以下两种类型：

• ShuffleMapTask
• 负责rdd之间的transform，map的输出也就是shuffle write
• ResultTask
• job最后阶段的执行任务，也就是action操作。

2.2.1 shuffle write分析

• Hash Based Shuffle

​ 上图有四个ShuffleMapTask，假设在这四个都在一个worker node上运行，CPU的核为2,可以同时运行2个task（一个核运行两个线程，可能是超线程技术）。

​ 那么每个执行shuffle write的task，要为下一个stage创建多少个磁盘文件呢？很简单，下一个stage的task有多少个，当前stage的每个Map task就要创建多少份磁盘文件。比如下一个stage总共有100个task，那么当前stage的每个task都要创建100份磁盘文件。如果当前stage有50个task，总共有10个Executor，每个Executor执行5个Task，那么每个Executor上总共就要创建500个磁盘文件，所有Executor上会创建5000个磁盘文件。由此可见，未经优化的shuffle write操作所产生的磁盘文件的数量是极其惊人的。

​ 每个task包含R个缓冲区，R=Reducer的个数（也就是下个stage中task的个数），缓冲区被称为bucket。在将数据写入磁盘之前，会先将数据写入内存缓冲中，当内存缓冲填满之后，才会溢写到磁盘文件中去

但上述可能会出现下面几个问题：

• 本地磁盘根据bucket产生的blockfile很多，ShuffleMap task产生R个blockfile，M个ShuffleMapTask产生M×R个文件。一般Spark Job的M与R都很大，因此磁盘上会有大量的blockfile文件
• 缓冲区内存占用空间大
  每个 ShuffleMapTask 需要开 R 个 bucket，M 个 ShuffleMapTask 就会产生 M * R 个 bucket。实际情况下，在一个worker node上，可以并行运行cores个ShuffleMapTask，一个机器上的bucket个数达到cores×R个，这会占用大量的内存空间。

对于第二个问题，由于从内存往磁盘写数据一定得开缓冲区（内存与磁盘速度不匹配），所以对于第二个问题而言，没有较好的方法解决！但第一个问题可以通过下面的方法解决。

• Consolidation机制的Shuffle

​ 在一个 core 上连续执行的 ShuffleMapTasks 可以共用一个输出文件 ShuffleFile。先执行完ShuffleMapTask 形成 Shuf