简介由清华大学推出的大数据数据清洗课程第五章《数据抽取》PPT课件共48页并附习题面向高校学生、职场新人及数据从业者。内容系统讲解文本文件抽取、Web数据抽取、数据库数据抽取和增量数据抽取四大主题并辅以Kettle工具演示MySQL连接、分隔符识别、字段类型设置、JSON/XML解析等关键操作。压缩包内为1个pptx文件大小3.78MB课件结构清晰、图文结合便于自学和教学演示。通过学习可理解数据抽取在ETL流程中的核心作用掌握增量抽取的常见实现思路提升实际数据清洗与集成能力。已有1088人学习下载。1. 从PPT标题到线上清洗链路文本、web、数据库、增量抽取的四种脏数据很多团队把数据清洗理解成“数据进仓之后用pandas去重、补空、改格式”但真正让清洗工作量爆炸的往往是抽取阶段就混进来的脏数据。光看这个课件标题就很有代表性文本、web、数据库、增量数据抽取被塞进同一章表面是四种数据来源实质是两类问题——一边是静态源的格式脏一边是增量链路的逻辑脏。格式脏能用规则清逻辑脏得靠水印、快照和幂等设计兜底。这篇博文就按这个顺序展开先讲清洗前怎么立规则再分别处理文本/web、数据库两类静态源最后把增量数据抽取单独拎出来说清楚。适合正在搭数据管道、被“清洗脚本天天改、跑完不知道对不对”折磨的工程师。2. 清洗前先立规则数据画像与校验规则库决定清洗下限2.1 为什么先画像后清洗脏数据率是清洗参数的唯一依据接手一条数据清洗任务我一般不会先写清洗函数而是先回答三个问题哪些字段有空值、哪些字段的值域超出预期、哪些字段的重复率异常。这三个问题决定清洗脚本怎么设计阈值、怎么定规则优先级。比如某张订单表里“省份”字段有 12 种写法这在画像阶段就能看到清洗阶段只需要一张映射表但如果是“单价”字段混入了“元/斤”这样的文本那就得走类型推断逻辑根本不是简单替换能解决的。数据画像的结果通常写成一张脏数据率清单每个字段单独统计缺失率、唯一值数量、格式合规率。这张清单最大的价值不是看一遍就扔而是作为清洗脚本的输入参数。比如唯一值数量异常的字段要加大采样比例格式合规率低于 95% 的字段要先做类型规范化再做业务清洗。常见做法是把画像结果落成一份 JSON 基准快照和清洗后的结果做对照这样每次跑批都能知道“这轮清洗比上轮好在哪里、坏在哪里”。2.2 用pandas快速完成数据画像与基准快照import pandas as pd import numpy as np import json def profile_dataframe(df, sample_size10000): 对DataFrame做数据画像输出每个字段的基础统计和脏数据指标。 sample_size: 超过该行数时只采样统计避免大表全扫拖慢画像速度。 if len(df) sample_size: df df.sample(nsample_size, random_state42) profile {} for col in df.columns: s df[col] # 对常见数据类型分别处理 if pd.api.types.is_numeric_dtype(s): dirty_ratio s.isna().mean() np.isinf(s.astype(float)).mean() min_val, max_val s.min(), s.max() else: # 文本字段单独统计空白值和长度异常 empty_mask s.astype(str).str.strip().eq() dirty_ratio s.isna().mean() empty_mask.mean() min_val, max_val None, None profile[col] { dtype: str(s.dtype), missing_ratio: round(s.isna().mean(), 4), dirty_ratio: round(float(dirty_ratio), 4), unique_count: int(s.nunique(dropnaFalse)), min: min_val if min_val is not None else None, max: max_val if max_val is not None else None, } return profile # 使用示例读入一份web日志或数据库导出文件 df pd.read_csv(raw_orders.csv, dtypestr, nrows20000) profile_result profile_dataframe(df) with open(profile_baseline.json, w, encodingutf-8) as f: json.dump(profile_result, f, ensure_asciiFalse, indent2)这段代码把每个字段的缺失率、脏数据率和唯一值数量一次性算出来并落盘为基准文件。参数说明dtypestr强制所有列按文本读入因为数据库导出经常把“00123”这种编号转成整数导致前导零丢失文本读入能保住原始形态nrows20000是只看一个抽样窗口画像是为了定规则不是为了精确统计全表扫一遍反而浪费时间。unique_count里dropnaFalse是把缺失值也当一个独立值计入避免把“大量缺失但值域正常”的字段误判为高基数。2.3 校验规则库把清洗逻辑变成可维护的配置画像定的是“现状”校验规则库定的是“目标”。我习惯把清洗规则拆成原子操作每条规则回答一个问题这个字段的合法值域是什么、非法值怎么处理、处理不了是丢弃还是置默认值。{ field_rules: [ { field: phone, check: regex, pattern: ^1[3-9]\\d{9}$, on_violation: drop_row }, { field: province, check: in_list, allowed: [北京市, 上海市, 广东省], on_violation: map_by_dict, mapping_dict: province_alias.json }, { field: amount, check: numeric_range, min: 0, max: 1000000, on_violation: set_default, default: 0 } ], global_policy: { empty_string_as_null: true, strip_all_text: true, dedup_key: [order_id, product_id] } }规则库设计有三个关键点。第一check和on_violation必须拆开同一种校验失败可以走不同处理路径比如手机号格式错是丢弃而金额范围超限是置默认值这两种语义不能混。第二mapping_dict指向外部字典文件省份别名、类目映射这类经常变动的映射不进代码进配置改起来不用重发。第三global_policy里的dedup_key是清洗的全局约束执行顺序一定是先做字段级规则、再做去重、最后做类型转换顺序反了会导致去重时类型不一致。2.4 规则引擎执行时的三个常见误用规则引擎本身不复杂但团队里常见三个误用。第一个是把规则写死在 if-else 里看起来每行都好懂实际上规则之间互相调用改一条规则要通读整个函数建议至少拆成“加载规则-执行校验-执行处理”三段。第二个是用规则库直接改原始 DataFrame一旦规则写错原始数据被污染想重跑就得重新抽取所以我一般要求清洗脚本输入输出都走文件或表原始层只读。第三个是忽略规则的顺序依赖比如手机号字段要先去空格再校验正则如果清洗顺序是“先校验、后strip”大量合法带空格的数据会被误杀这个问题在画像阶段其实是能看出来的——dirty_ratio明显偏高但正则检查全过基本就是顺序写反了。3. 文本与web数据抽取清洗标签、编码、噪声的正则与语义清理3.1 web数据抽取的两种脏HTML标签噪声和埋点混淆从web页面抓下来的数据脏的形式和文件导入完全不同。最常见的是“清洗HTML文档中无意义数据”具体表现为三种一是标签残渣div classprice299/div抽完没剥干净文本里残留 div 和 class二是导航与脚本噪声页面里的菜单、页脚、script里的埋点参数全被捞进来尤其是带_trackEvent、utm_这类参数的字符串混进业务字段里特别难查三是HTML实体nbsp;、amp;这类字符在渲染时是空格和 但抽出来就是原样字符串直接导致文本比对失真。做web清洗我通常分两层第一层是结构清洗用解析器按标签语义剥掉 script、style、nav 这些非内容块只保留正文节点第二层是文本清洗处理实体字符、控制字符和多余空白。这两个层次不能合并因为结构清洗依赖DOM树文本清洗依赖正则和字典混在一起写会让脚本既处理不了解析失败也处理不了乱码。3.2 用BeautifulSoup剥壳、用正则做语义正则化from bs4 import BeautifulSoup import re import html def clean_html_to_text(raw_html: str) - str: 把HTML字符串清洗成纯文本。 分两层先按标签剥除再做实体与噪声字符清理。 # 第一层结构清洗去掉脚本、样式、导航等非正文块 soup BeautifulSoup(raw_html, html.parser) for tag in soup([script, style, nav, footer, iframe]): tag.decompose() # 提取文本并按行拆开避免所有内容挤成一行 text soup.get_text(\n, stripTrue) # 第二层语义清洗 text html.unescape(text) # amp; - , nbsp; - 空格 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f], , text) # 控制字符 text re.sub(r[ \t], , text) # 多个空格换成一个 text re.sub(r\n{3,}, \n\n, text) # 三个以上换行收敛 return text.strip()这段代码的层次很关键。tag.decompose()是彻底移除节点不是在get_text之后再过滤字符串因为埋点脚本里的参数可能恰好和正文文本相似用正则误杀面太大。html.unescape放在结构清洗之后优先处理实体字符顺序反了会导致lt;scriptgt;这种被转义过的标签变成真实标签干扰结构解析。get_text(\n, stripTrue)里的stripTrue会在每个标签边界自动清首尾空白替后续正则减少工作量。实际项目里经常要加一步处理nbsp;造成的假空格。html.unescape会把它转成普通空格所以要在\s匹配中统一收掉。如果用 pandas 批量清洗文本列上面这个函数配合df[text].apply(clean_html_to_text)就能跑完整列跑之前建议先看一版dirty_ratio和清洗后的文本长度分布确认没有整列被清空的情况。3.3 编码乱码与不可见字符chardet与latin1陷阱web数据的第二个大坑是编码。HTTP 响应头写的 charset 经常和实际字节流对不上最常见的是 utf-8 页面里混入 gbk 编码的片段以及latin1误读导致的 乱码。我一般用chardet或者 Python 3.10 自带的charset_normalizer先检测再按检测结果解码。import chardet def robust_decode(data: bytes) - str: 尝试用多种编码解码字节流优先信检测结果失败时回退。 data: 从web抓包的原始字节来自 response.content。 # 先看BOMBOM比检测结果更可信 if data.startswith(b\xef\xbb\xbf): return data.decode(utf-8-sig) if data.startswith(b\xff\xfe): return data.decode(utf-16-le) detected chardet.detect(data[:2000]) encoding detected.get(encoding) or utf-8 confidence detected.get(confidence, 0) if confidence 0.7: # 置信度低时按常见页面编码尝试 for enc in (utf-8, gb18030, latin1): try: return data.decode(enc) except UnicodeDecodeError: continue return data.decode(encoding, errorsreplace)这个函数有几个参数值得注意。chardet.detect(data[:2000])只取前 2000 字节检测因为编码特征在文件头就足够判断全量检测在大文本上会拖慢好几倍。confidence 0.7的阈值是经验值低于这个阈值说明检测结果可能是猜的此时按常见的 utf-8、gb18030 顺序硬解gb18030 是 gbk 的超集兼容性最好。最后一行errorsreplace是兜底策略宁可让乱码字段变成替换字符也不能让整个任务崩溃——替换字符可以通过后续的字符范围清洗识别出来。3.4 文本清洗的边界不要把脏数据洗成假数据文本清洗最容易犯的错是“洗过头”。比如把商品标题里的品牌词全清掉理由是“这些词对分析没意义”等业务侧要按品牌维度分析时发现数据已经被破坏。我一般保留两层数据原始文本存一份清洗后文本存一份清洗脚本只产出后者的新字段不覆盖原始字段。再比如正则写得太宽.*把英文单引号后面的内容全吞了清洗前后的字符量对比就能看出来——清洗后文本比清洗前短了 60% 的字段要逐个检查。还有一个边界是大小写和中文全半角习惯性全转半角会让英文缩写和中文标点混在一起实际上很多下游模型对中文逗号和英文逗号是区别对待的这个要看具体用途不能一刀切。4. 数据库数据抽取清洗约束、去重、规范化与“先抽后洗”4.1 数据库里的脏和文件里的脏不一样约束破坏与类型漂移从数据库抽数据很多人默认“库里是干净的”实际刚好相反。数据库能保证的是约束不是质量——主键唯一但业务主键可能选错非空字段可能是空字符串日期字段可能混进“2024-13-01”这种非法值。做数据库清洗我关注四个专项约束冲突主键重复、外键悬空、类型漂移字符型字段混入数字和日期、单位不一致有的行按“元”有的按“万元”、以及时区问题订单时间有时是北京时间有时是UTC。前两个用 SQL 就能扫描后两个得靠业务语义清洗脚本里一定要有字典和换算表。4.2 抽取即清洗SQL层面的trim、cast、默认值兜底数据库清洗有个特殊优势可以在 SQL 里直接做不用把数据拉到应用层。这样既省网络传输又能利用数据库的算子做并行处理。我给一个最常用的“抽取即清洗”SQL模板-- 从业务库抽取订单表抽取过程中完成基础清洗 INSERT INTO dwd_orders_cleaned ( order_id, customer_id, order_date, amount, status ) SELECT TRIM(CAST(order_id AS CHAR(32))) AS order_id, COALESCE(TRIM(CAST(customer_id AS CHAR(20))), UNKNOWN) AS customer_id, CASE WHEN order_date IS NULL OR order_date THEN NULL WHEN order_date NOT REGEXP ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ THEN NULL ELSE order_date END AS order_date, CAST(amount AS DECIMAL(12,2)) AS amount, CASE WHEN status IN (P, S, F) THEN status ELSE X END AS status FROM raw_orders WHERE order_id IS NOT NULL;参数说明TRIM(CAST(order_id AS CHAR(32)))是为了处理数据库导出的 char 类型自动补空格问题很多老系统用char(32)存订单号取值自带右填充空格不在 SQL 里 trim 会导致下游 join 永远失配。COALESCE(..., UNKNOWN)是把空客户ID映射成显式值避免下游外键关联时大量 NULL。amount直接CAST AS DECIMAL(12,2)如果原字段有非数字内容会直接报错所以需要结合REGEXP先做合法性过滤。最后WHERE order_id IS NOT NULL是在源头卡掉主键缺失的行。4.3 主键重复与业务去重ROW_NUMBER窗口去重主键重复在“先抽后洗”链路里最常见。原因是业务库的复合主键和你定义的自然键不一致例如业务主键是(order_id, line_no)而数据仓库要求订单维度唯一。直接SELECT DISTINCT只能去完全重复的行做不到按优先级保留一条WITH ranked_orders AS ( SELECT order_id, customer_id, amount, ROW_NUMBER() OVER ( PARTITION BY order_id ORDER BY updated_at DESC, id DESC ) AS rn FROM raw_orders ) SELECT order_id, customer_id, amount FROM ranked_orders WHERE rn 1;ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC, id DESC)是这张表里去重保留策略的核心同一个order_id出现多次时按updated_at最新一条优先如果更新时间相同再按自增id大的优先。这里有个容易被忽视的点UPDATE时间相同并不能说明两条数据一样所以id DESC作为次级排序是必要的否则结果不确定。跑完这步之后再单独做一次COUNT(*)对比确认去重后的行数和COUNT(DISTINCT order_id)一致。4.4 规范化冲突同值不同形与同形不同值数据库清洗里最花时间的是“同值不同形”同一个用户手机号有的带86前缀有的不带同一个公司名有的带“有限公司”后缀有的不带。处理这种冲突不需要机器学习一张别名映射表就够了-- 手机号统一去前缀 UPDATE dwd_customer_clean SET phone REGEXP_REPLACE(phone, ^\?86, ) WHERE phone REGEXP ^\?86[0-9]{11}$; -- 公司名映射统一 UPDATE dwd_customer_clean c LEFT JOIN company_alias a ON c.company_name a.alias_name SET c.company_name COALESCE(a.standard_name, c.company_name);第一句的正则^\?86匹配可选加号开头的 86把国际区号前缀去掉。这里有个坑如果直接REPLACE(phone, 86, )会把手机号中间出现的 86 也删掉所以必须先REGEXP限定是“开头且后跟 11 位数字”再做替换。第二句是典型的查表映射company_alias表里存的是“别名 - 标准名”LEFT JOIN 加 COALESCE 的含义是“有映射就换没映射保留原值”这样清洗脚本可以反复跑不会因为一次映射不全会丢数据。4.5 先抽进再洗还是洗好再抽两条路线的取舍表场景先抽再洗先洗再抽源库是OLTP核心库不能跑重SQL优先不选清洗逻辑需要看多张表关联优先不选源库和目标库是同一套引擎可选可选数据量百亿级清洗后再写目标慎用优先清洗规则需要频繁调参优先重跑成本低不选重跑要拉全量我个人的默认路线是先抽再洗。原因很简单清洗规则一定会调先抽到清洗层每次调参只需要重跑清洗作业不需要再连业务库如果先洗再抽规则一变就得重新连源库拉数据业务库的压力和网络成本都会翻倍。唯一例外是数据量特别大、清洗只是做类型转换的场景这种情况直接在抽取 SQL 里做掉最省事。5. 增量数据抽取的清洗陷阱水印、快照对比与幂等回放5.1 增量抽取不是“只取新增”是“重跑不变的产物”很多人第一次做增量数据抽取以为就是在WHERE后面加个create_time 昨天跑完发现下游报表数据对不上。增量抽取的清洗难点在于增量数据必须可重放。业务库的行可能被更新多次昨天抽取的行今天就可能变了如果你只按创建时间抽新增更新的数据永远抽不进来。所以增量抽取通常要同时维护两条时间戳create_time管新增update_time管变更。5.2 水印表设计让增量清洗的起点可配置-- 水印表只存一张记录每个表最近一次的抽取位点 CREATE TABLE dwd_incremental_watermark ( source_table VARCHAR(64) PRIMARY KEY, last_high_water DATETIME NOT NULL, extract_batch_no VARCHAR(32), updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ); -- 原子更新水印只有本次抽取成功后才推进水印 UPDATE dwd_incremental_watermark SET last_high_water 2024-11-01 00:00:00, extract_batch_no BATCH_20241101_01 WHERE source_table raw_orders AND last_high_water 2024-11-01 00:00:00;水印表的核心逻辑在最后一行last_high_water 2024-11-01 00:00:00翻译过来是“只有新水位大于旧水位时才推进”这个条件保证幂等——同一批次重复执行时第二次因为条件不满足不会重复推进水位。extract_batch_no字段用来标记每个批次的水印更新方便出问题时把数据回滚到指定批次。注意水印更新必须和清洗目标表的写入放在同一个事务里否则会出现“数据写了一半水位已经推进”的问题这种情况比不推水位还难查。5.3 快照对比法当时间戳不可信时的增量识别有些老系统的表根本没有update_time字段或者字段有但不更新。这时候增量识别只能靠快照对比全量抽取一张小表和上一次抽取结果做比对差异行就是增量。常见做法是两张表做FULL OUTER JOINSELECT COALESCE(a.id, b.id) AS id, CASE WHEN b.id IS NULL THEN INSERT WHEN a.id IS NULL THEN DELETE WHEN a.col1 b.col1 OR a.col2 b.col2 THEN UPDATE ELSE UNCHANGED END AS change_type FROM snapshot_prev a FULL OUTER JOIN snapshot_curr b ON a.id b.id;这个写法把 INSERT、DELETE、UPDATE 一次全识别出来。COALESCE(a.id, b.id)处理的就是“一边有另一边没有”的行。UNCHANGED的分类结果会进入下游过滤实际写数据时只保留前三种。快照对比的缺点也明显每次要拉全量表做 join数据量大时成本很高所以一般只用于“表行数在百万级以内、又没有时间戳”的场景千万级以上还是得推动业务表加update_time。5.4 数据漂移与重复拉取增量清洗的经典翻车现场数据漂移是增量链路里最难查的问题业务库凌晨计算了当天的统计值第二天又修正你的增量抽取已经在零点跑完了修正值丢了下游。处理方式一般是把每天的抽取窗口从“零点跑一次”改成“延迟 2 小时再跑并回刷最近 3 天的分区”具体回刷天数看业务修正周期。另一个翻车现场是重复拉取比如任务重跑导致同一个BATCH_20241101_01的数据被插入两次如果目标表没有做唯一键约束就会产生重复数据。所以增量清洗的目标表我一般都会建唯一键写入用INSERT ... ON DUPLICATE KEY UPDATE或者MERGE语义保证同一批次跑两次结果一致。6. 增量回放验证与清洗链路的高频踩坑增量清洗做完之后验证比实现更重要。推荐一个低成本高回报的验证方法回放测试。方法是把过去 7 天的源数据重新跑一遍清洗流程要求输出的结果和当初增量跑出来的完全一致。如果回放结果对不上说明清洗链路里有非确定性逻辑常见的三个来源是用了RAND()、用了当前时间NOW()做默认值、或者读取了外部状态比如字典表被改过。回放测试跑通了增量清洗的幂等性才有底。增量链路上有个高频踩坑点是时区处理。数据库里的update_time可能是应用服务器写入的北京时间也可能是数据库服务器的系统时间两者混在一个表里时增量水位会来回跳。排查方法是抽一个时间段同时看update_time和create_time的分布如果发现凌晨的数据更新时间晚于白天基本就是时区混用。解决方法是清洗层统一转 UTC 存储展示层再转北京时间不要在不同阶段混用两种时区。最后给出一个增量抽取自检清单每一条都是我在生产环境里被坑过的检查项验证方法水印是否只在成功后推进在事务未提交时查水印表确认值未变同一批次重跑是否幂等连续跑两次对比两次输出行数和哈希值更新时间与业务时间的边界create_time 水位的行是否被重复抽取目标表主键是否约束完整查询目标表重复键数量应为 0web抽取文本是否随页面改版漂移用一个固定URL样本做回归测试每周跑一次增量清洗做到最后拼的不是规则写得多精巧而是能不能保证“每次跑、跑几次、什么时候跑结果都稳定”。把水印、幂等、回放验证这三件事做扎实任何格式的脏数据都只是规则表的增删不会演变成链路事故。本文还有配套的精品资源点击获取