1. 数据清洗在Hadoop里到底在洗什么很多人第一次听到“Hadoop数据清洗”脑子里浮现的画面是拿个刷子把硬盘擦干净。实际上ETL里的清洗环节干的是把原始数据中那些“不规矩”的东西挑出来、修好或者扔掉。原始数据可能来自日志文件、业务数据库导出、第三方接口回传格式五花八门字段缺失、编码混乱、时间戳格式不统一、数值字段混进了字母这些都是常态。我在实际项目里见过最离谱的一份日志数据同一个用户ID字段有的记录是纯数字有的前面带了“user_”前缀还有的干脆是空字符串。如果直接拿这种数据去做分析结果一定是灾难性的。所以数据清洗的核心目标就三个格式统一、脏数据剔除、字段标准化。在Hadoop生态里这件事通常用MapReduce或者Spark来完成因为数据量摆在那里单机脚本跑不动。这篇文章适合谁看如果你已经了解HDFS和MapReduce的基本概念但不知道一个完整的ETL流程该怎么落地那这篇内容就是写给你的。我会从原始数据的常见问题讲起一步步拆解清洗逻辑的设计思路给出可运行的代码框架最后分享几个我在实际跑任务时踩过的坑。全程不堆砌理论只讲能直接拿去用的东西。注意数据清洗没有“万能模板”不同数据源的清洗规则差异极大。本文给出的方案是通用骨架你需要根据自己数据的实际情况调整过滤条件和转换逻辑。2. 原始数据里最常见的五类脏数据在动手写代码之前先搞清楚你要对付的敌人长什么样。根据我处理过的日志、CSV导出、JSON回传等各类数据源的经验脏数据基本可以归为以下五类。每一类的处理策略不同有的需要丢弃有的可以修复有的必须做字段映射。2.1 缺失值与空字段这是最普遍的问题。一条记录里某个字段为空或者用“NULL”“null”“-1”“unknown”等占位符表示缺失。处理方式取决于字段的重要性如果是主键或时间戳缺失通常直接丢弃整条记录如果是次要字段缺失可以填充默认值或者标记为“未知”。这里有个容易犯的错误不要用全局常量去填充所有缺失值。比如把缺失的年龄统一填成0后续做平均值计算时就会严重偏低。更合理的做法是标记为null在分析阶段再决定如何处理。2.2 格式不一致同一个字段在不同记录里格式不同典型的有时间字段有的用“2024-01-15 10:30:00”有的用“2024/01/15”有的用时间戳数字数值字段有的带千分位逗号“1,234”有的带货币符号“¥1234”布尔字段有的用“true/false”有的用“1/0”有的用“Y/N”这类问题的处理思路是先归一化再解析。比如时间字段统一转成标准格式后再输出数值字段先去掉非数字字符再转换类型。2.3 编码与字符集混乱中文数据里最常见的是UTF-8和GBK混用导致出现乱码。还有一种情况是特殊字符如换行符、制表符混入了字段值内部导致CSV解析时列错位。处理这类问题需要在读取阶段就指定正确的字符集并在清洗时对字段值做转义或替换。2.4 重复记录完全重复的记录相对好处理用去重逻辑就能搞定。麻烦的是“近似重复”——比如同一个用户短时间内提交了两次相似的表单只有时间戳差了几秒。这种需要根据业务规则定义去重键比如用“用户ID操作类型时间窗口”来判断。2.5 逻辑矛盾与越界值比如年龄字段出现了200岁订单金额出现了负数开始时间晚于结束时间。这类数据不一定能自动修复但必须被识别出来。我的做法是设置合理的值域范围超出范围的记录单独输出到一个“异常数据”目录后续人工排查。脏数据类型常见表现推荐处理策略缺失值空字符串、NULL、占位符主键缺失丢弃次要字段标记格式不一致时间/数值格式多样归一化后统一解析编码混乱乱码、列错位指定字符集转义特殊字符重复记录完全重复或近似重复定义去重键窗口去重逻辑矛盾越界值、时间倒挂值域校验异常分流3. 用MapReduce搭建清洗流程的核心思路搞清楚脏数据的类型之后接下来要设计清洗流程。在Hadoop里做ETLMapReduce是最基础的方案虽然写起来比Spark繁琐但理解它对掌握整个生态的运行机制很有帮助。一个典型的清洗任务分为三个阶段Map阶段负责逐条解析和初步过滤Combiner阶段做本地聚合可选Reduce阶段做最终的去重和汇总输出。3.1 Mapper端的设计要点Mapper的输入是原始数据的一行输出是清洗后的键值对。这里的关键决策是哪些逻辑放在Mapper哪些留给Reducer。我的经验是能在Mapper做的过滤和转换就不要拖到Reducer。原因很简单Mapper是并行执行的数据在本地就能被过滤掉减少网络传输量。比如格式校验、字段解析、空值过滤这些都应该在Mapper里完成。具体来说Mapper里要做这几件事跳过表头行如果是CSV文件按分隔符拆分字段检查字段数量是否匹配对每个字段做类型转换和格式校验过滤掉不符合基本要求的记录输出清洗后的键值对public class DataCleanMapper extends MapperLongWritable, Text, Text, Text { private Text outputKey new Text(); private Text outputValue new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString().trim(); // 跳过空行和表头 if (line.isEmpty() || line.startsWith(user_id)) { return; } String[] fields line.split(\t); // 字段数量校验 if (fields.length 5) { context.getCounter(CLEAN, FIELD_MISMATCH).increment(1); return; } String userId fields[0].trim(); String timestamp fields[1].trim(); String action fields[2].trim(); String amount fields[3].trim(); String status fields[4].trim(); // 主键不能为空 if (userId.isEmpty()) { context.getCounter(CLEAN, EMPTY_USER_ID).increment(1); return; } // 时间格式归一化 String normalizedTime normalizeTimestamp(timestamp); if (normalizedTime null) { context.getCounter(CLEAN, BAD_TIMESTAMP).increment(1); return; } // 金额字段清洗 String cleanAmount amount.replaceAll([^0-9.], ); if (cleanAmount.isEmpty()) { cleanAmount 0; } outputKey.set(userId); outputValue.set(normalizedTime \t action \t cleanAmount \t status); context.write(outputKey, outputValue); } private String normalizeTimestamp(String ts) { // 支持多种时间格式的归一化逻辑 // 实际实现需要根据数据源格式编写 return ts; } }3.2 Reducer端的去重与聚合Reducer的输入是Mapper输出的键值对相同key的记录会被分到同一个Reducer。利用这个特性可以很方便地做去重对于每个key只保留第一条记录或者按照某个规则选出一条“最优”记录。但这里有个坑如果数据量极大单个key对应的记录数可能非常多导致Reducer内存溢出。解决办法是在Mapper端先用HashSet做本地去重减少传输到Reducer的数据量。或者改用Spark的dropDuplicates底层做了更优化的处理。public class DataCleanReducer extends ReducerText, Text, Text, Text { private Text result new Text(); Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { // 去重相同key只输出第一条 IteratorText it values.iterator(); if (it.hasNext()) { result.set(it.next().toString()); context.write(key, result); } } }3.3 计数器清洗过程的“仪表盘”上面代码里用到了context.getCounter这是MapReduce里非常实用的一个功能。每次过滤掉一条记录就对应一个计数器加一。任务跑完之后你可以在日志里看到每个计数器的值清楚地知道有多少条记录因为什么原因被丢弃。我习惯在项目里设置这几类计数器EMPTY_USER_ID主键为空BAD_TIMESTAMP时间格式错误FIELD_MISMATCH字段数量不对OUT_OF_RANGE数值越界这些数据在排查问题时非常有用。比如某天发现清洗后的数据量突然少了很多一看计数器发现BAD_TIMESTAMP暴涨那就说明上游数据源的时间格式变了需要及时调整解析逻辑。4. 从MapReduce到Spark清洗逻辑的进阶写法MapReduce能跑通清洗流程但写起来确实啰嗦。一个简单的过滤逻辑要写几十行Java代码调试也不方便。如果你们的集群已经上了Spark我强烈建议用Spark SQL或者DataFrame API来做清洗代码量能减少一半以上而且可读性更好。4.1 用DataFrame API做字段级清洗Spark的DataFrame API提供了丰富的内置函数处理缺失值、格式转换、去重这些操作都有现成的方法。下面这段代码展示了如何用PySpark完成和上面MapReduce等价的清洗逻辑from pyspark.sql import SparkSession from pyspark.sql.functions import col, when, regexp_replace, to_timestamp, trim from pyspark.sql.types import DoubleType spark SparkSession.builder \ .appName(DataCleanETL) \ .getOrCreate() # 读取原始数据 raw_df spark.read.csv(hdfs:///data/raw/events/, sep\t, headerTrue, inferSchemaFalse) # 字段清洗 cleaned_df raw_df \ .filter(col(user_id).isNotNull() (trim(col(user_id)) ! )) \ .withColumn(amount, regexp_replace(col(amount), [^0-9.], ).cast(DoubleType())) \ .withColumn(amount, when(col(amount).isNull(), 0.0).otherwise(col(amount))) \ .withColumn(event_time, to_timestamp(col(timestamp), yyyy-MM-dd HH:mm:ss)) \ .filter(col(event_time).isNotNull()) \ .dropDuplicates([user_id, event_time, action]) # 输出清洗后的数据 cleaned_df.write.mode(overwrite).parquet(hdfs:///data/cleaned/events/)这段代码的逻辑和MapReduce版本完全一致但表达上直观得多。filter对应过滤withColumn对应字段转换dropDuplicates对应去重。而且Spark会自动优化执行计划把能并行的地方都并行起来。4.2 处理格式不一致的时间字段时间字段的清洗是ETL里最烦人的部分之一。不同数据源的时间格式可能完全不同甚至同一个文件里混着多种格式。我的做法是写一个UDF用户自定义函数按优先级依次尝试多种格式from pyspark.sql.functions import udf from pyspark.sql.types import TimestampType from datetime import datetime def parse_multiple_formats(ts_str): if ts_str is None: return None formats [ %Y-%m-%d %H:%M:%S, %Y/%m/%d %H:%M:%S, %Y-%m-%d, %Y%m%d%H%M%S ] for fmt in formats: try: return datetime.strptime(ts_str.strip(), fmt) except (ValueError, AttributeError): continue return None parse_ts_udf udf(parse_multiple_formats, TimestampType()) df df.withColumn(event_time, parse_ts_udf(col(raw_time)))这个UDF会依次尝试四种常见格式只要有一种匹配就返回解析结果全部失败则返回null。后续再用filter(col(event_time).isNotNull())把解析失败的记录过滤掉。提示UDF的性能比内置函数差因为无法被Spark的Catalyst优化器优化。如果数据量特别大建议先用内置的to_timestamp尝试主要格式只对解析失败的记录调用UDF做二次处理。4.3 用SQL风格做聚合校验除了字段级清洗Spark SQL还适合做跨字段的逻辑校验。比如检查订单的开始时间是否早于结束时间或者统计每个用户的行为次数是否在合理范围内cleaned_df.createOrReplaceTempView(events) valid_df spark.sql( SELECT user_id, event_time, action, amount FROM events WHERE amount 0 AND amount 1000000 AND event_time 2020-01-01 AND event_time current_timestamp() )这种写法比手写过滤条件清晰得多而且方便后续调整规则。我通常会把所有校验规则集中写在一个SQL文件里改规则的时候不用动代码只改SQL就行。5. 清洗任务上线后最容易踩的四个坑代码写完、任务跑通只是第一步。真正把清洗任务放到生产环境上定时调度才会遇到各种意想不到的问题。下面这四个坑是我和身边同行都反复踩过的每一个都值得你提前防范。5.1 小文件过多拖垮NameNodeMapReduce和Spark默认都会为每个Reducer或Partition输出一个文件。如果清洗任务每天跑一次每次输出几百个小文件一个月下来就是上万个文件。HDFS的NameNode需要为每个文件维护元数据文件数量过多会严重拖慢集群响应速度。解决办法有两个一是在写入前用coalesce或repartition减少分区数二是在任务后面加一个合并小文件的步骤。我通常会在清洗任务的最后加一行cleaned_df.coalesce(10).write.mode(overwrite).parquet(output_path)coalesce(10)会把输出文件数控制在10个左右。具体数字根据数据量调整一般建议每个文件100MB到200MB比较合适。5.2 数据倾斜导致个别Reducer卡死如果按照user_id做key进行聚合而某个大V用户的记录数远超普通用户就会出现数据倾斜——大部分Reducer很快跑完个别Reducer要处理海量数据整个任务卡在99%不动。应对数据倾斜常用的手段有加盐给key拼接随机前缀打散到多个Reducer最后再去掉前缀合并两阶段聚合先局部聚合再全局聚合广播小表如果是join操作导致的倾斜把小表广播出去在清洗场景里如果只是去重而不涉及聚合其实可以不用Reduce阶段直接用Spark的dropDuplicates底层会自动处理倾斜问题。5.3 字符集问题导致中文乱码这个问题在跨系统数据交换时特别常见。上游系统用GBK编码导出文件下游Hadoop默认按UTF-8读取结果中文全部变成乱码。更麻烦的是乱码数据在清洗阶段可能不会被过滤掉因为不是空值也不是格式错误直接进入了结果表。我的做法是在读取阶段就显式指定编码并且在清洗逻辑里加一个中文校验如果某个字段应该包含中文但实际全是乱码字符就标记为异常。Spark读取时可以这样指定spark.read.option(encoding, GBK).csv(input_path)5.4 任务重跑导致数据重复清洗任务因为故障需要重跑时如果输出目录没有做覆盖或清理就会产生重复数据。Parquet格式虽然支持覆盖写入但如果用的是append模式重跑一次就多一份数据。我的习惯是清洗任务的输出一律用overwrite模式并且输出路径按日期分区。这样即使重跑也只会覆盖当天的分区不会影响历史数据。如果业务要求保留多版本可以在路径里加上任务执行的批次号。常见问题根本原因推荐解决方案小文件过多分区数过多coalesce控制输出文件数数据倾斜key分布不均加盐或两阶段聚合中文乱码编码不一致读取时指定编码加校验重跑数据重复写入模式不当使用overwrite日期分区6. 清洗效果怎么验证几个实用的检查手段任务跑完不代表万事大吉你还得确认清洗后的数据确实“干净”了。我通常从四个维度做验证记录数对比、字段完整性、值域分布、抽样人工检查。6.1 记录数对比清洗前后的量级关系最直观的指标就是清洗前后的记录数。如果清洗后只剩原来的一半那说明过滤条件可能太严格了需要检查是不是误杀了正常数据。反过来如果清洗后记录数没怎么变那可能说明清洗逻辑根本没生效。我一般会计算一个“保留率”保留率 清洗后记录数 / 清洗前记录数根据经验保留率在70%到95%之间比较正常。低于70%要排查过滤条件高于95%要确认是否真的做了有效清洗。6.2 字段完整性检查对每个关键字段统计非空记录的比例。比如user_id字段的非空率应该是100%event_time的非空率至少95%以上。如果某个字段的非空率突然下降说明上游数据质量出了问题。from pyspark.sql.functions import count, when, col total cleaned_df.count() completeness cleaned_df.select([ (count(when(col(c).isNotNull(), c)) / total).alias(c _completeness) for c in cleaned_df.columns ]) completeness.show()6.3 值域分布发现隐藏的异常对数值字段做min/max/avg统计对枚举字段做频次统计。如果发现年龄字段的最大值是999或者状态字段出现了预期之外的取值就说明清洗规则还有漏洞。这一步最好做成自动化脚本每次清洗任务跑完后自动执行把统计结果写入一张监控表。一旦某个指标超出预设范围就触发告警。6.4 抽样人工检查机器替代不了的环节不管自动化检查做得多完善我始终保留一个习惯每次清洗任务上线或规则调整后随机抽取100条清洗后的记录人工看一遍。很多逻辑上的问题——比如字段映射错了、时间解析偏了时区——只有人眼才能发现。抽样检查不需要每次都做但在以下场景必须做清洗规则首次上线上游数据源格式变更清洗后数据量出现明显波动下游分析结果出现异常7. 一些让清洗流程更稳的工程化习惯最后分享几个我在多个项目里总结出来的工程化习惯。这些做法不会让清洗逻辑本身变复杂但能显著提升整个流程的稳定性和可维护性。7.1 把清洗规则配置化不要把过滤条件、字段映射这些硬编码在代码里。我习惯把它们抽到一个配置文件YAML或JSON里代码只负责读取配置并执行。这样调整规则时不需要重新编译打包改配置文件重启任务就行。# clean_rules.yaml filters: - field: user_id condition: not_null action: drop - field: amount condition: range min: 0 max: 1000000 action: mark_anomaly transformations: - field: event_time type: timestamp formats: - %Y-%m-%d %H:%M:%S - %Y/%m/%d7.2 异常数据单独分流清洗过程中被过滤掉的记录不要直接扔掉。把它们写入一个单独的“异常数据”目录保留原始内容和过滤原因。这些数据在排查问题时非常有用有时候还能从中发现新的业务机会。7.3 给清洗任务加监控和告警清洗任务是数据链路的第一环它出问题会直接影响下游所有分析。我通常会给清洗任务配置这几个监控指标任务执行时长超过阈值告警输入/输出记录数波动超过20%告警各计数器数值异常过滤量突增告警输出目录文件数小文件过多告警这些监控不需要多复杂的系统用脚本定期检查任务日志和HDFS目录状态就能实现。7.4 版本化你的清洗逻辑每次修改清洗规则都要记录改了什么、为什么改、影响范围是什么。我见过太多团队因为没做版本管理导致某天数据对不上时完全查不出是哪次改动引起的。最简单的做法是用Git管理清洗脚本和配置文件每次上线打一个tag。清洗这件事技术难度不算高但细节特别多。同样一份数据不同的人写出来的清洗逻辑效果可能天差地别。关键还是在于对数据的理解——你得知道数据从哪来、长什么样、要往哪去才能写出真正靠谱的清洗规则。上面这些内容是我在实际项目里一点点积累出来的希望能帮你少走些弯路。 
长河君 · 2026/10/22 10:29:38 · 阅读 1.2k Hadoop数据清洗实战：从MapReduce到Spark的ETL流程与避坑指南
1. 数据清洗在Hadoop里到底在洗什么很多人第一次听到“Hadoop数据清洗”脑子里浮现的画面是拿个刷子把硬盘擦干净。实际上ETL里的清洗环节干的是把原始数据中那些“不规矩”的东西挑出来、修好或者扔掉。原始数据可能来自日志文件、业务数据库导出、第三方接口回传格式五花八门字段缺失、编码混乱、时间戳格式不统一、数值字段混进了字母这些都是常态。我在实际项目里见过最离谱的一份日志数据同一个用户ID字段有的记录是纯数字有的前面带了“user_”前缀还有的干脆是空字符串。如果直接拿这种数据去做分析结果一定是灾难性的。所以数据清洗的核心目标就三个格式统一、脏数据剔除、字段标准化。在Hadoop生态里这件事通常用MapReduce或者Spark来完成因为数据量摆在那里单机脚本跑不动。这篇文章适合谁看如果你已经了解HDFS和MapReduce的基本概念但不知道一个完整的ETL流程该怎么落地那这篇内容就是写给你的。我会从原始数据的常见问题讲起一步步拆解清洗逻辑的设计思路给出可运行的代码框架最后分享几个我在实际跑任务时踩过的坑。全程不堆砌理论只讲能直接拿去用的东西。注意数据清洗没有“万能模板”不同数据源的清洗规则差异极大。本文给出的方案是通用骨架你需要根据自己数据的实际情况调整过滤条件和转换逻辑。2. 原始数据里最常见的五类脏数据在动手写代码之前先搞清楚你要对付的敌人长什么样。根据我处理过的日志、CSV导出、JSON回传等各类数据源的经验脏数据基本可以归为以下五类。每一类的处理策略不同有的需要丢弃有的可以修复有的必须做字段映射。2.1 缺失值与空字段这是最普遍的问题。一条记录里某个字段为空或者用“NULL”“null”“-1”“unknown”等占位符表示缺失。处理方式取决于字段的重要性如果是主键或时间戳缺失通常直接丢弃整条记录如果是次要字段缺失可以填充默认值或者标记为“未知”。这里有个容易犯的错误不要用全局常量去填充所有缺失值。比如把缺失的年龄统一填成0后续做平均值计算时就会严重偏低。更合理的做法是标记为null在分析阶段再决定如何处理。2.2 格式不一致同一个字段在不同记录里格式不同典型的有时间字段有的用“2024-01-15 10:30:00”有的用“2024/01/15”有的用时间戳数字数值字段有的带千分位逗号“1,234”有的带货币符号“¥1234”布尔字段有的用“true/false”有的用“1/0”有的用“Y/N”这类问题的处理思路是先归一化再解析。比如时间字段统一转成标准格式后再输出数值字段先去掉非数字字符再转换类型。2.3 编码与字符集混乱中文数据里最常见的是UTF-8和GBK混用导致出现乱码。还有一种情况是特殊字符如换行符、制表符混入了字段值内部导致CSV解析时列错位。处理这类问题需要在读取阶段就指定正确的字符集并在清洗时对字段值做转义或替换。2.4 重复记录完全重复的记录相对好处理用去重逻辑就能搞定。麻烦的是“近似重复”——比如同一个用户短时间内提交了两次相似的表单只有时间戳差了几秒。这种需要根据业务规则定义去重键比如用“用户ID操作类型时间窗口”来判断。2.5 逻辑矛盾与越界值比如年龄字段出现了200岁订单金额出现了负数开始时间晚于结束时间。这类数据不一定能自动修复但必须被识别出来。我的做法是设置合理的值域范围超出范围的记录单独输出到一个“异常数据”目录后续人工排查。脏数据类型常见表现推荐处理策略缺失值空字符串、NULL、占位符主键缺失丢弃次要字段标记格式不一致时间/数值格式多样归一化后统一解析编码混乱乱码、列错位指定字符集转义特殊字符重复记录完全重复或近似重复定义去重键窗口去重逻辑矛盾越界值、时间倒挂值域校验异常分流3. 用MapReduce搭建清洗流程的核心思路搞清楚脏数据的类型之后接下来要设计清洗流程。在Hadoop里做ETLMapReduce是最基础的方案虽然写起来比Spark繁琐但理解它对掌握整个生态的运行机制很有帮助。一个典型的清洗任务分为三个阶段Map阶段负责逐条解析和初步过滤Combiner阶段做本地聚合可选Reduce阶段做最终的去重和汇总输出。3.1 Mapper端的设计要点Mapper的输入是原始数据的一行输出是清洗后的键值对。这里的关键决策是哪些逻辑放在Mapper哪些留给Reducer。我的经验是能在Mapper做的过滤和转换就不要拖到Reducer。原因很简单Mapper是并行执行的数据在本地就能被过滤掉减少网络传输量。比如格式校验、字段解析、空值过滤这些都应该在Mapper里完成。具体来说Mapper里要做这几件事跳过表头行如果是CSV文件按分隔符拆分字段检查字段数量是否匹配对每个字段做类型转换和格式校验过滤掉不符合基本要求的记录输出清洗后的键值对public class DataCleanMapper extends MapperLongWritable, Text, Text, Text { private Text outputKey new Text(); private Text outputValue new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString().trim(); // 跳过空行和表头 if (line.isEmpty() || line.startsWith(user_id)) { return; } String[] fields line.split(\t); // 字段数量校验 if (fields.length 5) { context.getCounter(CLEAN, FIELD_MISMATCH).increment(1); return; } String userId fields[0].trim(); String timestamp fields[1].trim(); String action fields[2].trim(); String amount fields[3].trim(); String status fields[4].trim(); // 主键不能为空 if (userId.isEmpty()) { context.getCounter(CLEAN, EMPTY_USER_ID).increment(1); return; } // 时间格式归一化 String normalizedTime normalizeTimestamp(timestamp); if (normalizedTime null) { context.getCounter(CLEAN, BAD_TIMESTAMP).increment(1); return; } // 金额字段清洗 String cleanAmount amount.replaceAll([^0-9.], ); if (cleanAmount.isEmpty()) { cleanAmount 0; } outputKey.set(userId); outputValue.set(normalizedTime \t action \t cleanAmount \t status); context.write(outputKey, outputValue); } private String normalizeTimestamp(String ts) { // 支持多种时间格式的归一化逻辑 // 实际实现需要根据数据源格式编写 return ts; } }3.2 Reducer端的去重与聚合Reducer的输入是Mapper输出的键值对相同key的记录会被分到同一个Reducer。利用这个特性可以很方便地做去重对于每个key只保留第一条记录或者按照某个规则选出一条“最优”记录。但这里有个坑如果数据量极大单个key对应的记录数可能非常多导致Reducer内存溢出。解决办法是在Mapper端先用HashSet做本地去重减少传输到Reducer的数据量。或者改用Spark的dropDuplicates底层做了更优化的处理。public class DataCleanReducer extends ReducerText, Text, Text, Text { private Text result new Text(); Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { // 去重相同key只输出第一条 IteratorText it values.iterator(); if (it.hasNext()) { result.set(it.next().toString()); context.write(key, result); } } }3.3 计数器清洗过程的“仪表盘”上面代码里用到了context.getCounter这是MapReduce里非常实用的一个功能。每次过滤掉一条记录就对应一个计数器加一。任务跑完之后你可以在日志里看到每个计数器的值清楚地知道有多少条记录因为什么原因被丢弃。我习惯在项目里设置这几类计数器EMPTY_USER_ID主键为空BAD_TIMESTAMP时间格式错误FIELD_MISMATCH字段数量不对OUT_OF_RANGE数值越界这些数据在排查问题时非常有用。比如某天发现清洗后的数据量突然少了很多一看计数器发现BAD_TIMESTAMP暴涨那就说明上游数据源的时间格式变了需要及时调整解析逻辑。4. 从MapReduce到Spark清洗逻辑的进阶写法MapReduce能跑通清洗流程但写起来确实啰嗦。一个简单的过滤逻辑要写几十行Java代码调试也不方便。如果你们的集群已经上了Spark我强烈建议用Spark SQL或者DataFrame API来做清洗代码量能减少一半以上而且可读性更好。4.1 用DataFrame API做字段级清洗Spark的DataFrame API提供了丰富的内置函数处理缺失值、格式转换、去重这些操作都有现成的方法。下面这段代码展示了如何用PySpark完成和上面MapReduce等价的清洗逻辑from pyspark.sql import SparkSession from pyspark.sql.functions import col, when, regexp_replace, to_timestamp, trim from pyspark.sql.types import DoubleType spark SparkSession.builder \ .appName(DataCleanETL) \ .getOrCreate() # 读取原始数据 raw_df spark.read.csv(hdfs:///data/raw/events/, sep\t, headerTrue, inferSchemaFalse) # 字段清洗 cleaned_df raw_df \ .filter(col(user_id).isNotNull() (trim(col(user_id)) ! )) \ .withColumn(amount, regexp_replace(col(amount), [^0-9.], ).cast(DoubleType())) \ .withColumn(amount, when(col(amount).isNull(), 0.0).otherwise(col(amount))) \ .withColumn(event_time, to_timestamp(col(timestamp), yyyy-MM-dd HH:mm:ss)) \ .filter(col(event_time).isNotNull()) \ .dropDuplicates([user_id, event_time, action]) # 输出清洗后的数据 cleaned_df.write.mode(overwrite).parquet(hdfs:///data/cleaned/events/)这段代码的逻辑和MapReduce版本完全一致但表达上直观得多。filter对应过滤withColumn对应字段转换dropDuplicates对应去重。而且Spark会自动优化执行计划把能并行的地方都并行起来。4.2 处理格式不一致的时间字段时间字段的清洗是ETL里最烦人的部分之一。不同数据源的时间格式可能完全不同甚至同一个文件里混着多种格式。我的做法是写一个UDF用户自定义函数按优先级依次尝试多种格式from pyspark.sql.functions import udf from pyspark.sql.types import TimestampType from datetime import datetime def parse_multiple_formats(ts_str): if ts_str is None: return None formats [ %Y-%m-%d %H:%M:%S, %Y/%m/%d %H:%M:%S, %Y-%m-%d, %Y%m%d%H%M%S ] for fmt in formats: try: return datetime.strptime(ts_str.strip(), fmt) except (ValueError, AttributeError): continue return None parse_ts_udf udf(parse_multiple_formats, TimestampType()) df df.withColumn(event_time, parse_ts_udf(col(raw_time)))这个UDF会依次尝试四种常见格式只要有一种匹配就返回解析结果全部失败则返回null。后续再用filter(col(event_time).isNotNull())把解析失败的记录过滤掉。提示UDF的性能比内置函数差因为无法被Spark的Catalyst优化器优化。如果数据量特别大建议先用内置的to_timestamp尝试主要格式只对解析失败的记录调用UDF做二次处理。4.3 用SQL风格做聚合校验除了字段级清洗Spark SQL还适合做跨字段的逻辑校验。比如检查订单的开始时间是否早于结束时间或者统计每个用户的行为次数是否在合理范围内cleaned_df.createOrReplaceTempView(events) valid_df spark.sql( SELECT user_id, event_time, action, amount FROM events WHERE amount 0 AND amount 1000000 AND event_time 2020-01-01 AND event_time current_timestamp() )这种写法比手写过滤条件清晰得多而且方便后续调整规则。我通常会把所有校验规则集中写在一个SQL文件里改规则的时候不用动代码只改SQL就行。5. 清洗任务上线后最容易踩的四个坑代码写完、任务跑通只是第一步。真正把清洗任务放到生产环境上定时调度才会遇到各种意想不到的问题。下面这四个坑是我和身边同行都反复踩过的每一个都值得你提前防范。5.1 小文件过多拖垮NameNodeMapReduce和Spark默认都会为每个Reducer或Partition输出一个文件。如果清洗任务每天跑一次每次输出几百个小文件一个月下来就是上万个文件。HDFS的NameNode需要为每个文件维护元数据文件数量过多会严重拖慢集群响应速度。解决办法有两个一是在写入前用coalesce或repartition减少分区数二是在任务后面加一个合并小文件的步骤。我通常会在清洗任务的最后加一行cleaned_df.coalesce(10).write.mode(overwrite).parquet(output_path)coalesce(10)会把输出文件数控制在10个左右。具体数字根据数据量调整一般建议每个文件100MB到200MB比较合适。5.2 数据倾斜导致个别Reducer卡死如果按照user_id做key进行聚合而某个大V用户的记录数远超普通用户就会出现数据倾斜——大部分Reducer很快跑完个别Reducer要处理海量数据整个任务卡在99%不动。应对数据倾斜常用的手段有加盐给key拼接随机前缀打散到多个Reducer最后再去掉前缀合并两阶段聚合先局部聚合再全局聚合广播小表如果是join操作导致的倾斜把小表广播出去在清洗场景里如果只是去重而不涉及聚合其实可以不用Reduce阶段直接用Spark的dropDuplicates底层会自动处理倾斜问题。5.3 字符集问题导致中文乱码这个问题在跨系统数据交换时特别常见。上游系统用GBK编码导出文件下游Hadoop默认按UTF-8读取结果中文全部变成乱码。更麻烦的是乱码数据在清洗阶段可能不会被过滤掉因为不是空值也不是格式错误直接进入了结果表。我的做法是在读取阶段就显式指定编码并且在清洗逻辑里加一个中文校验如果某个字段应该包含中文但实际全是乱码字符就标记为异常。Spark读取时可以这样指定spark.read.option(encoding, GBK).csv(input_path)5.4 任务重跑导致数据重复清洗任务因为故障需要重跑时如果输出目录没有做覆盖或清理就会产生重复数据。Parquet格式虽然支持覆盖写入但如果用的是append模式重跑一次就多一份数据。我的习惯是清洗任务的输出一律用overwrite模式并且输出路径按日期分区。这样即使重跑也只会覆盖当天的分区不会影响历史数据。如果业务要求保留多版本可以在路径里加上任务执行的批次号。常见问题根本原因推荐解决方案小文件过多分区数过多coalesce控制输出文件数数据倾斜key分布不均加盐或两阶段聚合中文乱码编码不一致读取时指定编码加校验重跑数据重复写入模式不当使用overwrite日期分区6. 清洗效果怎么验证几个实用的检查手段任务跑完不代表万事大吉你还得确认清洗后的数据确实“干净”了。我通常从四个维度做验证记录数对比、字段完整性、值域分布、抽样人工检查。6.1 记录数对比清洗前后的量级关系最直观的指标就是清洗前后的记录数。如果清洗后只剩原来的一半那说明过滤条件可能太严格了需要检查是不是误杀了正常数据。反过来如果清洗后记录数没怎么变那可能说明清洗逻辑根本没生效。我一般会计算一个“保留率”保留率 清洗后记录数 / 清洗前记录数根据经验保留率在70%到95%之间比较正常。低于70%要排查过滤条件高于95%要确认是否真的做了有效清洗。6.2 字段完整性检查对每个关键字段统计非空记录的比例。比如user_id字段的非空率应该是100%event_time的非空率至少95%以上。如果某个字段的非空率突然下降说明上游数据质量出了问题。from pyspark.sql.functions import count, when, col total cleaned_df.count() completeness cleaned_df.select([ (count(when(col(c).isNotNull(), c)) / total).alias(c _completeness) for c in cleaned_df.columns ]) completeness.show()6.3 值域分布发现隐藏的异常对数值字段做min/max/avg统计对枚举字段做频次统计。如果发现年龄字段的最大值是999或者状态字段出现了预期之外的取值就说明清洗规则还有漏洞。这一步最好做成自动化脚本每次清洗任务跑完后自动执行把统计结果写入一张监控表。一旦某个指标超出预设范围就触发告警。6.4 抽样人工检查机器替代不了的环节不管自动化检查做得多完善我始终保留一个习惯每次清洗任务上线或规则调整后随机抽取100条清洗后的记录人工看一遍。很多逻辑上的问题——比如字段映射错了、时间解析偏了时区——只有人眼才能发现。抽样检查不需要每次都做但在以下场景必须做清洗规则首次上线上游数据源格式变更清洗后数据量出现明显波动下游分析结果出现异常7. 一些让清洗流程更稳的工程化习惯最后分享几个我在多个项目里总结出来的工程化习惯。这些做法不会让清洗逻辑本身变复杂但能显著提升整个流程的稳定性和可维护性。7.1 把清洗规则配置化不要把过滤条件、字段映射这些硬编码在代码里。我习惯把它们抽到一个配置文件YAML或JSON里代码只负责读取配置并执行。这样调整规则时不需要重新编译打包改配置文件重启任务就行。# clean_rules.yaml filters: - field: user_id condition: not_null action: drop - field: amount condition: range min: 0 max: 1000000 action: mark_anomaly transformations: - field: event_time type: timestamp formats: - %Y-%m-%d %H:%M:%S - %Y/%m/%d7.2 异常数据单独分流清洗过程中被过滤掉的记录不要直接扔掉。把它们写入一个单独的“异常数据”目录保留原始内容和过滤原因。这些数据在排查问题时非常有用有时候还能从中发现新的业务机会。7.3 给清洗任务加监控和告警清洗任务是数据链路的第一环它出问题会直接影响下游所有分析。我通常会给清洗任务配置这几个监控指标任务执行时长超过阈值告警输入/输出记录数波动超过20%告警各计数器数值异常过滤量突增告警输出目录文件数小文件过多告警这些监控不需要多复杂的系统用脚本定期检查任务日志和HDFS目录状态就能实现。7.4 版本化你的清洗逻辑每次修改清洗规则都要记录改了什么、为什么改、影响范围是什么。我见过太多团队因为没做版本管理导致某天数据对不上时完全查不出是哪次改动引起的。最简单的做法是用Git管理清洗脚本和配置文件每次上线打一个tag。清洗这件事技术难度不算高但细节特别多。同样一份数据不同的人写出来的清洗逻辑效果可能天差地别。关键还是在于对数据的理解——你得知道数据从哪来、长什么样、要往哪去才能写出真正靠谱的清洗规则。上面这些内容是我在实际项目里一点点积累出来的希望能帮你少走些弯路。