数据清洗

MapReduce 数据清洗：脏数据进来，干净数据出去，连 Reduce 都省了

数据清洗是数据分析的第一步，也是最重要的一步——“垃圾进，垃圾出”，数据不干净，后面的分析全是错的。

清洗就是三件事：

1. 过滤掉坏数据（格式错、字段缺、值不对）
2. 标准化好数据（日期统一、大小写统一）
3. 去掉重复的

在 MapReduce 里做数据清洗有个特别的优势：只需要 Mapper，不需要 Reducer。

因为清洗是逐条处理——每条数据独立判断、独立转换，不需要聚合。省掉 Reduce 阶段，就省掉了 Shuffle，数据直接从 Mapper 写到 HDFS，快得多。

为什么数据清洗不需要 Reduce？

MapReduce 的标准流程是 Map → Shuffle → Reduce。但清洗场景不需要分组、不需要聚合。

每条数据独立处理，过滤掉坏的，转换好的，然后直接输出。

flowchart TD  
    A[原始数据HDFS] --> B[InputFormat 分片]  
    B --> C[Mapper 读取并清洗数据]  
    C -->|过滤/转换| D[符合条件的数据输出]  
    D --> E[OutputFormat 写入 HDFS]

所以配置就是一句话：

job.setNumReduceTasks(0);

Reduce 数量设成 0，Shuffle 就不会发生，数据直接走 Mapper → OutputFormat → HDFS。

省掉了排序、省掉了网络传输、省掉了磁盘溢写——对于只需要”过滤+转换”的作业，这是最大的性能优化。

一个例子：清洗 Nginx 日志

原始日志：

192.168.1.1 - [2023-10-01 12:00:00] "GET /index.html" 200 1024
192.168.1.2 - [2023-10-01 12:01:00] "GET /error.html" 404 512
192.168.1.3 - [2023-10-01 12:02:00] "POST /submit" 200 2048

清洗目标：

• 只保留状态码 200 的日志（过滤掉 404、500 等错误日志）
• 检查格式是否正确（字段数够不够）

Mapper 代码：

public class LogCleanMapper extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) {
        String line = value.toString();
        
        // 1. 过滤空行
        if (line == null || line.trim().isEmpty()) return;
        
        // 2. 检查字段数
        String[] fields = line.split(" ");
        if (fields.length < 9) return;  // 格式错误，扔掉
        
        // 3. 提取状态码（第 8 个字段）
        String status = fields[8];
        
        // 4. 只保留状态码 200
        if ("200".equals(status)) {
            context.write(null, new Text(line));
        }
        // 其他状态码直接丢弃
    }
}

Driver：

job.setMapperClass(LogCleanMapper.class);
job.setNumReduceTasks(0);  // 核心：省掉 Reduce
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

进阶：把脏数据也记下来

有时候光扔掉不够——你需要知道扔掉了多少、为什么扔。

用 MultipleOutputs 把干净数据和脏数据分开放：

public class LogCleanMapper extends Mapper<...> {
    private MultipleOutputs<Text, Text> mos;
    
    @Override
    protected void setup(Context context) {
        mos = new MultipleOutputs<>(context);
    }
    
    @Override
    protected void map(LongWritable key, Text value, Context context) {
        String line = value.toString();
        String[] fields = line.split(" ");
        
        if (fields.length < 9) {
            // 格式错误 → 记到 invalid 目录
            mos.write("invalid", null, new Text(line), "invalid/");
            return;
        }
        
        String status = fields[8];
        if ("200".equals(status)) {
            // 有效数据 → 记到 valid 目录
            mos.write("valid", null, new Text(line), "valid/");
        } else {
            // 状态码异常 → 记到 invalid
            mos.write("invalid", null, new Text(line), "invalid/");
        }
    }
    
    @Override
    protected void cleanup(Context context) {
        mos.close();
    }
}

Driver 里注册两个输出：

MultipleOutputs.addNamedOutput(job, "valid", TextOutputFormat.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(job, "invalid", TextOutputFormat.class, Text.class, Text.class);

这样跑完你会得到两个目录：valid/ 是干净数据，invalid/ 是脏数据。可以回头分析脏数据的原因。

数据转换：把格式统一

除了过滤，数据清洗还经常做格式标准化。

比如日期格式不统一，有的是 2023-10-01，有的是 01/10/2023。清洗的时候统一成一种。

// 假设输入格式：user_id,action,date
// 目标：把日期统一成 yyyy-MM-dd
protected void map(LongWritable key, Text value, Context context) {
    String[] fields = value.toString().split(",");
    if (fields.length < 3) return;
    
    String userId = fields[0];
    String action = fields[1];
    String date = fields[2];
    
    // 转换日期格式
    String normalizedDate = normalizeDate(date);  // 自定义方法
    if (normalizedDate == null) return;  // 日期格式不合法，丢弃
    
    String cleaned = userId + "," + action + "," + normalizedDate;
    context.write(null, new Text(cleaned));
}

数据转换的常见类型：

类型	示例
日期标准化	01/10/2023 → 2023-10-01
大小写统一	"Active" / "active" → "active"
去除首尾空格	" hello " → "hello"
缺失值补全	空字段 → "unknown"
格式校验	邮箱、手机号正则校验

性能优化

1. 省掉 Reduce 是最重要的优化

setNumReduceTasks(0) 省掉了 Shuffle，这是数据清洗能跑得快的核心原因。

2. 开启压缩

清洗后的数据如果很大，开启输出压缩：

FileOutputFormat.setCompressOutput(job, true);
FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);

3. 调整 Mapper 内存

清洗作业是纯 Mapper 作业，内存主要给 Map 用：

<property>
  <name>mapreduce.map.memory.mb</name>
  <value>2048</value>
</property>

4. 使用 CombineTextInputFormat 合并小文件

如果输入是大量小文件，用 CombineTextInputFormat 减少 Map 任务数：

job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job, 128 * 1024 * 1024);