1. 数据清洗在Hadoop体系中的定位与整体设计思路

1.1 为什么数据清洗是Hadoop项目的第一道生死关

做过离线数仓的人都有一个共识：原始数据永远比你想象的脏。日志里混着半截JSON、CSV字段里嵌着换行符、时间戳格式五花八门、用户ID一会儿是数字一会儿带字母前缀，这些情况在真实生产环境里几乎是标配。数据清洗（ETL中的T，即Transform）就是把这些"毛坯数据"加工成能进数仓、能跑指标、能喂给下游分析的"精装数据"。

在Hadoop生态里做数据清洗，核心思路和单机Python脚本完全不同。单机脚本处理几百MB数据随便写，但一旦数据量上到TB级别，你就必须考虑分而治之：MapReduce负责把清洗逻辑并行化，YARN负责资源调度，HDFS负责存储中间结果。这也是为什么很多初学者用Pandas写清洗脚本跑得挺爽，一上生产环境就崩——不是逻辑不对，是架构不对。

这个案例要解决的问题很明确：把存放在HDFS上的原始日志/业务数据，通过MapReduce作业完成字段校验、格式规整、脏数据过滤、缺失值处理，最终输出规整的、可直接加载到Hive表的分区数据。适合有一定Java或Python基础、正在从单机数据处理向分布式数据处理转型的开发者，也适合需要搭建离线数仓ETL链路的数仓工程师参考。

1.2 整体方案选型：为什么是MapReduce而不是Spark

很多人会问：现在Spark这么火，为什么还要用MapReduce做数据清洗？这个问题我在实际项目里被问过不下十次。答案其实很现实：

• 存量系统惯性：很多公司的离线清洗链路是五六年前搭的，全是MapReduce作业，迁移成本极高，稳定跑着的东西没人愿意动。
• 资源隔离清晰：MapReduce作业对内存的占用相对可控，不会像Spark那样动不动就OOM，在共享集群里更容易做资源规划。
• 逻辑简单场景够用：如果清洗逻辑只是"过滤+字段映射+格式转换"，MapReduce的Map-only作业完全够用，甚至不需要Reduce阶段，效率反而更高。

当然，如果清洗逻辑涉及多表Join、复杂窗口计算，那Spark SQL确实更合适。但作为理解Hadoop ETL的入门案例，MapReduce是绕不开的基本功。我建议的选型原则是：

本案例采用Map-only + 可选Reduce去重的混合设计，既能处理大部分清洗场景，又保留了扩展空间。

1.3 数据流转的完整链路设计

一个完整的Hadoop数据清洗链路，从数据落地到最终可用，通常经过这几个环节：

1. 原始层（ODS）：数据从业务系统或日志采集端落到HDFS，通常是按天分区的原始文件，格式可能是文本、SequenceFile或Avro。
2. 清洗层（DWD）：通过MapReduce作业读取ODS层数据，执行清洗逻辑，输出到DWD层。这一层的数据已经是规整的、字段类型明确的。
3. 汇总层（DWS）：在DWD基础上做轻度聚合，供上层报表使用。

本案例聚焦在ODS到DWD这一层，这是整个ETL链路里最脏、最累、但价值最高的环节。设计时要考虑几个关键点：输出格式用什么（TextFile还是ORC/Parquet）、分区怎么切（按天还是按小时）、脏数据往哪扔（直接丢弃还是写到侧输出流）。

提示：脏数据千万不要直接丢弃，一定要单独写到/bad_data/目录下，方便后续排查数据质量问题。这个习惯能帮你在数据对不上账的时候快速定位问题。

2. 核心细节解析与实操要点

2.1 清洗逻辑的四大核心操作

数据清洗听起来玄乎，拆开来看无非四类操作，我把它们称为"清洗四件套"：

第一件：字段校验与过滤。检查每行数据的字段数量是否符合预期，比如日志规定是12个字段，结果某行只有9个，那这行就是残缺数据，直接过滤。还要检查关键字段是否为空、是否符合正则规则（比如手机号必须是11位数字）。

第二件：格式规整。时间戳统一成yyyy-MM-dd HH:mm:ss格式，金额统一保留两位小数，字符串统一去空格和转义字符。这一步最繁琐，但必须做，否则下游Hive表里会出现各种诡异的分区。

第三件：缺失值处理。缺失值有三种处理策略：直接丢弃、填充默认值、用统计值填充（均值/中位数）。选择哪种取决于业务含义——用户ID缺失基本只能丢弃，但商品价格缺失可能可以填0或均值。

第四件：去重。同一用户同一秒可能产生多条重复日志，需要按业务主键去重。这一步通常需要Reduce阶段，利用Shuffle按Key分组后只保留一条。

