Spark任务输出文件过程详解

系统管理员 2022-01-14 09:21 357阅读 0赞

### 文章目录 ###

*   *  一、Spark任务输出文件的总过程
     *  二、Commit细节分析
     *   *  1、commitTask 介绍
         *   *  1.1、判断是否需要commit
             *  1.2、task的commit细节
         *  2、commitJob 介绍
     *  三、V1和V2 commiter版本比较
     *   *  1、性能方面
         *  2、数据一致性方面
         *  3、总结
     *  参考资料

## 一、Spark任务输出文件的总过程 ##

当一个Job开始执行后，输出文件的相关过程大概如下：

1、Job启动时创建一个目录: $\{output.dir\}/\_temporary/$\{appAttemptId\} 作为本次运行的**输出临时目录**

2、当有task开始运行后，会创建 $\{output.dir\}/\_temporary/$\{appAttemptId\}/\_temporary/$\{taskAttemptId\}/$\{fileName\} 文件，**后面这个task的所有输出都会被写到这个文件中**

3、当task运行完后，需要检查是否要commit，如果需要commit，会调用**OutputCommitter\#commitTask()方法**。commit的细节后面说

4、等整个Job执行完就调用**OutputCommitter\#commitJob()方法**。具体的过程也在下面介绍commit时说。

output.dir表示用户指定的输出目录，appAttemptId表示任务的attemptId，一般从0开始一直递增。taskAttemptId表示task的attemptId，比如taskId是0，第一次运行，这个id就是0.0。taskAttemptId的输出格式是这样的(**toString方法实现**)  
1、 当task的type是map时，格式为attempt\_\{时间表达字符串\}\_\{stageId\}\_m\_\{sparkPartitionId\}\_\{TaskId\}，比如attempt目录是attempt\_20190801032358\_0020\_m\_000093\_7683，则表示stageId为20，type是m，对应map task(reduce task则为r)，sparkPartitionId表示该task在stage中的id是93，最后，它对应的TaskId是7683。

2、 当task的type是reduce时，格式为attempt\_\{时间表达字符串\}\_\{rddId\}\_r\_\{sparkPartitionId\}\_\{attemptNumber\}，比如attempt目录是attempt\_20190801042010\_1478\_r\_000970\_1，则表示对应的rddId是1478，r表示type是reduce task，970表示partiton的Id，最后的1表示这是该task的第二次运行（第一次运行是0）—— 一个stage中可能会有多个rdd，所以stageId并不等于rddId

> OutputCommitter 只是一个抽象类，spark运行时会从配置中获取指定的实现类，如果配置中没指定，spark默认会使用 org.apache.hadoop.mapred.FileOutputCommitter 的实现。

## 二、Commit细节分析 ##

### 1、commitTask 介绍 ###

#### 1.1、判断是否需要commit ####

当task执行完后，会去检查以下状态，如果下面的条件达成，就不会执行commit

*  $\{output.dir\}/\_temporary/$\{appAttemptId\}/\_temporary/$\{taskAttempt\} 目录不存在 (说明这个task的临时输出目录不存在，明显是有问题的)
 *  如果开启了Output commit coordination，就需要通过rpc询问Driver是否可以commit (根据spark.hadoop.outputCommitCoordination.enabled参数，默认为true.如果开启了推测执行，这个一定要设置为true)
 *  Driver的CommitCoordinator判断task运行失败 （task运行失败就没必要commit了）
 *  Driver的CommitCoordinator判断该task的其他attempt已经commit过了 （如果commit的taskAttemptId和当前一样，那么可以再次commit，说明task commit是一个幂等的操作）

#### 1.2、task的commit细节 ####

因为我们大部分情况下用的都是FileOutputCommitter，所以下面主要介绍一下这个类的commitTask实现。

FileOutputCommitter的实际commitTask细节和参数 mapreduce.fileoutputcommitter.algorithm.version 有关(默认值是1)。

当**mapreduce.fileoutputcommitter.algorithm.version=1**时：

commit的操作是将 $\{output.dir\}/\_temporary/$\{appAttemptId\}/\_temporary/$\{taskAttemptId\} 重命名为 $\{output.dir\}/\_temporary/$\{appAttemptId\}/$\{taskId\}

当**mapreduce.fileoutputcommitter.algorithm.version=2**时：

commit的操作是将 $\{output.dir\}/\_temporary/$\{appAttemptId\}/\_temporary/$\{taskAttemptId\} 下的文件移动到 $\{output.dir\} 目录下 （也就是最终的输出目录）

> taskId的输出格式就是taskAttemptId少了最后具体的Id的输出格式，如之前的attempt目录 attempt\_20190801032358\_0020\_m\_000093\_7683 会重命名成 task\_20190801032358\_0020\_m\_000093

> spark任务可以通过设置spark配置 spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2来开启版本2的commit逻辑

