背景
當(dāng)spark最終輸出文件stage，task數(shù)量非常多時(shí)准浴，會(huì)在driver端單線程執(zhí)行大量的rename操作事扭，比較耗時(shí)，如何解決呢乐横？
分析
罪魁禍?zhǔn)浊箝希褪莌adoop代碼里的commitJobInternal函數(shù)，這里會(huì)單線程調(diào)mergePaths葡公，會(huì)把每個(gè)task輸出在_temporary目錄的結(jié)果罐农，移動(dòng)到最終的輸出目錄。
@VisibleForTesting
protected void commitJobInternal(JobContext context) throws IOException {
if (hasOutputPath()) {
Path finalOutput = getOutputPath();
FileSystem fs = finalOutput.getFileSystem(context.getConfiguration());
// 如果v1催什，就執(zhí)行
if (algorithmVersion == 1) {
for (FileStatus stat: getAllCommittedTaskPaths(context)) {
mergePaths(fs, stat, finalOutput);
}
}
if (skipCleanup) {
LOG.info("Skip cleanup the _temporary folders under job's output " +
"directory in commitJob.");
} else {
// delete the _temporary folder and create a _done file in the o/p
// folder
try {
cleanupJob(context);
} catch (IOException e) {
if (ignoreCleanupFailures) {
// swallow exceptions in cleanup as user configure to make sure
// commitJob could be success even when cleanup get failure.
LOG.error("Error in cleanup job, manually cleanup is needed.", e);
} else {
// throw back exception to fail commitJob.
throw e;
}
}
}
// True if the job requires output.dir marked on successful job.
// Note that by default it is set to true.
if (context.getConfiguration().getBoolean(
SUCCESSFUL_JOB_OUTPUT_DIR_MARKER, true)) {
Path markerPath = new Path(outputPath, SUCCEEDED_FILE_NAME);
// If job commit is repeatable and previous/another AM could write
// mark file already, we need to set overwritten to be true explicitly
// in case other FS implementations don't overwritten by default.
if (isCommitJobRepeatable(context)) {
fs.create(markerPath, true).close();
} else {
fs.create(markerPath).close();
}
}
} else {
LOG.warn("Output Path is null in commitJob()");
}
}
如果mapreduce.fileoutputcommitter.algorithm.version配成v1涵亏，就會(huì)執(zhí)行上面的for循環(huán)，那有v1就有v2，v2又是怎樣的呢溯乒？
簡(jiǎn)單的說，v1就是每個(gè)task執(zhí)行的最終結(jié)果豹爹，輸出到_temporary目錄裆悄，所有task執(zhí)行結(jié)束后，由CommitCoordinator臂聋，一并執(zhí)行rename到最終的輸出目錄光稼。
而v2則是task執(zhí)行結(jié)果執(zhí)行輸出到最終的輸出目錄。
那么為什么不用v2呢孩等？v2有什么問題呢艾君？
答案是有問題的，一致性問題肄方。所謂一致性冰垄，就是正確性。
壞情況1：
如果執(zhí)行100個(gè)task权她，50個(gè)跑完了虹茶，剩下的掛了，在v2里隅要，輸出的結(jié)果就會(huì)直接殘留在最終的輸出目錄蝴罪。
壞情況2：
同樣是執(zhí)行100個(gè)task，但每個(gè)執(zhí)行的速度不一樣步清，有95個(gè)很快跑完了要门，5個(gè)還沒跑完，這時(shí)有人來訪問輸出文件廓啊，得到的就是個(gè)錯(cuò)誤的結(jié)果欢搜。
具體可以看下Spark CommitCoordinator 保證數(shù)據(jù)一致性這篇文章。
所以如果前面的一致性問題影響不大谴轮，那就用v2狂巢，否則依然只能用v1。v1該怎么優(yōu)化呢书聚？
- 減少輸出文件的那個(gè)stage的task數(shù)唧领，其實(shí)task數(shù)太多，很可能是文件太小雌续。
- 前面hadoop源碼里單線程的for循環(huán)斩个，是不是可以改成多線程？
參考
分布式系統(tǒng)中的一致性
Spark CommitCoordinator 保證數(shù)據(jù)一致性
二階段提交-維基百科
