1. 数据清洗在Hadoop体系中的定位与整体设计思路1.1 为什么数据清洗是Hadoop项目的第一道生死关做过离线数仓的人都有一个共识数据清洗ETL做得好不好直接决定了后面所有分析任务的天花板。我见过太多项目前期为了赶进度把原始日志直接往Hive表里一灌就开始跑报表结果到了业务方对数据的时候各种脏数据冒出来——字段错位、时间格式五花八门、关键字段为空、重复记录成堆最后不得不返工重做整条链路代价是前期投入的好几倍。数据清洗的本质是把原始数据Raw Data加工成可信、规整、可分析的数据。在Hadoop生态里这件事通常发生在数据从原始层比如ODS层流向明细层DWD层的过程中。它要解决的核心问题就三类格式统一、脏数据剔除、字段补全与标准化。为什么一定要放在Hadoop上做因为数据量摆在那里。单机用Python的pandas处理几个G的数据还行一旦到了TB级别内存直接爆掉。Hadoop的MapReduce或者Spark on YARN天生就是干这种大批量、可并行、对延迟不敏感的活儿的。你可以把它理解成一条工业流水线原始数据是矿石MapReduce是破碎机和筛选机最后出来的才是能进高炉的精选料。这个案例适合谁看如果你是刚接触大数据、正在做第一个ETL项目的开发者或者你已经在用Hive但清洗逻辑总是写得乱七八糟那这篇内容就是给你准备的。我会从整体设计讲到具体代码再到踩过的坑尽量把每一步的“为什么”说清楚。1.2 整体架构选型MapReduce还是SparkHive还是自己写在动手之前第一个要做的决策是技术选型。标题里说的是Hadoop案例那核心肯定是MapReduce但实际生产中我们往往有更多选择。我把常见的几种方案列出来对比一下方案适用场景优势劣势原生MapReduce数据量极大、逻辑简单、集群资源紧张稳定、资源占用可控、不依赖额外组件开发效率低、代码冗长、调试麻烦Hive SQL清洗逻辑能用SQL表达、团队SQL能力强开发快、易维护、上手门槛低复杂逻辑难表达、调优依赖经验Spark需要迭代计算、逻辑复杂、对速度有要求快、API丰富、支持多种语言需要额外维护Spark集群、内存调优复杂Pig半结构化数据、探索性清洗脚本简洁、适合快速试验社区活跃度下降、生产用得少这个案例我选择原生MapReduce来讲解原因有两个一是标题明确是Hadoop案例MapReduce是Hadoop的灵魂理解了它你再看Spark和Hive的底层执行逻辑会通透很多二是很多公司的老系统里核心清洗任务仍然是MapReduce写的维护和优化这些任务是真真切切的工作需求。当然实际项目中我不会建议你所有清洗都用MapReduce硬写。合理的做法是分层简单的字段映射、过滤用Hive SQL搞定复杂的、需要自定义解析逻辑的比如解析不规则日志、做多表关联补全用MapReduce或Spark。这个案例我们聚焦在最典型的一类清洗任务上日志类数据的解析、过滤和标准化。1.3 清洗流程的通用骨架设计不管用什么技术一个完整的ETL清洗流程基本都包含这几个阶段数据接入从原始存储HDFS上的日志文件、Kafka消息等读取数据。解析与校验把原始记录解析成结构化字段校验字段数量、类型、格式是否合法。清洗与转换剔除脏数据、统一格式、补全缺失字段、去重。输出与落地把清洗后的数据写到目标存储HDFS、Hive表、HBase等。在MapReduce里这个流程天然映射到Mapper和Reducer两个阶段。Mapper负责解析、校验、初步过滤Reducer负责去重、聚合、最终输出。如果清洗逻辑不涉及聚合其实可以只用Mapper把Reducer数量设为0这样效率更高。这个判断很关键很多人不管三七二十一就套完整的MapReduce模板结果白白多了一次shuffle性能差一大截。提示是否需要Reducer判断标准是“清洗后是否需要按key聚合或去重”。如果只是逐条转换坚决不用Reducer。2. 核心细节解析脏数据的类型与清洗策略2.1 常见脏数据分类与识别方法在写代码之前得先搞清楚我们要对付的敌人长什么样。根据我这些年处理过的日志数据脏数据基本逃不出下面这几类字段缺失一条记录本该有12个字段结果只有9个后面全空了。常见于日志采集不完整或分隔符异常。字段错位分隔符在某个字段的值里也出现了导致split之后字段数量对不上后面的字段全部偏移。格式不统一时间字段有的用“yyyy-MM-dd HH:mm:ss”有的用时间戳有的用“yyyy/MM/dd”。非法值比如年龄字段出现负数、IP字段出现“unknown”、金额字段出现空字符串。重复记录同一条日志因为采集端重试被写入了多次。编码问题中文乱码、特殊字符导致解析失败。识别这些问题的办法我通常分两步走先抽样看再统计看。抽样就是拿几百条原始数据肉眼过一遍心里有个数统计就是用MapReduce跑一个简单的计数任务统计字段数量分布、空值比例、重复率等指标。这一步千万别省我吃过亏——凭感觉写清洗规则结果上线后发现漏了一种脏数据又得重新跑一遍全量。2.2 清洗策略的取舍严格模式还是宽松模式清洗策略的选择本质上是在数据质量和数据量之间做权衡。这里有两种典型模式严格模式只要有一条字段不合法整条记录直接丢弃。优点是输出数据质量极高下游用起来放心缺点是可能丢掉大量数据尤其是原始数据质量差的时候清洗后剩不下多少。宽松模式能补的补能修的修实在不行才丢弃。比如时间格式不对就尝试多种解析方式字段缺失就用默认值填充。优点是保留数据多缺点是可能引入错误数据下游需要额外校验。我的经验是核心字段用严格模式非核心字段用宽松模式。比如用户ID、时间戳这种关键字段不合法就必须丢弃因为下游关联全靠它而像“设备型号”这种字段缺失了可以填“unknown”不影响主流程。具体到这个案例我设定的规则是字段数量必须等于预期数量否则丢弃防止错位。时间字段必须能解析成标准格式否则丢弃。用户ID不能为空且必须是数字否则丢弃。其他字段如果为空用默认值填充。这套规则不是拍脑袋定的是根据前面统计阶段的数据特征来的。你在实际项目里也要这样先看数据再定规则。2.3 字段标准化与格式统一的实操要点字段标准化是清洗里最琐碎但也最重要的部分。我挑几个高频场景说一下时间格式统一原始日志里的时间可能是时间戳如1698765432、可能是“2024-11-05 10:30:00”、也可能是“05/Nov/2024:10:30:00 0800”。统一的做法是写一个解析函数按优先级依次尝试各种格式解析成功就转成标准格式我一般用“yyyy-MM-dd HH:mm:ss”。这里要注意时区问题如果原始数据带时区一定要统一转换到同一个时区否则跨天统计会出错。字符串清洗去除首尾空格、把连续空格替换成单个、把特殊字符如\t、\n替换掉。这些操作看起来简单但不做的话后面join的时候“北京”和“北京 ”会被当成两个不同的值。枚举值归一比如性别字段有“男”“M”“male”“1”多种写法需要统一成一种。做法是维护一个映射表在代码里做转换。数值字段处理把字符串转成数字类型转换失败的置为默认值或丢弃。注意精度问题金额字段建议用BigDecimal而不是double。这些操作在MapReduce的Mapper里逐条处理即可逻辑不复杂但一定要写成独立的工具方法方便复用和单元测试。我见过太多人把所有逻辑堆在map方法里几百行代码改一个地方牵一发动全身。3. 实操过程从原始日志到清洗后数据的完整实现3.1 原始数据格式与目标格式定义先明确输入和输出。假设我们有一批电商用户行为日志原始格式是这样的字段用“|”分隔1698765432|100234|click|product_888|北京 |2024-11-05 10:30:00|192.168.1.1|extra_info_here字段依次是时间戳、用户ID、行为类型、商品ID、城市、事件时间、IP、扩展信息。目标格式清洗后用“\t”分隔写到HDFS2024-11-05 10:30:00 100234 click product_888 北京 192.168.1.1清洗规则字段数量必须为8否则丢弃。时间戳和事件时间都能解析优先用事件时间解析失败则用时间戳转换。用户ID必须为纯数字否则丢弃。城市字段去空格为空则填“未知”。行为类型只保留click、view、buy三种其他丢弃。去掉扩展信息字段不需要。3.2 Mapper阶段的解析与过滤实现Mapper的核心工作是读取一行split校验转换输出。下面是我实际用的代码骨架Javapublic class LogCleanMapper extends MapperLongWritable, Text, Text, NullWritable { private SimpleDateFormat sdf new SimpleDateFormat(yyyy-MM-dd HH:mm:ss); private SetString validActions new HashSet(Arrays.asList(click, view, buy)); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString().trim(); if (line.isEmpty()) { return; // 空行直接跳过 } String[] fields line.split(\\|, -1); // -1保证空字段也保留 if (fields.length ! 8) { context.getCounter(CLEAN, FIELD_COUNT_ERROR).increment(1); return; } String userId fields[1].trim(); if (!userId.matches(\\d)) { context.getCounter(CLEAN, USER_ID_ERROR).increment(1); return; } String action fields[2].trim().toLowerCase(); if (!validActions.contains(action)) { context.getCounter(CLEAN, ACTION_ERROR).increment(1); return; } String eventTime parseTime(fields[5], fields[0]); if (eventTime null) { context.getCounter(CLEAN, TIME_ERROR).increment(1); return; } String city fields[4].trim(); if (city.isEmpty()) { city 未知; } String ip fields[6].trim(); String productId fields[3].trim(); String output String.join(\t, eventTime, userId, action, productId, city, ip); context.write(new Text(output), NullWritable.get()); } private String parseTime(String eventTime, String timestamp) { // 先尝试解析事件时间 try { Date date sdf.parse(eventTime.trim()); return sdf.format(date); } catch (ParseException e) { // 失败则用时间戳 } try { long ts Long.parseLong(timestamp.trim()); return sdf.format(new Date(ts * 1000)); } catch (NumberFormatException e) { return null; } } }这段代码有几个细节值得说split用“\|”并加-1参数这是为了防止字段末尾为空时被丢弃。比如“a|b||”用普通split会得到2个元素加-1才会得到4个。这个坑我踩过导致字段数量校验一直误判。计数器Counter的使用每丢弃一条数据就对应一个计数器加一。任务跑完后在日志里能看到各类错误的数量。这是排查数据质量问题的利器比你自己写日志方便得多。时间解析的优先级先试事件时间失败再用时间戳。为什么因为事件时间是业务时间更准确时间戳是采集时间可能有延迟。这个优先级要根据业务定不是固定的。输出用Text而不是自定义Writable如果字段少、格式简单直接用拼接字符串最省事。字段多或者需要类型安全时才定义自定义Writable。别为了“规范”过度设计。3.3 是否需要Reducer去重与输出的决策前面说了如果只是逐条转换不需要Reducer。但这个案例里有一个需求是去重——同一条日志可能被采集多次。去重就需要按某个key聚合这时候Reducer就派上用场了。去重的key怎么选我一般用“用户ID行为类型商品ID事件时间”组合成一个唯一标识。在Mapper里输出key为这个标识value为清洗后的整行Reducer里对每个key只输出第一条。public class DedupReducer extends ReducerText, Text, Text, NullWritable { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { // 只取第一条实现去重 Text first values.iterator().next(); context.write(first, NullWritable.get()); } }这里有个性能考量如果重复率很低去重的代价一次完整的shuffle可能不划算。我的做法是先跑一次统计任务看重复率如果重复率低于1%就不做Reducer去重而是在下游用Hive的row_number去重。如果重复率高于5%那就在MapReduce里做掉避免下游反复处理。注意Reducer去重时values的迭代顺序是不确定的所以“取第一条”得到的结果可能不稳定。如果对输出去重后的具体内容有要求需要在Mapper里对value做排序或者用更复杂的逻辑。3.4 驱动类配置与任务提交驱动类Driver负责配置Job、设置输入输出路径、指定Mapper和Reducer。关键配置如下Job job Job.getInstance(conf, log_clean_etl); job.setJarByClass(LogCleanDriver.class); job.setMapperClass(LogCleanMapper.class); job.setReducerClass(DedupReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); // 输入输出路径 FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 设置Reduce数量 job.setNumReduceTasks(10); System.exit(job.waitForCompletion(true) ? 0 : 1);Reduce数量的设置是个经验活。太少单个Reducer压力大跑得慢太多产生大量小文件下游读取效率低。我的经验公式是Reduce数量 输入数据量(GB) / 2然后根据集群资源调整。比如输入100GB设50个Reducer左右比较合适。这个案例数据量假设是50GB我设了10个偏保守实际可以再调大。输出路径必须不存在否则Job会直接报错。这是Hadoop的保护机制防止误覆盖。所以每次跑任务前要么换个路径要么先删掉旧路径。Combiner的使用如果Mapper输出数据量很大可以在Mapper端加一个Combiner做本地聚合减少网络传输。但去重场景下Combiner要慎用因为Combiner的逻辑必须和Reducer一致且满足结合律去重逻辑用Combiner容易出错。这个案例我没用Combiner。4. 常见问题与排查技巧实录4.1 数据倾斜某个Reducer跑得特别慢数据倾斜是MapReduce最常见的性能问题。表现是大部分Reducer几分钟就跑完了有一两个Reducer跑了半小时还没结束。原因通常是某个key的数据量特别大。在这个清洗案例里如果按“用户ID行为商品时间”去重一般不会倾斜因为组合key的分散度很高。但如果你的去重key选得不好比如只用“行为类型”那“click”这个key可能占了80%的数据对应Reducer就会成为瓶颈。排查方法看Job日志里各个Reducer的完成时间或者用Hadoop的Counter统计每个key的分布。解决办法有几种加盐给key加随机前缀让数据分散到多个Reducer然后再跑一个Job去掉前缀。二次聚合先局部聚合再全局聚合。换key选分散度更高的字段做key。我个人的经验是清洗阶段尽量不做聚合把去重、聚合推到下游用Hive或Spark做那些框架对倾斜的处理更成熟。MapReduce清洗就专注做逐条转换简单可靠。4.2 内存溢出Mapper或Reducer OOMOOM通常发生在几个地方Mapper里加载了太大的字典或映射表到内存。Reducer里把values全部缓存到List再处理。单条记录太大比如一行日志几百MB。解决办法字典类数据用分布式缓存DistributedCache分发或者放到每个节点本地。Reducer里不要缓存全部values能流式处理就流式处理。调整JVM堆内存mapreduce.map.java.opts-Xmx2048m根据实际情况调大。我遇到过一次OOM原因是Mapper里用了一个静态的HashMap做城市映射数据量有几十万条每个Mapper实例都加载一份直接把内存撑爆了。后来改成从HDFS读取映射文件用的时候查一次问题解决。4.3 输出小文件过多如果Reduce数量设得太多或者输入数据本身有很多小文件输出目录里会产生大量小文件。小文件对HDFS不友好占用NameNode内存对下游任务也不友好每个小文件启动一个Map任务。解决办法输入阶段用CombineFileInputFormat合并小文件。输出阶段控制Reduce数量或者跑一个合并任务。如果输出要进Hive可以在Hive侧用concatenate命令合并。4.4 常见问题速查表问题现象可能原因排查方法解决方案任务卡在Map 99%某个Map任务数据量大或数据倾斜看日志中哪个Map慢加Combiner、调大split sizeReducer OOMvalues缓存过多看异常堆栈流式处理、调大堆内存输出文件为空过滤条件太严或输入路径错看Counter统计放宽规则、检查输入路径中文乱码编码不一致检查文件编码统一用UTF-8时间解析全失败格式不匹配抽样看原始数据补充解析格式任务跑得比预期慢Reduce数量不合理看各Reducer耗时调整Reduce数量4.5 几个我踩过的坑和独家技巧坑一分隔符冲突。原始日志用“|”分隔但某个字段的值里也含“|”导致split后字段错位。解决办法是先用正则把字段内的分隔符转义或者改用更复杂的分隔符如“\u0001”。我现在的习惯是采集端就用不可见字符做分隔符从源头避免这个问题。坑二时间戳单位不统一。有的日志时间戳是秒级有的是毫秒级差1000倍。如果不判断转换出来的时间会差几十年。我的做法是判断时间戳长度10位按秒处理13位按毫秒处理。坑三空值判断用equals还是isEmpty。字段为null时调isEmpty会空指针必须先判null。这个低级错误我见过太多人犯包括我自己早期。技巧一用Counter做数据质量监控。每次清洗任务跑完把各类错误的Counter值记录下来做成趋势图。如果某天某个错误突然飙升说明上游数据出问题了能第一时间发现。技巧二清洗规则配置化。不要把规则硬编码在代码里而是写到配置文件或数据库里。这样业务方要改规则时不用重新编译打包改配置重启任务就行。我现在的项目里清洗规则全部放在一个JSON文件里代码只负责读取和执行。技巧三保留原始数据。清洗后的数据要存原始数据也要存。万一清洗逻辑有bug还能用原始数据重新跑一遍。我一般把原始数据放在ODS层清洗后的放DWD层两层都保留。技巧四小批量验证再全量跑。写清洗逻辑时先拿1%的数据跑一遍人工抽查输出结果。确认没问题了再跑全量。直接跑全量一旦逻辑错了浪费的时间和资源是巨大的。5. 清洗任务的性能调优与生产化建议5.1 MapReduce参数调优实战默认参数在生产环境往往不够用几个关键参数需要根据实际情况调整参数默认值建议值说明mapreduce.map.memory.mb10242048-4096Mapper容器内存mapreduce.reduce.memory.mb10242048-4096Reducer容器内存mapreduce.map.java.opts-Xmx820m-Xmx1600mMapper JVM堆mapreduce.reduce.java.opts-Xmx820m-Xmx1600mReducer JVM堆mapreduce.task.io.sort.mb100200-400排序缓冲区mapreduce.job.reduces1按数据量算Reduce数量mapreduce.input.fileinputformat.split.maxsizeLong.MAX128MB-256MB最大split调优的核心思路是让每个容器有足够内存处理数据同时不让资源浪费。内存设太小会OOM设太大则并发度上不去。我一般从默认值开始观察任务运行时的内存使用情况逐步调整。5.2 清洗任务的生产化改造一个能跑通的Demo和一个能上生产的任务差距是很大的。生产化改造要关注这几点幂等性任务重跑不能产生重复数据。做法是输出到临时目录成功后原子性地替换目标目录或者输出时带上分区重跑时覆盖对应分区。可监控任务运行状态、数据质量指标、耗时等要能实时看到。我一般用Counter日志外部监控系统三管齐下。可配置清洗规则、输入输出路径、资源参数都要能通过配置文件或命令行参数指定不能硬编码。容错任务失败要能自动重试重试次数和间隔可配置。Hadoop本身有任务级别的重试但Job级别的重试需要自己控制。数据血缘记录每条数据的来源和去向方便问题追溯。这个在大数据平台里通常由元数据管理系统来做。5.3 从MapReduce迁移到Spark或Hive的时机MapReduce不是终点。当你的清洗逻辑越来越复杂或者对时效性要求越来越高时就该考虑迁移了。几个信号清洗逻辑里出现了多表join、窗口函数等复杂操作MapReduce写起来很痛苦。任务耗时太长业务方等不及。团队里新人多MapReduce代码维护成本高。迁移到Spark或Hive后同样的逻辑代码量能减少70%以上性能也能提升几倍。但迁移不是一蹴而就的我的建议是新任务用新框架老任务逐步迁移不要一次性全改风险太大。6. 写在最后的一些个人体会数据清洗这件事技术难度其实不高难的是对业务的理解和对细节的把控。同样一份日志不同的人清洗出来的结果可能天差地别差别就在那些看似不起眼的细节里时间格式有没有统一、空值有没有处理、重复有没有去掉、字段有没有错位。我做了这么多年ETL最大的体会是清洗规则一定要和业务方对齐。你觉得某个字段为空就该丢弃但业务方可能觉得这个字段不重要丢了可惜。你觉得时间格式统一成一种就行但业务方可能要求保留原始格式。这些都要提前沟通清楚写进需求文档里不能自己拍脑袋决定。另外不要追求一次完美。清洗规则是迭代出来的先跑通主流程再逐步完善细节。我见过有人为了设计一个“完美”的清洗方案纠结了两周还没动手最后项目延期。正确的做法是先用最简规则跑起来看到输出结果后再根据实际问题调整规则。最后分享一个小技巧把每次清洗任务的输入数据量、输出数据量、丢弃数据量、各类错误数量记录下来做成一个简单的报表。时间长了你就能看出数据的规律和异常。比如某天丢弃率突然从2%涨到20%那肯定是上游出问题了能第一时间发现并处理。这个习惯让我避免了好几次线上事故。清洗是脏活累活但也是最有价值的基础工作。把这一层做扎实了后面的分析、挖掘、报表才能跑得稳、信得过。