Hadoop MapReduce：分片失败时终止文件后续处理的实现方法

这个需求我在之前的大数据处理项目里刚好实现过，核心要搞定两个关键：只要有一个分片失败就立刻停掉整个作业，以及绝对不能在失败时往输出目录写东西。下面是一步步的实现方案：

默认Hadoop会对失败的Mapper进行重试（默认重试4次），这不符合我们“立即停止”的要求。所以首先要做两件事：

• 在Mapper中捕获处理失败的异常，直接抛出致命异常（比如IOException），不要自行处理吞掉异常；
• 配置作业的mapreduce.map.maxattempts参数为1，确保失败的Mapper不会被重试，直接标记作业为失败状态。

示例Mapper代码：

public class FailFastCSVMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        try {
            // 这里写你的CSV行处理逻辑
            processCSVRecord(value.toString());
            // 正常输出（如果需要的话）
            context.write(new Text("processed"), new IntWritable(1));
        } catch (Exception e) {
            // 一旦分片处理出错，直接抛出异常终止当前Mapper，并触发作业失败
            throw new IOException(String.format("分片 %s 处理失败，终止整个作业", context.getInputSplit()), e);
        }
    }

    private void processCSVRecord(String line) throws Exception {
        // 模拟分片失败场景：比如某行数据格式错误
        if (line.contains("invalid_data")) {
            throw new RuntimeException("发现无效数据，分片处理失败");
        }
        // 正常业务逻辑...
    }
}

Hadoop的默认输出机制其实已经具备原子性：作业运行时会把临时数据写到_temporary子目录，只有当整个作业完全成功时，才会把临时目录的内容移到最终输出目录，失败则会自动清理临时文件。但我们可以额外配置参数强化这个行为：

Configuration conf = new Configuration();
// Mapper失败后不重试，直接终止作业
conf.setInt("mapreduce.map.maxattempts", 1);
// 开启作业失败时的增量清理，及时删除临时文件
conf.setBoolean("mapreduce.job.failures.incremental.cleanup", true);
// 禁止作业失败时保留部分输出标记文件
conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs", false);

为了更稳妥，我们可以添加一个自定义的JobListener，在作业失败时主动清理输出目录，避免极端情况下的残留：

Job job = Job.getInstance(conf, "FailFastCSVProcessing");
// 设置作业的Mapper类
job.setMapperClass(FailFastCSVMapper.class);
// 如果不需要Reducer，设置为0
job.setNumReduceTasks(0);
// 设置输入输出路径
FileInputFormat.addInputPath(job, new Path("/path/to/input/csv"));
FileOutputFormat.setOutputPath(job, new Path("/path/to/output"));

// 添加作业状态监听
job.addJobListener(new JobListener() {
    @Override
    public void jobSubmitted(JobContext jobContext) {}

    @Override
    public void jobCompleted(JobContext jobContext, JobStatus jobStatus) {
        if (!jobStatus.isSuccessful()) {
            // 作业失败，主动删除输出目录
            try {
                Path outputPath = FileOutputFormat.getOutputPath(job);
                FileSystem fs = FileSystem.get(conf);
                if (fs.exists(outputPath)) {
                    fs.delete(outputPath, true);
                    System.err.println("作业失败，已清理输出目录：" + outputPath);
                }
            } catch (IOException e) {
                System.err.println("清理输出目录失败：" + e.getMessage());
            }
        }
    }

    @Override
    public void jobFailed(JobContext jobContext, JobStatus jobStatus) {
        System.err.println("作业因分片处理失败终止，状态：" + jobStatus.getState());
    }
});

// 提交作业并等待完成
boolean jobSuccess = job.waitForCompletion(true);
System.exit(jobSuccess ? 0 : 1);

• 不要在Mapper中捕获异常后继续运行，必须让异常向上抛出，才能触发Hadoop的作业失败逻辑；
• 如果你的作业有Reducer，同样要配置mapreduce.reduce.maxattempts为1，确保Reducer失败也立即终止；
• 确保Hadoop集群的权限配置允许作业进程删除输出目录，否则兜底清理会失败。

内容的提问来源于stack exchange，提问作者Thomas