简介《ETL设计详解数据抽取、清洗与转换》是一份面向BI项目实施人员、数据仓库开发工程师及数据入门学习者的Word文档。文档围绕ETL在BI项目中约占三分之一工作量的重要地位展开按数据抽取、数据清洗转换、数据加载三部分进行系统讲解其中抽取环节包含调研阶段需要确认的四个关键问题、同源/异源数据库及文件型数据源的处理方式清洗环节梳理了不完整、错误、重复三类脏数据的判别与过滤策略转换环节则涉及不一致数据转换、粒度聚合和业务规则计算加载部分还对比了OWB、DTS、SSIS等工具实现与SQL方式的适用场景并补充了ETL日志和警告发送的维护要点。资源包为1个docx文档大小仅20KB内容紧凑、要点齐备便于在方案设计或学习时快速翻阅。目前已有1534人学习适合作为数据仓库入门培训或项目选型的轻量参考。1. ETL 设计详解数据抽取、清洗与转换决定数仓的下限提到 ETL 设计详解数据抽取、清洗与转换很多人的第一反应是写几条insert into ... select ...把数据搬进数仓。真正把任务跑上生产三个月才会意识到 ETL 的难点不在写代码而在设计源系统变更时抽取不能断脏数据不能悄悄污染报表重跑任意一天都要得到一致结果。这篇文章把抽取、清洗、转换三个阶段拆开讲清楚从全量与增量选型、字段级清洗、粒度转换到调度重跑和验证给出一套可以直接照做的落地思路。2. 数据抽取全量增量与 CDC先定策略再写代码抽取是 ETL 的第一道关口也是最容易失控的环节。很多任务挂掉不是因为目标数仓不够强而是抽取方式没想清楚不分青红皂白全量拉表几亿行的事实表每次同步都让源库吃满 IO或者只用单条 SQL 不分页数据量一大就内存溢出。我的习惯是写任何抽取脚本之前先回答三个问题源表多大、更新频率多高、历史数据会不会被修改。回答完这三个问题方式基本就出来了。2.1 全量、增量和 CDC三种抽取方式怎么选三种方式的边界要分清楚不能只凭感觉。全量抽取适合维度表、配置表这类小表直接truncate后重灌简单且能保证目标与源完全一致增量抽取适合大表按自增主键或更新时间拉取新增数据对源库压力小CDC 则是在源库开启日志解析把插入、更新、删除都同步过来适合对实时性要求高、且源库允许开启日志的场景。抽取方式适用场景优点主要成本全量小维度表、配置表、慢速变化表实现简单数据一致性好每次全扫数据量大后扛不住增量主键/时间戳大流水表、append-only 表对源库压力小易实现断点续传无法可靠捕捉历史修改CDC需要同步增删改、实时性高的业务表完整记录变更延迟低依赖源库日志配置运维复杂度高对于没有专职 DBA、源库也不敢乱动的团队我一般推荐“小表全量大表增量”。增量字段优先选自增主键或update_time但要注意一个坑如果源系统把历史行的update_time一起刷掉增量会漏数。此时只能用 CDC 或每天对近 N 天数据做一次比对补偿具体做法在避坑章节展开。2.2 增量抽取的最小实现游标分页与增量水位先看一段常见的 Python 增量抽取骨架。它不停用游标也不是一次性把大结果集装进内存而是按 ID 逐批拉取拉完一批更新一次水位。def fetch_incremental(db, table, last_id, page_size5000): 按自增主键增量抽取 db: 数据库连接对象 table: 源表名 last_id: 上一批的最大主键第一次传 0 page_size: 每批行数 while True: rows db.execute( SELECT * FROM {table} WHERE id %(last_id)s ORDER BY id ASC LIMIT %(page_size)s, {last_id: last_id, page_size: page_size}, ) if not rows: break for row in rows: yield row # 把本批最后一行主键作为新水位 last_id rows[-1][id]逻辑说明:WHERE id last_id保证每一批都从上一批停下的地方继续ORDER BY id让排序稳定LIMIT控制单批内存。yield让调用方可以边拉边写目标表避免大量行积压在内存里。last_id这个水位既可以用在内存循环中也可以写到数据库表里任务中断后从库里读回来接着跑。参数说明:page_size建议从 5000 开始调。源库性能好、字段少可以加到 10000字段多或者源库是核心交易库降到 2000 更稳妥。还有一种常见做法是用WHERE update_time last_update如果时间是精确到秒的很容易漏掉同一秒内写入的数据所以要配合AND update_time next_boundary去重。2.3 断点续传与目标表增量水位断点续传不能只靠代码里的局部变量。程序一重启内存里的last_id就丢了所以要把水位落到持久化位置。最简单的方式是单独一张抽取状态表每次任务启动先读状态每批拉完就更新。-- 抽取状态表每个业务表一行 CREATE TABLE etl_state ( table_name VARCHAR(128) PRIMARY KEY, last_id BIGINT, last_update TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 每次抽取前读取水位 SELECT last_id, last_update FROM etl_state WHERE table_name ods_order;水位表的设计逻辑:last_id负责主键增量last_update负责下次任务开始的时间参考。我习惯把“读取水位”和“写完数据”放在同一个事务边界里避免数据写成功了但水位没更新导致下一轮重复抽取。如果你用的是自研调度平台把这个状态表当成普通目标表一样维护失败重跑时从状态表恢复即可。抽取还有一条红线: 永远不要用SELECT *去拉一张几百列的表。只选目标模型需要的列既减少网络传输也降低源库临时表空间消耗。给源库查询加SET TRANSACTION READ ONLY可以防止开发时手滑把更新语句写进抽取任务。3. 数据清洗把脏数据挡在事实表之前清洗是最看不出功绩、但最重要的环节。业务库里的数据往往带着各种历史包袱订单号有空格、金额字段混着货币符号、用户手机号有的带区号有的不带、同一个用户在两张表里主键不一致。清洗的目的不是把所有数据变成“标准答案”而是把质量基线定清楚保住核心指标的可信度。3.1 清洗规则先于代码字段质量基线我见过最典型的失败案例是一上来就写 Python 函数清洗数据结果清洗规则散落在脚本各处换了数据源就找不到哪条规则在哪生效。正确做法是先维护一张字段质量基线表把每个字段的规则、告警阈值、处理动作写出来。字段质量规则异常处理动作告警阈值订单号非空、长度 8-32 位、无首尾空格丢弃并记录_bad表缺失率 0.1%金额数字类型、大于 0非法值置 NULL参与后续 COALESCE负值条数 0手机号11 位数字统一去除非数字字符位数不符置空空号率 5%订单时间能解析为日期、不晚于当前时间解析失败的落_bad表失败率 0.05%这条基表的用处是让清洗变得可审计。开发时按规则实现上线后按阈值告警而不是等业务方发现报表数字诡异才回头查。基线表本身可以用一份简单的配置或 SQL 表维护不需要引入复杂规则引擎。3.2 空值和重复SQL 去重与 Python 清洗清洗最常处理的两类问题是空值和重复。先说重复同一个业务主键出现多行往往是因为源表有多条更新记录。此时最稳的做法是用窗口函数保留最新一条WITH deduped AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY order_no ORDER BY updated_at DESC ) AS rn FROM raw_order ) SELECT * FROM deduped WHERE rn 1;逻辑说明:PARTITION BY order_no界定重复范围ORDER BY updated_at DESC指定保留最新更新。如果源表没有更新时间就用自增主键倒序排。注意rn 1会过滤掉重复的旧数据清洗结果行数一定小于等于原表下游如果发现指标少了很多要能接受这个预期。Python 侧处理适合实验性分析或字段格式不稳定的场景。我一般用 pandas 做探查再用 DataFrame 清洗逻辑固化成任务import pandas as pd df pd.read_csv(raw_order.csv, dtype{order_no: string}) # 去空格、统一手机号字母形式 df[order_no] df[order_no].str.strip() df[phone] df[phone].str.replace(r\D, , regexTrue) # 按业务主键去重保留最后一条 df df.drop_duplicates(subset[order_no], keeplast) # 金额转数值解析失败置空再补默认值 df[amount] pd.to_numeric(df[amount], errorscoerce).fillna(0)逻辑说明: 先做格式规整再做去重是因为原始数据可能同一个手机号一个带空格一个不带直接做主键去重会漏掉真正重复的行。errorscoerce会把无法解析的金额变成NaN再用fillna(0)兜底。参数说明:keeplast表示保留最后一条如果你更信任先发生的数据改成first。subset必须是真正的业务主键不要用全字段去重否则源表一个字段的微小差异就会让重复行全部保留。3.3 字符编码和时区两个最容易翻车的清洗项字符类问题很隐蔽。业务库写入时用的是应用侧编码到了数仓可能变成乱码尤其常见的是中文环境下出现utf8和latin1混用。表象是主键冲突、字符串比较永远不相等。处理方式是在抽取阶段就统一转成 UTF-8到了清洗阶段再发现乱码往往已经丢了原始信息很难逆推。时区问题则是指标对不齐的常见来源。如果订单创建时间存的是 UTC日报却按东八区汇总每天凌晨 0 到 8 点的单子会被分到前一天。统一做法是让数仓所有表都存 UTC展示层再转目标时区如果源系统已经存了本地时间必须在抽取时记录时区偏移量不能靠猜。from datetime import datetime, timezone, timedelta raw_ts 2024-06-01T12:00:00Z # 解析 UTC 时间 dt datetime.fromisoformat(raw_ts.replace(Z, 00:00)) # 转成目标偏移时区 local_dt dt.astimezone(timezone(timedelta(hours8))) print(local_dt.strftime(%Y-%m-%d %H:%M:%S))代码逻辑是先把字符串末尾的Z改成00:00让fromisoformat识别为标准 UTC 时间再用astimezone做偏移转换。这里的参数timedelta(hours8)就是目标时区相对 UTC 的小时数换成其他业务时区就改这个值不要在代码里写死多个字符串格式。最重要的是清洗阶段做完时间统一后下游所有任务都不得再自行换算否则一处统一一处乱排查起来非常痛苦。4. 数据转换从字段映射到维度建模转换阶段解决的核心问题是把源系统面向业务操作的表结构改成面向分析的目标结构。这里涉及到字段名统一、类型对齐、粒度选择、维度补充。很多人把转换理解为写长 SQL 把表 join 起来但对生产环境真正的难点是你怎么保证 join 之后没放大数据、没丢数据以及重跑时结果不变。4.1 字段映射与类型转换源库的主键可能是字符串目标数仓要求 bigint源库的金额可能带千分位目标模型需要数值类型。这些转换看着简单却最容易因为脏数据跑挂整个任务。建议在映射阶段做一步“显式 cast 错误隔离”。SELECT order_no, SAFE_CAST(user_id AS BIGINT) AS user_key, TRIM(COALESCE(NULLIF(phone, ), 0)) AS phone, SAFE_CAST(REGEXP_REPLACE(amount, ,, ) AS DECIMAL(12, 2)) AS amount FROM stage_order;逻辑说明:SAFE_CAST是先把类型转换做成安全函数遇到非法字符返回NULL不会让整个任务报错NULLIF(phone, )把空字符串变成NULL再配合COALESCE给一个占位值避免下游 join 出现莫名其妙的空值。REGEXP_REPLACE去掉金额里的千分位逗号再做数值转换。参数说明:DECIMAL(12,2)表示最大 12 位数字、2 位小数在金额超过亿级时改大精度否则会因溢出而失败。这里的关键点是转换失败不能静默。用SAFE_CAST后要把原本会失败的行单独抽样记录例如记录user_id非法的行数和样例方便后续反向修数。直接不加保护地CAST一个坏数据能让整批任务重跑半小时这种奶酪式错误在真实生产里见得太多了。4.2 粒度转换明细到汇总先定粒度再写 GROUP BY粒度决定了每一行代表什么也决定了下游指标口径。从订单明细转换到每日商品汇总一个常见错误是先 join 再聚合结果 join 产生的重复行导致销量翻倍。正确做法是先想清楚“最终一行是什么”再倒推需要哪些维度字段。SELECT DATE(order_time) AS stat_date, product_id, COUNT(*) AS order_cnt, SUM(order_qty) AS qty, COUNT(DISTINCT user_id) AS uv FROM dwd_order_detail WHERE order_time 2024-06-01 AND order_time 2024-06-02 GROUP BY DATE(order_time), product_id;逻辑说明:DATE(order_time)把订单时间落到天级粒度product_id是第二个分组维度所以最终每个分区是“某天 某商品”一条记录。COUNT(DISTINCT user_id)用于统计去重用户数不能直接COUNT(user_id)否则一个用户下多单会被多算。参数说明: 时间过滤采用左闭右开即 开始 AND 次日这样不会漏掉 23:59:59.999 的数据也方便后续做增量。转换时还要防“join 放大”。一个常见自检方法是join 前先分别统计左右表的行数再统计 join 后行数如果 join 后行数明显大于左表说明右表存在一对多关系需要先对右表做去重或在 join 条件里补维度。这个操作看起来笨但这几分钟能避免一次数据质量事故。4.3 维表补充与缓慢变化维的取舍事实表通常只存业务键比如user_id、product_id分析时需要补齐用户等级、商品类目等描述字段。最稳妥的维表补充是时间快照关联事实表下单时间和维表生效日期匹配尽量不直接关联当前维表否则历史订单会跟着维表变化历史汇总口径全变。SELECT f.order_no, d.user_level, d.user_region FROM dwd_order_fact f LEFT JOIN dim_user_snapshot d ON f.user_id d.user_id AND f.dt d.effective_date AND f.dt d.expire_date;这段 SQL 用了拉链表的时间段关联可以精确还原每个用户在某一天当时的等级。参数说明:effective_date和expire_date是拉链表的生效区间f.dt必须是分区日期否则范围关联会扫描全表。如果维表不需要历史回溯直接LEFT JOIN当前版本表即可但你要能接受历史订单的用户等级是当前值。缓慢变化维的取舍要看分析需求。报表需要“上个月下单时用户是什么等级”必须保留历史版本也就是 SCD2 拉链表报表只关心当前等级用 SCD1 覆盖更新即可。我的建议是不要一上来就建拉链表先问业务方三个问题看历史吗、统计口径变过吗、能接受当前值吗。都回答“要”再上 SCD2否则维护成本远大于收益。5. ETL 避坑指南我跑挂过的 5 个真实问题这一章的每条经验都来自线上事故现象、原因、解决方式写得直白一点希望能帮你少踩一次坑。ETL 上线后最可怕的不是任务失败而是任务显示成功但数据是错的。5.1 抽取把源库拖垮一跑全表扫描锁竞争严重现象某次凌晨 2 点运行增量抽取后业务系统第二天早晨反馈核心交易变慢数据库监控显示大量SELECT长时间占用 IO。原因新接手的 A 同学给一张 2 亿行的订单流水表做了全量抽取且没有分页查询直接把源库临时表空间打满。解决把全量改成增量用modified_time 上次水位拉取如果业务必须用全量则拆成 2000 行一批的分页任务并把运行时间放在业务低峰期。另一个有效手段是给查询加/* */并行 hint 限制但不要依赖这个优先从数据量上控制。5.2 空值被“清洗”过滤指标少了一块现象销售日报中某个大区 GMV 突然下降 12%但任务状态全部成功。原因清洗阶段用了WHERE user_id IS NOT NULL过滤行但很多订单的user_id在源系统确实为空属于匿名下单应该保留条数但把用户维度设为unknown而不是整行丢弃。解决清洗规则中区分“非法值”和“缺失值”缺失值保留非法值落_bad表。给出的建议是每一个过滤条件都反查一下被过滤的数据量如果预期过滤率超过 1%必有问题。从此我把所有清洗脚本都加了“过滤行数统计日志”。5.3 字符集不一致主键冲突离奇出现现象目标表报主键冲突但抽样看每条订单号都不重复。原因源表是 Ubuntu 下写入的表情符号目标库按 Latin1 解析把同一个订单号转成了不同的乱码导致实际字节相同但显示不同数据库层面则判断为重复。解决在 JDBC/连接串中显式指定utf8mb4并在抽取后对order_no做一次十六进制校验。最简单的办法是每次同步后跑一遍SELECT COUNT(DISTINCT order_no) vs COUNT(*)不一致就立刻告警不能等到跑数仓模型时才发现。5.4 重跑任务不幂等同一份订单被记两次现象某天调度任务因网络超时挂了运维手动重跑结果月报订单量几乎翻倍。原因第一次任务实际已经写入了数据只是在返回成功前连接断开重跑时没有先清理目标分区导致相同主键的数据写入两次。解决给所有目标表加“重跑前清理”逻辑比如DELETE WHERE dt 2024-06-01或直接删分区。这个习惯非常重要。凡是支持重跑的任务必须做到“同一参数、同一结果”否则数据质量根本无法保证。我现在把幂等性当成代码评审的一个硬指标。5.5 只看成功失败行数腰斩却没告警现象任务状态绿色但当天数据行数从 200 万跌到 90 万直到业务方看报表才暴露。原因源系统发布新版本后接口返回格式变化抽取程序解析 JSON 少了一层字段但语法没有报错空数据被当成正常结果写入。解决在所有抽取和转换任务里加行数波动检查和近 7 天平均值对比波动超过 20% 就发告警。具体实现可以是一个简单的 SQL 检查也可以是调度平台的传感器。只要做到这一步数据量暴跌就不会过夜才发现。6. 把 ETL 设计成能长期维护的流水线前三章解决的是单任务怎么写最后一章解决的是整套流程怎么维护。ETL 不是跑起来就结束的代码它会被调、被改、被重跑所以分层、依赖验证和调试手段必须一开始就设计好。6.1 任务分层与依赖配置我习惯把任务按 ODS、DWD、DM 三层拆开ODS 只做抽取和原样落地DWD 做清洗转换DM 做汇总输出。上层任务不直接依赖源库只依赖下一层的结果表这样源库变更时影响范围能被隔离在某一层。调度上可以用 cron 设置“先抽取后转换再汇总”的依赖顺序。一个最小 crontab 配置可以是# 每天凌晨 01:00 抽取原始数据 0 1 * * * bash run_extract.sh --date $(date -d yesterday %F) # 02:00 清洗转换 0 2 * * * bash run_transform.sh --date $(date -d yesterday %F) # 03:00 汇总指标 0 3 * * * bash run_summary.sh --date $(date -d yesterday %F)这些命令的意义是让每层任务之间有明确的前后置关系而不是把所有步骤塞进一个大脚本。参数--date表示业务日期保证重跑时始终处理的是指定分区避免“当前时间”这种隐式参数带来的不确定性。6.2 三步数据质量验证法我每次任务上线前都会写一个质量检查脚本只做三件事行数对比、主键唯一性、关键字段合理性。SELECT COUNT(*) AS total_cnt, COUNT(DISTINCT order_no) AS unique_cnt, SUM(CASE WHEN amount 0 THEN 1 ELSE 0 END) AS negative_amt_cnt FROM dwd_order WHERE dt 2024-06-01;逻辑说明:total_cnt和unique_cnt如果不相等说明有重复订单negative_amt_cnt如果大于 0说明金额清洗没有完全生效。这三个指标足够了它们能从整体上判断转换是否正确。再配合一个源端计数查询把源表按日分组的计数和目标表做差值差值在允许范围内就通过否则阻断下游任务。6.3 我习惯留的三个调试钩子长期维护的 ETL 最怕“黑匣子式”运行。我给自己提了三个硬性要求所有转换函数都支持dry_run参数只打印执行 SQL 不真正写入支持sample参数抽取前 100 行落到调试表失败数据自动落到_bad表而不是直接丢弃。def transform(sql, dry_runFalse, sample0): if dry_run: print(sql) return if sample 0: result query(sql f LIMIT {sample}) save_debug(result) return execute(sql)参数说明:dry_run用于联调时快速验证语法sample用于观察真实数据形态LIMIT只影响调试查询不影响正常任务。这三个钩子帮我解决过大量“为什么跑出来不对”的问题。我吃过最亏的一次是凌晨接到告警却无法快速定位是源数据变化还是转换逻辑变化后来强制在任务日志里打印“源表行数、清洗后行数、过滤行数”排查效率直接翻倍。希望这个习惯也能帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/13 16:31:15 · 阅读 1.2k ETL设计详解：数据抽取、清洗与转换的实践指南
简介《ETL设计详解数据抽取、清洗与转换》是一份面向BI项目实施人员、数据仓库开发工程师及数据入门学习者的Word文档。文档围绕ETL在BI项目中约占三分之一工作量的重要地位展开按数据抽取、数据清洗转换、数据加载三部分进行系统讲解其中抽取环节包含调研阶段需要确认的四个关键问题、同源/异源数据库及文件型数据源的处理方式清洗环节梳理了不完整、错误、重复三类脏数据的判别与过滤策略转换环节则涉及不一致数据转换、粒度聚合和业务规则计算加载部分还对比了OWB、DTS、SSIS等工具实现与SQL方式的适用场景并补充了ETL日志和警告发送的维护要点。资源包为1个docx文档大小仅20KB内容紧凑、要点齐备便于在方案设计或学习时快速翻阅。目前已有1534人学习适合作为数据仓库入门培训或项目选型的轻量参考。1. ETL 设计详解数据抽取、清洗与转换决定数仓的下限提到 ETL 设计详解数据抽取、清洗与转换很多人的第一反应是写几条insert into ... select ...把数据搬进数仓。真正把任务跑上生产三个月才会意识到 ETL 的难点不在写代码而在设计源系统变更时抽取不能断脏数据不能悄悄污染报表重跑任意一天都要得到一致结果。这篇文章把抽取、清洗、转换三个阶段拆开讲清楚从全量与增量选型、字段级清洗、粒度转换到调度重跑和验证给出一套可以直接照做的落地思路。2. 数据抽取全量增量与 CDC先定策略再写代码抽取是 ETL 的第一道关口也是最容易失控的环节。很多任务挂掉不是因为目标数仓不够强而是抽取方式没想清楚不分青红皂白全量拉表几亿行的事实表每次同步都让源库吃满 IO或者只用单条 SQL 不分页数据量一大就内存溢出。我的习惯是写任何抽取脚本之前先回答三个问题源表多大、更新频率多高、历史数据会不会被修改。回答完这三个问题方式基本就出来了。2.1 全量、增量和 CDC三种抽取方式怎么选三种方式的边界要分清楚不能只凭感觉。全量抽取适合维度表、配置表这类小表直接truncate后重灌简单且能保证目标与源完全一致增量抽取适合大表按自增主键或更新时间拉取新增数据对源库压力小CDC 则是在源库开启日志解析把插入、更新、删除都同步过来适合对实时性要求高、且源库允许开启日志的场景。抽取方式适用场景优点主要成本全量小维度表、配置表、慢速变化表实现简单数据一致性好每次全扫数据量大后扛不住增量主键/时间戳大流水表、append-only 表对源库压力小易实现断点续传无法可靠捕捉历史修改CDC需要同步增删改、实时性高的业务表完整记录变更延迟低依赖源库日志配置运维复杂度高对于没有专职 DBA、源库也不敢乱动的团队我一般推荐“小表全量大表增量”。增量字段优先选自增主键或update_time但要注意一个坑如果源系统把历史行的update_time一起刷掉增量会漏数。此时只能用 CDC 或每天对近 N 天数据做一次比对补偿具体做法在避坑章节展开。2.2 增量抽取的最小实现游标分页与增量水位先看一段常见的 Python 增量抽取骨架。它不停用游标也不是一次性把大结果集装进内存而是按 ID 逐批拉取拉完一批更新一次水位。def fetch_incremental(db, table, last_id, page_size5000): 按自增主键增量抽取 db: 数据库连接对象 table: 源表名 last_id: 上一批的最大主键第一次传 0 page_size: 每批行数 while True: rows db.execute( SELECT * FROM {table} WHERE id %(last_id)s ORDER BY id ASC LIMIT %(page_size)s, {last_id: last_id, page_size: page_size}, ) if not rows: break for row in rows: yield row # 把本批最后一行主键作为新水位 last_id rows[-1][id]逻辑说明:WHERE id last_id保证每一批都从上一批停下的地方继续ORDER BY id让排序稳定LIMIT控制单批内存。yield让调用方可以边拉边写目标表避免大量行积压在内存里。last_id这个水位既可以用在内存循环中也可以写到数据库表里任务中断后从库里读回来接着跑。参数说明:page_size建议从 5000 开始调。源库性能好、字段少可以加到 10000字段多或者源库是核心交易库降到 2000 更稳妥。还有一种常见做法是用WHERE update_time last_update如果时间是精确到秒的很容易漏掉同一秒内写入的数据所以要配合AND update_time next_boundary去重。2.3 断点续传与目标表增量水位断点续传不能只靠代码里的局部变量。程序一重启内存里的last_id就丢了所以要把水位落到持久化位置。最简单的方式是单独一张抽取状态表每次任务启动先读状态每批拉完就更新。-- 抽取状态表每个业务表一行 CREATE TABLE etl_state ( table_name VARCHAR(128) PRIMARY KEY, last_id BIGINT, last_update TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 每次抽取前读取水位 SELECT last_id, last_update FROM etl_state WHERE table_name ods_order;水位表的设计逻辑:last_id负责主键增量last_update负责下次任务开始的时间参考。我习惯把“读取水位”和“写完数据”放在同一个事务边界里避免数据写成功了但水位没更新导致下一轮重复抽取。如果你用的是自研调度平台把这个状态表当成普通目标表一样维护失败重跑时从状态表恢复即可。抽取还有一条红线: 永远不要用SELECT *去拉一张几百列的表。只选目标模型需要的列既减少网络传输也降低源库临时表空间消耗。给源库查询加SET TRANSACTION READ ONLY可以防止开发时手滑把更新语句写进抽取任务。3. 数据清洗把脏数据挡在事实表之前清洗是最看不出功绩、但最重要的环节。业务库里的数据往往带着各种历史包袱订单号有空格、金额字段混着货币符号、用户手机号有的带区号有的不带、同一个用户在两张表里主键不一致。清洗的目的不是把所有数据变成“标准答案”而是把质量基线定清楚保住核心指标的可信度。3.1 清洗规则先于代码字段质量基线我见过最典型的失败案例是一上来就写 Python 函数清洗数据结果清洗规则散落在脚本各处换了数据源就找不到哪条规则在哪生效。正确做法是先维护一张字段质量基线表把每个字段的规则、告警阈值、处理动作写出来。字段质量规则异常处理动作告警阈值订单号非空、长度 8-32 位、无首尾空格丢弃并记录_bad表缺失率 0.1%金额数字类型、大于 0非法值置 NULL参与后续 COALESCE负值条数 0手机号11 位数字统一去除非数字字符位数不符置空空号率 5%订单时间能解析为日期、不晚于当前时间解析失败的落_bad表失败率 0.05%这条基表的用处是让清洗变得可审计。开发时按规则实现上线后按阈值告警而不是等业务方发现报表数字诡异才回头查。基线表本身可以用一份简单的配置或 SQL 表维护不需要引入复杂规则引擎。3.2 空值和重复SQL 去重与 Python 清洗清洗最常处理的两类问题是空值和重复。先说重复同一个业务主键出现多行往往是因为源表有多条更新记录。此时最稳的做法是用窗口函数保留最新一条WITH deduped AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY order_no ORDER BY updated_at DESC ) AS rn FROM raw_order ) SELECT * FROM deduped WHERE rn 1;逻辑说明:PARTITION BY order_no界定重复范围ORDER BY updated_at DESC指定保留最新更新。如果源表没有更新时间就用自增主键倒序排。注意rn 1会过滤掉重复的旧数据清洗结果行数一定小于等于原表下游如果发现指标少了很多要能接受这个预期。Python 侧处理适合实验性分析或字段格式不稳定的场景。我一般用 pandas 做探查再用 DataFrame 清洗逻辑固化成任务import pandas as pd df pd.read_csv(raw_order.csv, dtype{order_no: string}) # 去空格、统一手机号字母形式 df[order_no] df[order_no].str.strip() df[phone] df[phone].str.replace(r\D, , regexTrue) # 按业务主键去重保留最后一条 df df.drop_duplicates(subset[order_no], keeplast) # 金额转数值解析失败置空再补默认值 df[amount] pd.to_numeric(df[amount], errorscoerce).fillna(0)逻辑说明: 先做格式规整再做去重是因为原始数据可能同一个手机号一个带空格一个不带直接做主键去重会漏掉真正重复的行。errorscoerce会把无法解析的金额变成NaN再用fillna(0)兜底。参数说明:keeplast表示保留最后一条如果你更信任先发生的数据改成first。subset必须是真正的业务主键不要用全字段去重否则源表一个字段的微小差异就会让重复行全部保留。3.3 字符编码和时区两个最容易翻车的清洗项字符类问题很隐蔽。业务库写入时用的是应用侧编码到了数仓可能变成乱码尤其常见的是中文环境下出现utf8和latin1混用。表象是主键冲突、字符串比较永远不相等。处理方式是在抽取阶段就统一转成 UTF-8到了清洗阶段再发现乱码往往已经丢了原始信息很难逆推。时区问题则是指标对不齐的常见来源。如果订单创建时间存的是 UTC日报却按东八区汇总每天凌晨 0 到 8 点的单子会被分到前一天。统一做法是让数仓所有表都存 UTC展示层再转目标时区如果源系统已经存了本地时间必须在抽取时记录时区偏移量不能靠猜。from datetime import datetime, timezone, timedelta raw_ts 2024-06-01T12:00:00Z # 解析 UTC 时间 dt datetime.fromisoformat(raw_ts.replace(Z, 00:00)) # 转成目标偏移时区 local_dt dt.astimezone(timezone(timedelta(hours8))) print(local_dt.strftime(%Y-%m-%d %H:%M:%S))代码逻辑是先把字符串末尾的Z改成00:00让fromisoformat识别为标准 UTC 时间再用astimezone做偏移转换。这里的参数timedelta(hours8)就是目标时区相对 UTC 的小时数换成其他业务时区就改这个值不要在代码里写死多个字符串格式。最重要的是清洗阶段做完时间统一后下游所有任务都不得再自行换算否则一处统一一处乱排查起来非常痛苦。4. 数据转换从字段映射到维度建模转换阶段解决的核心问题是把源系统面向业务操作的表结构改成面向分析的目标结构。这里涉及到字段名统一、类型对齐、粒度选择、维度补充。很多人把转换理解为写长 SQL 把表 join 起来但对生产环境真正的难点是你怎么保证 join 之后没放大数据、没丢数据以及重跑时结果不变。4.1 字段映射与类型转换源库的主键可能是字符串目标数仓要求 bigint源库的金额可能带千分位目标模型需要数值类型。这些转换看着简单却最容易因为脏数据跑挂整个任务。建议在映射阶段做一步“显式 cast 错误隔离”。SELECT order_no, SAFE_CAST(user_id AS BIGINT) AS user_key, TRIM(COALESCE(NULLIF(phone, ), 0)) AS phone, SAFE_CAST(REGEXP_REPLACE(amount, ,, ) AS DECIMAL(12, 2)) AS amount FROM stage_order;逻辑说明:SAFE_CAST是先把类型转换做成安全函数遇到非法字符返回NULL不会让整个任务报错NULLIF(phone, )把空字符串变成NULL再配合COALESCE给一个占位值避免下游 join 出现莫名其妙的空值。REGEXP_REPLACE去掉金额里的千分位逗号再做数值转换。参数说明:DECIMAL(12,2)表示最大 12 位数字、2 位小数在金额超过亿级时改大精度否则会因溢出而失败。这里的关键点是转换失败不能静默。用SAFE_CAST后要把原本会失败的行单独抽样记录例如记录user_id非法的行数和样例方便后续反向修数。直接不加保护地CAST一个坏数据能让整批任务重跑半小时这种奶酪式错误在真实生产里见得太多了。4.2 粒度转换明细到汇总先定粒度再写 GROUP BY粒度决定了每一行代表什么也决定了下游指标口径。从订单明细转换到每日商品汇总一个常见错误是先 join 再聚合结果 join 产生的重复行导致销量翻倍。正确做法是先想清楚“最终一行是什么”再倒推需要哪些维度字段。SELECT DATE(order_time) AS stat_date, product_id, COUNT(*) AS order_cnt, SUM(order_qty) AS qty, COUNT(DISTINCT user_id) AS uv FROM dwd_order_detail WHERE order_time 2024-06-01 AND order_time 2024-06-02 GROUP BY DATE(order_time), product_id;逻辑说明:DATE(order_time)把订单时间落到天级粒度product_id是第二个分组维度所以最终每个分区是“某天 某商品”一条记录。COUNT(DISTINCT user_id)用于统计去重用户数不能直接COUNT(user_id)否则一个用户下多单会被多算。参数说明: 时间过滤采用左闭右开即 开始 AND 次日这样不会漏掉 23:59:59.999 的数据也方便后续做增量。转换时还要防“join 放大”。一个常见自检方法是join 前先分别统计左右表的行数再统计 join 后行数如果 join 后行数明显大于左表说明右表存在一对多关系需要先对右表做去重或在 join 条件里补维度。这个操作看起来笨但这几分钟能避免一次数据质量事故。4.3 维表补充与缓慢变化维的取舍事实表通常只存业务键比如user_id、product_id分析时需要补齐用户等级、商品类目等描述字段。最稳妥的维表补充是时间快照关联事实表下单时间和维表生效日期匹配尽量不直接关联当前维表否则历史订单会跟着维表变化历史汇总口径全变。SELECT f.order_no, d.user_level, d.user_region FROM dwd_order_fact f LEFT JOIN dim_user_snapshot d ON f.user_id d.user_id AND f.dt d.effective_date AND f.dt d.expire_date;这段 SQL 用了拉链表的时间段关联可以精确还原每个用户在某一天当时的等级。参数说明:effective_date和expire_date是拉链表的生效区间f.dt必须是分区日期否则范围关联会扫描全表。如果维表不需要历史回溯直接LEFT JOIN当前版本表即可但你要能接受历史订单的用户等级是当前值。缓慢变化维的取舍要看分析需求。报表需要“上个月下单时用户是什么等级”必须保留历史版本也就是 SCD2 拉链表报表只关心当前等级用 SCD1 覆盖更新即可。我的建议是不要一上来就建拉链表先问业务方三个问题看历史吗、统计口径变过吗、能接受当前值吗。都回答“要”再上 SCD2否则维护成本远大于收益。5. ETL 避坑指南我跑挂过的 5 个真实问题这一章的每条经验都来自线上事故现象、原因、解决方式写得直白一点希望能帮你少踩一次坑。ETL 上线后最可怕的不是任务失败而是任务显示成功但数据是错的。5.1 抽取把源库拖垮一跑全表扫描锁竞争严重现象某次凌晨 2 点运行增量抽取后业务系统第二天早晨反馈核心交易变慢数据库监控显示大量SELECT长时间占用 IO。原因新接手的 A 同学给一张 2 亿行的订单流水表做了全量抽取且没有分页查询直接把源库临时表空间打满。解决把全量改成增量用modified_time 上次水位拉取如果业务必须用全量则拆成 2000 行一批的分页任务并把运行时间放在业务低峰期。另一个有效手段是给查询加/* */并行 hint 限制但不要依赖这个优先从数据量上控制。5.2 空值被“清洗”过滤指标少了一块现象销售日报中某个大区 GMV 突然下降 12%但任务状态全部成功。原因清洗阶段用了WHERE user_id IS NOT NULL过滤行但很多订单的user_id在源系统确实为空属于匿名下单应该保留条数但把用户维度设为unknown而不是整行丢弃。解决清洗规则中区分“非法值”和“缺失值”缺失值保留非法值落_bad表。给出的建议是每一个过滤条件都反查一下被过滤的数据量如果预期过滤率超过 1%必有问题。从此我把所有清洗脚本都加了“过滤行数统计日志”。5.3 字符集不一致主键冲突离奇出现现象目标表报主键冲突但抽样看每条订单号都不重复。原因源表是 Ubuntu 下写入的表情符号目标库按 Latin1 解析把同一个订单号转成了不同的乱码导致实际字节相同但显示不同数据库层面则判断为重复。解决在 JDBC/连接串中显式指定utf8mb4并在抽取后对order_no做一次十六进制校验。最简单的办法是每次同步后跑一遍SELECT COUNT(DISTINCT order_no) vs COUNT(*)不一致就立刻告警不能等到跑数仓模型时才发现。5.4 重跑任务不幂等同一份订单被记两次现象某天调度任务因网络超时挂了运维手动重跑结果月报订单量几乎翻倍。原因第一次任务实际已经写入了数据只是在返回成功前连接断开重跑时没有先清理目标分区导致相同主键的数据写入两次。解决给所有目标表加“重跑前清理”逻辑比如DELETE WHERE dt 2024-06-01或直接删分区。这个习惯非常重要。凡是支持重跑的任务必须做到“同一参数、同一结果”否则数据质量根本无法保证。我现在把幂等性当成代码评审的一个硬指标。5.5 只看成功失败行数腰斩却没告警现象任务状态绿色但当天数据行数从 200 万跌到 90 万直到业务方看报表才暴露。原因源系统发布新版本后接口返回格式变化抽取程序解析 JSON 少了一层字段但语法没有报错空数据被当成正常结果写入。解决在所有抽取和转换任务里加行数波动检查和近 7 天平均值对比波动超过 20% 就发告警。具体实现可以是一个简单的 SQL 检查也可以是调度平台的传感器。只要做到这一步数据量暴跌就不会过夜才发现。6. 把 ETL 设计成能长期维护的流水线前三章解决的是单任务怎么写最后一章解决的是整套流程怎么维护。ETL 不是跑起来就结束的代码它会被调、被改、被重跑所以分层、依赖验证和调试手段必须一开始就设计好。6.1 任务分层与依赖配置我习惯把任务按 ODS、DWD、DM 三层拆开ODS 只做抽取和原样落地DWD 做清洗转换DM 做汇总输出。上层任务不直接依赖源库只依赖下一层的结果表这样源库变更时影响范围能被隔离在某一层。调度上可以用 cron 设置“先抽取后转换再汇总”的依赖顺序。一个最小 crontab 配置可以是# 每天凌晨 01:00 抽取原始数据 0 1 * * * bash run_extract.sh --date $(date -d yesterday %F) # 02:00 清洗转换 0 2 * * * bash run_transform.sh --date $(date -d yesterday %F) # 03:00 汇总指标 0 3 * * * bash run_summary.sh --date $(date -d yesterday %F)这些命令的意义是让每层任务之间有明确的前后置关系而不是把所有步骤塞进一个大脚本。参数--date表示业务日期保证重跑时始终处理的是指定分区避免“当前时间”这种隐式参数带来的不确定性。6.2 三步数据质量验证法我每次任务上线前都会写一个质量检查脚本只做三件事行数对比、主键唯一性、关键字段合理性。SELECT COUNT(*) AS total_cnt, COUNT(DISTINCT order_no) AS unique_cnt, SUM(CASE WHEN amount 0 THEN 1 ELSE 0 END) AS negative_amt_cnt FROM dwd_order WHERE dt 2024-06-01;逻辑说明:total_cnt和unique_cnt如果不相等说明有重复订单negative_amt_cnt如果大于 0说明金额清洗没有完全生效。这三个指标足够了它们能从整体上判断转换是否正确。再配合一个源端计数查询把源表按日分组的计数和目标表做差值差值在允许范围内就通过否则阻断下游任务。6.3 我习惯留的三个调试钩子长期维护的 ETL 最怕“黑匣子式”运行。我给自己提了三个硬性要求所有转换函数都支持dry_run参数只打印执行 SQL 不真正写入支持sample参数抽取前 100 行落到调试表失败数据自动落到_bad表而不是直接丢弃。def transform(sql, dry_runFalse, sample0): if dry_run: print(sql) return if sample 0: result query(sql f LIMIT {sample}) save_debug(result) return execute(sql)参数说明:dry_run用于联调时快速验证语法sample用于观察真实数据形态LIMIT只影响调试查询不影响正常任务。这三个钩子帮我解决过大量“为什么跑出来不对”的问题。我吃过最亏的一次是凌晨接到告警却无法快速定位是源数据变化还是转换逻辑变化后来强制在任务日志里打印“源表行数、清洗后行数、过滤行数”排查效率直接翻倍。希望这个习惯也能帮到你。本文还有配套的精品资源点击获取