2.2 Mapper阶段的代码骨架与关键参数

Mapper是整个清洗作业的核心，所有清洗逻辑都在这里执行。下面是一个典型的Mapper骨架，我用Java写，因为生产环境里Java作业占比最高：

public class CleanMapper extends Mapper<LongWritable, Text, Text, NullWritable> { private Text outputKey = new Text(); private Counter badCounter; @Override protected void setup(Context context) { badCounter = context.getCounter("CLEAN_STATS", "BAD_RECORDS"); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString().trim(); // 1. 空行过滤 if (line.isEmpty()) { badCounter.increment(1); return; } // 2. 字段分割 String[] fields = line.split("\t", -1); if (fields.length != 12) { badCounter.increment(1); return; } // 3. 关键字段校验 if (!fields[0].matches("\\d{11}")) { badCounter.increment(1); return; } // 4. 格式规整 String cleanLine = String.join("\t", fields[0], // 手机号 fields[1].trim(), // 用户名 formatTime(fields[2]), // 时间戳 fields[3].toLowerCase(), // 事件类型 // ... 其余字段处理 fields[11] ); outputKey.set(cleanLine); context.write(outputKey, NullWritable.get()); } private String formatTime(String raw) { // 时间格式转换逻辑 return raw; } }

这里有几个关键点值得展开说：

split("\t", -1)的-1参数：这个参数表示保留末尾的空字符串。如果不加，"a\tb\t"会被分割成["a", "b"]，字段数就少了一个，导致误判为脏数据。这个坑我踩过，排查了半天才发现是split的问题。

Counter的使用：context.getCounter()是MapReduce自带的计数器，用来统计脏数据数量。作业跑完后在日志里能看到CLEAN_STATS.BAD_RECORDS的值，这是监控数据质量最直接的手段。比事后写脚本统计高效得多。

NullWritable的妙用：Map-only作业不需要输出Value，用NullWritable可以避免序列化开销。别小看这点优化，在TB级数据上能省不少IO。

2.3 输出格式与压缩的选择

输出格式的选择直接影响下游读取效率。常见选项对比：

本案例输出用TextFile + Gzip压缩，原因是清洗后的数据通常还要经过二次加工，TextFile兼容性最好，Gzip能把存储成本降下来。配置压缩只需在Driver里加两行：

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

注意：Gzip压缩的文件不可分割，如果单个输出文件很大，会导致下游只有一个Map任务读取，并行度上不去。如果数据量特别大，建议用Bzip2或Snappy。

2.4 分区设计：按天分区是基本操作

清洗后的数据几乎一定要按时间分区，这是数仓的基本规范。在MapReduce里实现分区有两种方式：

方式一：输出路径带分区。在Driver里根据业务日期动态设置输出路径，比如/dwd/user_log/dt=2024-01-15/。这种方式简单直接，适合按天调度的作业。

方式二：自定义Partitioner。如果一次作业要处理多天数据，可以自定义Partitioner按日期把数据分到不同Reduce，每个Reduce输出到不同分区。这种方式复杂但灵活。

我个人的经验是：按天调度的作业用方式一，补数场景用方式二。补数时经常要一次跑一个月的数据，用方式二能避免启动30个作业。

3. 实操过程与核心环节实现

3.1 环境准备与依赖配置

先把环境搭起来。假设你已经有可用的Hadoop集群（伪分布式或完全分布式都行），下面是Maven依赖：

<dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.4</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>3.3.4</version> </dependency> </dependencies>

打包时用maven-shade-plugin把依赖打进去，避免在集群上找不到类。这个细节很多新手会忽略，结果提交作业时报ClassNotFoundException。

<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.4.1</version> <executions> <execution> <phase>package</phase> <goals><goal>shade</goal></goals> </execution> </executions> </plugin>

3.2 完整清洗作业的Driver实现

Driver是作业的入口，负责配置Job参数。下面是一个可直接参考的Driver：

public class CleanDriver { public static void main(String[] args) throws Exception { if (args.length != 3) { System.err.println("Usage: CleanDriver <input> <output> <date>"); System.exit(1); } Configuration conf = new Configuration(); // 开启压缩 conf.setBoolean("mapreduce.map.output.compress", true); conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec"); Job job = Job.getInstance(conf, "UserLogClean_" + args[2]); job.setJarByClass(CleanDriver.class); // 输入输出路径 FileInputFormat.addInputPath(job, new Path(args[0])); Path outputPath = new Path(args[1] + "/dt=" + args[2]); FileOutputFormat.setOutputPath(job, outputPath); // Mapper配置 job.setMapperClass(CleanMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); // Map-only作业，设置Reduce数量为0 job.setNumReduceTasks(0); // 输出格式 job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); FileOutputFormat.setCompressOutput(job, true); FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); // 提交作业 boolean success = job.waitForCompletion(true); // 打印清洗统计 long badRecords = job.getCounters() .findCounter("CLEAN_STATS", "BAD_RECORDS").getValue(); System.out.println("脏数据条数: " + badRecords); System.exit(success ? 0 : 1); } }

这个Driver里有几个设计决策值得说明：

为什么用Snappy做Map输出压缩：Map输出会写到本地磁盘，Snappy压缩解压速度快，CPU开销小，适合中间数据。而最终输出用Gzip，因为要长期存储，压缩比更重要。

为什么输出路径带dt=：这是Hive分区表的命名规范，下游建表时用PARTITIONED BY (dt STRING)就能直接识别。如果命名不规范，还得额外写脚本重命名，纯属给自己找麻烦。

为什么打印脏数据条数：这是数据质量监控的第一道防线。如果某天脏数据突然暴涨，说明上游数据源出问题了，得赶紧排查。

3.3 提交作业与参数调优

打包后提交作业：

hadoop jar clean-job.jar com.example.CleanDriver \ /ods/user_log/2024-01-15 \ /dwd/user_log \ 2024-01-15

提交后要关注几个关键指标：

• Map任务数量：由输入文件的分片数决定。如果输入是100个128MB的块，就会有100个Map任务。任务太多会增加调度开销，太少则并行度不够。
• 每个Map处理的数据量：理想情况下每个Map处理128MB~256MB。如果某个Map处理特别慢，可能是数据倾斜。
• Counter统计：重点看BAD_RECORDS和MAP_OUTPUT_RECORDS，前者反映数据质量，后者反映清洗后的数据量。

调优参数参考：

3.4 清洗效果验证

作业跑完后，必须验证清洗效果。我通常用三步验证法：

第一步：抽样检查。用hdfs dfs -cat随机看几行输出，确认字段数、格式、编码都正常。

hdfs dfs -cat /dwd/user_log/dt=2024-01-15/part-m-00000 | head -20

第二步：数量核对。对比输入和输出的记录数，差值应该等于脏数据条数。

# 输入行数 hdfs dfs -cat /ods/user_log/2024-01-15/* | wc -l # 输出行数 hdfs dfs -cat /dwd/user_log/dt=2024-01-15/* | wc -l

第三步：建Hive表查询。把清洗后的数据挂到Hive外部表上，跑几个简单查询验证。

CREATE EXTERNAL TABLE dwd.user_log ( phone STRING, user_name STRING, event_time STRING, event_type STRING ) PARTITIONED BY (dt STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/dwd/user_log'; ALTER TABLE dwd.user_log ADD PARTITION (dt='2024-01-15'); SELECT COUNT(*) FROM dwd.user_log WHERE dt='2024-01-15';

4. 常见问题与排查技巧实录

4.1 数据倾斜：清洗作业最头疼的问题

数据倾斜在清洗作业里表现为：大部分Map/Reduce任务几分钟就跑完了，但有一两个任务跑了半小时还没结束。原因通常是某个Key的数据量远超其他Key。

在Map-only作业里，数据倾斜表现为某个Map任务处理的数据量特别大。解决办法是调整分片大小：

// 把分片大小从128MB调到64MB，让数据更均匀 FileInputFormat.setMaxInputSplitSize(job, 64 * 1024 * 1024);

在带Reduce的作业里，数据倾斜更常见。比如按用户ID去重时，某个大V用户产生了百万条日志，全分到一个Reduce上。解决办法是加盐打散：给Key加随机前缀，先局部聚合，再去掉前缀全局聚合。

4.2 中文乱码：编码问题排查

清洗日志时经常遇到中文乱码，根源是编码不一致。Hadoop默认用UTF-8，但有些业务系统输出的是GBK。排查步骤：

1. 用file -i命令查看原始文件编码。
2. 在Mapper里显式指定编码读取：new String(value.getBytes(), 0, value.getLength(), "GBK")。
3. 输出统一用UTF-8。

提示：如果乱码只在部分行出现，可能是数据里混了不可见字符，用cat -A能看到^M之类的符号，需要在清洗时用正则替换掉。

4.3 小文件问题：输出目录一堆小文件

Map-only作业的输出文件数等于Map任务数。如果输入有1000个小文件，输出就是1000个小文件，这对HDFS和下游Hive都是灾难。解决办法：

• 输入端合并：用CombineFileInputFormat把多个小文件合并成一个分片。
• 输出端合并：作业跑完后用hdfs dfs -getmerge合并，或者再跑一个合并作业。

我通常的做法是在调度系统里加一个后置步骤，自动合并小文件。这个习惯能省掉很多下游的麻烦。

4.4 常见问题速查表

4.5 独家避坑经验

坑一：不要在生产环境用TextInputFormat读大文件。它按行读，遇到超大文件（比如10GB的单文件）会切成很多分片，每个分片都要重新打开文件，IO开销大。改用CombineFileInputFormat能显著提升效率。

坑二：清洗逻辑一定要幂等。同一个作业跑两遍，结果应该一样。如果清洗逻辑里有随机数、时间戳，会导致每次结果不同，补数时对不上账。我见过有团队在清洗时加了System.currentTimeMillis()做字段，结果补数时数据全乱了。

坑三：脏数据要留痕。不要直接return丢弃，用MultipleOutputs把脏数据写到单独目录，带上丢弃原因。这样数据对不上时能快速定位。

// 用MultipleOutputs写脏数据 MultipleOutputs.addNamedOutput(job, "bad", TextOutputFormat.class, Text.class, NullWritable.class);

坑四：作业参数不要硬编码。输入路径、输出路径、日期都通过命令行参数传入，方便调度系统调用。硬编码的作业在补数时只能改代码重新打包，效率极低。

坑五：监控Counter变化趋势。单看一天的脏数据条数没意义，要看趋势。如果某天突然翻倍，说明上游出问题了。我通常会把Counter值写到监控系统，设置阈值告警。

5. 清洗作业的扩展与优化方向

5.1 从MapReduce迁移到Spark SQL的时机

当清洗逻辑越来越复杂，MapReduce代码会变得难以维护。这时候可以考虑迁移到Spark SQL。判断时机：

• 清洗逻辑涉及3张以上表的Join
• 需要窗口函数做复杂计算
• 团队里SQL能力强于Java
• 对执行速度有更高要求

迁移时不用一次性重写，可以先用Spark SQL实现新逻辑，老作业继续跑，等新作业稳定后再下线老的。这种渐进式迁移风险最低。

5.2 数据质量监控体系的搭建

清洗作业只是数据质量保障的一环。完整的监控体系应该包括：

• 事前：上游数据接入时做Schema校验，字段类型不对直接拒绝。
• 事中：清洗作业运行时监控Counter，脏数据比例超过阈值告警。
• 事后：定期跑数据质量报告，统计各表的空值率、重复率、分布异常。

这套体系搭起来后，数据问题的发现时间能从"下游报表出错"提前到"清洗作业跑完"，排查成本降低一个数量级。

5.3 增量清洗与全量清洗的选择

按天调度的作业通常是增量清洗，只处理当天的数据。但有些场景需要全量清洗，比如：

• 清洗规则变了，历史数据要重新洗
• 上游数据修正了，需要重跑
• 新加了一个字段，历史数据要补

全量清洗的挑战是数据量大、耗时长。我的经验是：按分区并行跑，把历史数据按天拆成多个作业并行执行，用调度系统控制并发度。这样能把几天的清洗时间压缩到几小时。

5.4 清洗规则的版本管理

清洗规则会随着业务变化不断调整，必须做版本管理。我的做法是：

• 每次规则变更都打Tag，记录变更原因和影响范围
• 清洗后的数据带上规则版本号，方便追溯
• 保留旧版本规则，支持历史数据重跑

这个习惯在数据审计时特别有用。当业务方质疑某个指标时，能快速定位是哪个版本的清洗规则导致的。

6. 我在实际项目中的几点体会

做数据清洗这些年，最大的体会是：清洗逻辑本身不难，难的是让清洗结果稳定可靠。同样一段代码，在测试环境跑得好好的，上生产就出问题，原因往往是数据分布变了、数据量涨了、上游格式改了。

所以我现在写清洗作业，一定会做三件事：第一，所有过滤条件都记Counter，跑完能说清楚每条数据为什么被过滤；第二，脏数据单独落盘，保留原始内容和丢弃原因；第三，清洗规则写成配置文件，改规则不用改代码。

还有一点，不要追求一次清洗到位。很多新手想把所有清洗逻辑塞进一个作业，结果作业又大又慢又难维护。正确的做法是分层清洗：第一层做基础校验和格式规整，第二层做业务规则过滤，第三层做聚合去重。每层职责单一，出问题容易定位，也方便复用。

最后分享一个小技巧：清洗作业的输出目录里，除了数据文件，再放一个_SUCCESS文件和一个_STATS.json文件，记录本次清洗的统计信息（输入条数、输出条数、脏数据条数、规则版本）。下游作业读取前先检查_SUCCESS，能避免读到不完整的数据。这个习惯是从一次惨痛的生产事故里学来的——当时上游作业失败但输出目录有部分数据，下游直接读了，导致报表数据少了一半，排查了一整天才找到原因。