> 在hadoop 2.7.0之前，FileOutputCommitter的实现没有区分版本，统一都是使用version=1的commit逻辑。因此如果spark的hadoop依赖包版本如果低于2.7.0,设置mapreduce.fileoutputcommitter.algorithm.version=2是没有用的

### 2、commitJob 介绍 ###

Job执行完后，会调用commitJob方法，我们还是看一下FileOutputCommitter的实现：

commitJob的细节也和mapreduce.fileoutputcommitter.algorithm.version 参数有关(默认值是1)

当**mapreduce.fileoutputcommitter.algorithm.version=1**时：

由 Driver 单线程遍历所有 committedTaskPath，也就是$\{output.dir\}/\_temporary/$\{appAttemptId\} 下的所有文件，然后移动到 $\{output.dir\} 目录下。然后创建\_SUCCESS表示任务结束

当**mapreduce.fileoutputcommitter.algorithm.version=2**时：

只需要创建\_SUCCESS文件，因为输出文件在task执行完后就已经移动到输出目录了

> 在commitJob完后，spark还会执行cleanupJob将$\{output.dir\}/\_temporary 目录删除

## 三、V1和V2 commiter版本比较 ##

**mapreduce.fileoutputcommitter.algorithm.version** 参数对文件输出有很大的影响，下面总结一下两种版本在各方面的优缺点。

### 1、性能方面 ###

v1在task结束后只是将输出文件拷到临时目录，然后在job结束后才由Driver把这些文件再拷到输出目录。如果文件数量很多，Driver就需要不断的和NameNode做交互，而且这个过程是单线程的，因此势必会增加耗时。如果我们碰到有spark任务所有task结束了但是任务还没结束，很可能就是Driver还在不断的拷文件。

v2在task结束后立马将输出文件拷贝到输出目录，后面Job结束后Driver就不用再去拷贝了。

因此，在性能方面，v2完胜v1。

### 2、数据一致性方面 ###

v1在Job结束后才批量拷文件，其实就是两阶段提交，它可以保证数据要么全部展示给用户，要么都没展示（当然，在拷贝过程中也无法保证完全的数据一致性，但是这个时间一般来说不会太长）。如果任务失败，也可以直接删了\_temporary目录，可以较好的保证数据一致性。

v2在task结束后就拷文件，就会造成spark任务还未完成就让用户看到一部分输出，这样就完全没办法保证数据一致性了。另外，如果任务在输出过程中失败，就会有一部分数据成功输出，一部分没输出的情况。

因此在数据一致性方面，v1完胜v2

### 3、总结 ###

很明显，如果我们执着于性能，不在乎数据输出时的一致性，完全可以将**mapreduce.fileoutputcommitter.algorithm.version**设置为2来提高性能。

但是如果我们对输出要求很高的数据一致性，那么最好不要为了性能将**mapreduce.fileoutputcommitter.algorithm.version**设置为2。

## 参考资料 ##

[https://issues.apache.org/jira/browse/MAPREDUCE-4815][https_issues.apache.org_jira_browse_MAPREDUCE-4815]

[https://zhuanlan.zhihu.com/p/45351972][https_zhuanlan.zhihu.com_p_45351972]

[https://mp.weixin.qq.com/s?\_\_biz=MzU3NTE2NzAxNQ==&mid=2247484099&idx=1&sn=0a0a3a1f407d30a22dcfbd85fab488e6&chksm=fd260d8bca51849d94e8df9f2249462d5a5dfc8079c45b4d9ab489aaea77c0ce14108a948f94&token=2064668791&lang=zh\_CN\#rd][https_mp.weixin.qq.com_s_biz_MzU3NTE2NzAxNQ_mid_2247484099_idx_1_sn_0a0a3a1f407d30a22dcfbd85fab488e6_chksm_fd260d8bca51849d94e8df9f2249462d5a5dfc8079c45b4d9ab489aaea77c0ce14108a948f94_token_2064668791_lang_zh_CN_rd]

[https_issues.apache.org_jira_browse_MAPREDUCE-4815]: https://issues.apache.org/jira/browse/MAPREDUCE-4815
[https_zhuanlan.zhihu.com_p_45351972]: https://zhuanlan.zhihu.com/p/45351972
[https_mp.weixin.qq.com_s_biz_MzU3NTE2NzAxNQ_mid_2247484099_idx_1_sn_0a0a3a1f407d30a22dcfbd85fab488e6_chksm_fd260d8bca51849d94e8df9f2249462d5a5dfc8079c45b4d9ab489aaea77c0ce14108a948f94_token_2064668791_lang_zh_CN_rd]: https://mp.weixin.qq.com/s?__biz=MzU3NTE2NzAxNQ==&mid=2247484099&idx=1&sn=0a0a3a1f407d30a22dcfbd85fab488e6&chksm=fd260d8bca51849d94e8df9f2249462d5a5dfc8079c45b4d9ab489aaea77c0ce14108a948f94&token=2064668791&lang=zh_CN#rd