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 MapReduce无需Shuffle速度快需要按Key聚合去重MapReduce带Reduce利用Shuffle天然分组多表关联复杂计算Spark SQL / Hive SQL表达能力强开发效率高实时性要求高Flink / Spark Streaming微批或流式处理本案例采用Map-only 可选Reduce去重的混合设计既能处理大部分清洗场景又保留了扩展空间。1.3 数据流转的完整链路设计一个完整的Hadoop数据清洗链路从数据落地到最终可用通常经过这几个环节原始层ODS数据从业务系统或日志采集端落到HDFS通常是按天分区的原始文件格式可能是文本、SequenceFile或Avro。清洗层DWD通过MapReduce作业读取ODS层数据执行清洗逻辑输出到DWD层。这一层的数据已经是规整的、字段类型明确的。汇总层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 MapperLongWritable, 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可读性好通用无Schema压缩比低中间结果、调试SequenceFile支持二进制可压缩只支持KV结构中间层数据ORC列式存储压缩比高写入稍慢数仓DWD/DWS层Parquet列式存储生态好小文件问题分析型场景本案例输出用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/dt2024-01-15/。这种方式简单直接适合按天调度的作业。方式二自定义Partitioner。如果一次作业要处理多天数据可以自定义Partitioner按日期把数据分到不同Reduce每个Reduce输出到不同分区。这种方式复杂但灵活。我个人的经验是按天调度的作业用方式一补数场景用方式二。补数时经常要一次跑一个月的数据用方式二能避免启动30个作业。3. 实操过程与核心环节实现3.1 环境准备与依赖配置先把环境搭起来。假设你已经有可用的Hadoop集群伪分布式或完全分布式都行下面是Maven依赖dependencies dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.4/version /dependency dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-common/artifactId version3.3.4/version /dependency /dependencies打包时用maven-shade-plugin把依赖打进去避免在集群上找不到类。这个细节很多新手会忽略结果提交作业时报ClassNotFoundException。plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.4.1/version executions execution phasepackage/phase goalsgoalshade/goal/goals /execution /executions /plugin3.2 完整清洗作业的Driver实现Driver是作业的入口负责配置Job参数。下面是一个可直接参考的Driverpublic 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前者反映数据质量后者反映清洗后的数据量。调优参数参考参数默认值建议值说明mapreduce.map.memory.mb10242048清洗逻辑复杂时调大mapreduce.task.io.sort.mb100200Map输出缓冲mapreduce.map.output.compressfalsetrue开启压缩mapreduce.job.reduces10Map-only作业设为03.4 清洗效果验证作业跑完后必须验证清洗效果。我通常用三步验证法第一步抽样检查。用hdfs dfs -cat随机看几行输出确认字段数、格式、编码都正常。hdfs dfs -cat /dwd/user_log/dt2024-01-15/part-m-00000 | head -20第二步数量核对。对比输入和输出的记录数差值应该等于脏数据条数。# 输入行数 hdfs dfs -cat /ods/user_log/2024-01-15/* | wc -l # 输出行数 hdfs dfs -cat /dwd/user_log/dt2024-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 (dt2024-01-15); SELECT COUNT(*) FROM dwd.user_log WHERE dt2024-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。排查步骤用file -i命令查看原始文件编码。在Mapper里显式指定编码读取new String(value.getBytes(), 0, value.getLength(), GBK)。输出统一用UTF-8。提示如果乱码只在部分行出现可能是数据里混了不可见字符用cat -A能看到^M之类的符号需要在清洗时用正则替换掉。4.3 小文件问题输出目录一堆小文件Map-only作业的输出文件数等于Map任务数。如果输入有1000个小文件输出就是1000个小文件这对HDFS和下游Hive都是灾难。解决办法输入端合并用CombineFileInputFormat把多个小文件合并成一个分片。输出端合并作业跑完后用hdfs dfs -getmerge合并或者再跑一个合并作业。我通常的做法是在调度系统里加一个后置步骤自动合并小文件。这个习惯能省掉很多下游的麻烦。4.4 常见问题速查表问题现象可能原因排查方法解决方案作业卡在Map 0%输入路径为空检查HDFS路径确认路径和权限ClassNotFoundException依赖没打进jar查看jar内容用shade插件打包输出文件为空过滤条件太严看Counter统计放宽过滤规则中文乱码编码不一致file -i查编码显式指定编码数据倾斜Key分布不均看任务耗时加盐打散小文件过多Map任务太多数输出文件数合并小文件内存溢出Map内存不足看GC日志调大map内存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能避免读到不完整的数据。这个习惯是从一次惨痛的生产事故里学来的——当时上游作业失败但输出目录有部分数据下游直接读了导致报表数据少了一半排查了一整天才找到原因。
