简介：围绕 ETL 数据抽取、清洗转换与加载的《ETL设计详解（数据抽取、清洗与转换）》，面向数据仓库与 BI 项目初学者、ETL 开发及实施人员，帮助梳理从业务系统调研、数据源识别到数据入库的关键设计问题。资源包共 1 个 docx 文件，约 20KB，以文档形式集中整理 ETL 实施要点，便于按章节查阅和项目对照。已有 1538 人学习下载，可作为入门框架梳理或方案设计参考。文档覆盖同构/异构数据源抽取、文件类数据导入、增量更新判断，以及不完整、错误、重复数据的清洗策略和不一致数据整合、粒度转换、商务规则计算；同时讨论 ETL 工具与 SQL 结合实现、日志记录和警告发送等运维细节。文档还强调 ETL 往往占 BI 项目约三分之一时间，其中清洗转换工作量较大，提醒在规则验证和用户确认上投入精力，有助于建立从 ODS 到 DW 的完整处理思路并积累排错优化经验。

1. 从 12 张看板说起：为什么 ETL 设计必须先立住再动手

接过一个给某公司做经营分析看板的活儿，业务方要的报表只有 12 张，看着开发量不大，排期一摊开却发现 ETL 的数据抽取、清洗转换、加载三段占掉了整整三分之一工期。真正拖时间的不是写查询语句，而是源库里客户名称缺一半、日期字段混着两种写法、同一个供应商在三套系统里有三个编码。ETL 设计做不好，后面的看板都是沙上建塔；这也是为什么这类 BI 项目里，清洗转换（T）往往单独吃掉三分之二的工作量。

下面这份 ETL 设计方案按抽取、清洗转换、加载三段拆开讲，从调研阶段该问哪些问题，到分层怎么划、增量水位怎么存、日志表怎么建，最后给出一份能被追问的验收清单。做课设和毕设的同学可以把业务字段换掉直接套结构，已经在岗的可以拿去对标手头那批脚本，看看哪一层是缺的。

2. 需求调研与总体架构：两层存储与四个模块的职责划分

2.1 调研阶段必须先落纸的六个问题

数据抽取不是拿到连接串就开始写 select，调研没做透，后面每一次返工都要重跑全量。我带项目时会先把这六个问题写成一张调研表，让业务方和各系统负责人签字，后面扯皮的时候这张表就是依据。

第一个问题：数据来自几个业务系统，每个系统归哪个部门管，谁有权确认字段口径。这个问题决定了后面异常数据该发给谁，也决定了清洗规则由谁拍板。

第二个问题：各源库的数据库类型、版本、字符集和时区。字符集不一致会在中文字段上出乱码，时区不一致会让「昨天」的增量跑出两天的数据，这两类问题在联调阶段极难定位。

第三个问题：单表数据量和日增量。十万行和五千万行的抽取策略完全不同，前者可以每晚全量覆盖，后者必须做增量加分区。

第四个问题：是否存在手工数据，格式是什么，多久更新一次，谁在维护。手工 Excel 是脏数据的主要来源，通常的做法是约定一个固定目录和固定表头，由程序定时扫目录入库，而不是让人直接往库里贴。

第五个问题：是否存在非结构化或半结构化数据，比如日志文件、JSON 报文、附件。这类数据要么在抽取阶段就拍平成宽表，要么单独建贴源表，不要混进业务明细。

第六个问题：业务库有没有可靠的更新时间字段，有没有物理删除。没有更新时间就只能靠自增主键或全量比对，有物理删除就必须补一个删除标记同步机制，否则数据仓库里的数据只增不减，越跑越错。

这六个问题问完，抽取方式、增量策略、清洗规则的轮廓基本就出来了。

2.2 分层架构：业务库到 ODS 再到 DW 的职责边界

整体链路是四段：业务系统库 → 抽取层 → ODS（贴源数据层）→ DW（数据仓库层）→ 应用与报表。划分原则只有一条：每一层只干自己那件事，出错时能一眼看出是哪一层的问题。

抽取层负责连接和搬运，不做业务判断，只做必要的类型转换。它的产出必须带三个审计字段：etl_batch_id（批次号）、etl_load_time（落库时间）、etl_src_system（来源系统标识）。少了这三个字段，出了错就只能靠猜。

ODS 层我一般会落两份：一份贴源原样表，字段名和源库保持一致，只加审计字段，用于回溯和重跑；一份清洗后的标准表，脏数据已经被分流到拒绝表，字段类型和编码已经统一。贴源原样表看着冗余，但它是唯一的后悔药——清洗规则写错了、映射表配错了，能靠它原地重算，不用回头再骚扰业务库。

DW 层分明细和汇总两级。明细层做不一致数据的统一和轻度聚合，汇总层落商务规则算出来的指标。业务系统存的是非常细的流水，数据仓库是拿来分析的，不需要细到每一笔，聚合放在这一层做，报表层就只负责取数和展示。

命名上统一约定：贴源表ods_源系统_表名，清洗表ods_std_表名，明细表dwd_主题_表名，汇总表dws_主题_表名，拒绝表统一rej_表名。这套约定不写进文档，三个月后接手的人就得反过来读脚本猜。

2.3 工具、SQL、混合三条路线怎么选

ETL 的实现常见三条路：纯商业 ETL 工具、纯 SQL（含存储过程与脚本）、工具加 SQL 混合。三条路没有绝对优劣，关键看团队里谁能维护。

纯工具的问题是复杂商务规则一旦落到组件里，调试和版本管理都很别扭，改一个指标要翻三层映射；纯 SQL 的问题是作业数量上来以后，几百个脚本没有统一调度和日志，排查全靠人肉。我一般选混合：调度、依赖、告警交给统一的调度框架，抽取和加载用脚本封装成模板，中间的清洗转换逻辑用 SQL 写，复杂规则（比如多表匹配打分、模糊去重）用 Python 单独写成可测试的函数。这样既保住了灵活性，又不至于让每个作业各自为政。

3. 数据抽取设计：四类数据源与一条水位线的落地方式

3.1 四类数据源与对应的抽取方式

按源系统的形态分四类，每类的抽法不一样，混着用一种方法写，早晚踩坑。

同构库最省事，直接在数据仓库服务器和业务系统之间建立只读链接，写查询就能访问。这里有一条血泪经验：抽取账号必须是只读的，曾经有人图省事用了业务账号，一个写错的 update 直接改到了生产库，虽然后来回滚了，但那晚的应急会开到凌晨两点。

异构库优先走 ODBC 或 JDBC，实在连不通就退化成文件中转：源端导出 txt 或 xls，程序定时扫指定目录入库。文件方案看着土，但稳定，前提是文件命名和表头必须按约定来，最好在入库前先做一次结构校验，表头对不上直接告警而不是硬塞。

文件类数据源还有一个现实问题：手工数据由业务人员维护。与其反复催他们导数据，不如培训他们用固定模板，程序按模板解析，解析失败的文件挪到异常目录并通知本人，形成闭环。

接口类要处理分页和超时。我一般把接口返回的原始报文整段落一份贴源表，再解析成宽表，这样接口字段变了还能回溯原始报文。

3.2 增量抽取：水位表设计与时间戳的坑

数据量一大就必须做增量。最常见也最好用的方案是时间戳增量：业务系统每行记录带更新时间，每次抽取前先取水位表里记录的最大时间，然后去源库取大于该时间的所有记录。

水位不能存在脚本变量里，必须落表，否则作业一重启就从头再来。水位表结构如下：

-- 增量水位表：每个作业一行，记录已经成功加载到的位置 create table etl_watermark ( job_name varchar(64) primary key, -- 作业名，如 ods_customer last_value timestamp not null, -- 已成功加载的最大更新时间 last_batch_id varchar(32) not null, -- 推进水位的批次号，便于对账 update_time timestamp not null default current_timestamp );

关键点在于水位推进的时机：必须在数据成功写入目标表之后，和加载操作放在同一个事务里提交。如果抽完就推进水位，加载失败重跑时这一段数据就永久丢了，这类问题在日志里看不出来，只能靠对账发现。

时间戳方案有三个已知问题。一是源库时间戳可能不更新，某些系统在批量修数据时绕过了应用层，更新时间没变，数据却变了，这种只能靠定期全量比对兜底。二是同一时间戳的多条记录会被截断，取数时要用「大于水位且小于等于当前批次启动时间」的闭区间，并把批次启动时间固定下来，不要每次查询都用当前时间。三是物理删除，时间戳方案天然抽不到删除，通常的做法是每晚跑一次主键比对，把目标表里源库已不存在的记录打上删除标记，而不是直接物理删除。

对于没有时间戳的表，退而求其次用自增主键做增量，或者对小表直接全量覆盖。全量覆盖只适合小表和维表，千万级事实表上做全量覆盖，跑一次就是一次事故。

3.3 抽取脚本落地：带水位推进和失败回滚的 Python 实现

下面这段脚本是一个最小可用的抽取模板，把连接、水位读取、分批取数、加载、水位推进串成一个事务链。实际项目里换掉表名和字段即可。

# extract_job.py —— 时间戳增量抽取模板 import datetime import pandas as pd from sqlalchemy import create_engine, text # 1. 源库只读账号，目标库读写账号，连接串从配置中心读取，不硬编码 SRC = create_engine("postgresql+psycopg2://etl_ro:***@src-host:5432/biz") DW = create_engine("postgresql+psycopg2://etl_rw:***@dw-host:5432/dw") JOB = "ods_customer" BATCH = datetime.datetime.now().strftime("%Y%m%d%H%M%S") # 批次启动时间固定下来，作为本次增量的上界，避免跑批期间新数据混入 UPPER = datetime.datetime.now().replace(microsecond=0) def read_watermark(conn): row = conn.execute( text("select last_value from etl_watermark where job_name = :j"), {"j": JOB}, ).fetchone() # 首次运行没有水位，从 1970 年开始 return row[0] if row else datetime.datetime(1970, 1, 1) def extract(): # 先读水位，再用同一个上界去源库取数 with DW.connect() as dw_conn: lower = read_watermark(dw_conn) sql = text(""" select customer_id, customer_name, region_code, supplier_code, status, updated_at from customer where updated_at > :lower and updated_at <= :upper """) df = pd.read_sql(sql, SRC, params={"lower": lower, "upper": UPPER}) if df.empty: print(f"[{JOB}] 无增量数据，lower={lower}") return # 补齐审计字段，任何一张 ODS 表都必须带这三个 df["etl_batch_id"] = BATCH df["etl_load_time"] = datetime.datetime.now() df["etl_src_system"] = "biz" # 加载与水位推进放在同一个事务里提交 with DW.begin() as dw_conn: df.to_sql("ods_biz_customer", dw_conn, if_exists="append", index=False) dw_conn.execute( text(""" insert into etl_watermark(job_name, last_value, last_batch_id) values (:j, :v, :b) on conflict (job_name) do update set last_value = excluded.last_value, last_batch_id = excluded.last_batch_id, update_time = current_timestamp """), {"j": JOB, "v": UPPER, "b": BATCH}, ) if __name__ == "__main__": extract()

几个参数要按实际情况改：UPPER用批次启动时间而不是查询时刻，是为了保证重跑时区间一致；if_exists="append"只适合目标表已建好的情况，如果表由程序自动创建，字段类型可能被推断错，通常的做法是提前用 DDL 建表、程序只追加；水位推进的on conflict是 PostgreSQL 写法，换成其他库改写成先 update 再判断影响行数、为 0 则 insert 即可。

跑失败时看三处：源库连接是否超时（看异常堆栈的第一层）、目标表字段是否和 DataFrame 对不上（看类型转换报错）、水位表里的last_value是否比源库最大值还大（水位被错误推进，需要手工回退到上一批次的last_value）。

4. 清洗转换与加载：三类脏数据的分流和三种转换的写法

4.1 三类脏数据的分流处理

清洗的任务是把不符合要求的数据过滤出来，交给业务主管部门确认是丢弃还是修正后再抽。脏数据分三类，处理动作完全不同，一律丢进拒绝表是不负责任的，因为有些数据修一修还能用。

不完整的数据要按缺失内容分文件导出，而不是混在一个大表里，业务方拿到一份「区域信息缺失清单」和一份「供应商编码缺失清单」，处理效率完全不同。

错误的数据里有一小类必须特殊对待：日期格式不正确或日期越界的记录会让整个作业直接失败。这类错误要在抽取前用单独一条校验 SQL 挑出来，交给源系统限期修正，修正之后再抽；不能等到进主流程才崩，那样整个批次都要重跑。

重复的数据在维表里最常见，通常是源系统没做主键约束。处理方式是把重复记录的所有字段导出来让人确认，业务确认保留哪条之后再写去重规则，不要自己拍脑袋留最后一条。

清洗是一个反复的过程，不可能几天内完成。我在项目初期会配置每天的过滤数据邮件，把当天被拒绝的记录数按规则汇总发给业务方，一方面催他们尽快修正，另一方面这份邮件本身就是后期验证数据完整性的依据。

提示：每一条过滤规则上线前都要做反向验证——拿一批已知有效的数据跑一遍，确认没有被误杀。误杀有效数据比放过脏数据更麻烦，因为没人会主动去查少掉的那部分。

4.2 三类转换：编码统一、粒度聚合、指标计算

转换在 ODS 到 DW 的过程中做，主要三类。

第一类是不一致数据转换，本质是整合。同一个供应商在结算系统里编码是 XX0001，在客户系统里是 YY0001，抽过来之后必须映射成一个统一编码。做法是建一张映射维表，字段包括源系统标识、源编码、统一编码、生效时间，转换时左连接这张表。映射表要有人维护，并且要有未匹配上的兜底逻辑——匹配不到的统一编码不能直接置空，应该落到一个「未知」默认值并单独告警，否则汇总时数据凭空少一块。

第二类是粒度转换。业务系统存的是明细流水，数据仓库按分析粒度聚合。比如订单明细按天、按区域、按品类汇总成日汇总表：

-- 粒度转换：明细流水按天+区域聚合到 DW 汇总层 insert into dws_order_daily(stat_date, region_code, order_cnt, order_amt) select date(create_time) as stat_date, region_code, count(distinct order_id) as order_cnt, sum(pay_amount) as order_amt from dwd_order_detail where etl_batch_id = :batch_id -- 只算本批次，避免重复累加 group by date(create_time), region_code;

这里的etl_batch_id过滤条件不能省。汇总表如果不是每天全量重算，而是按批次追加，少了这个条件就会把历史批次重复算一遍，金额翻倍而且很难发现。

第三类是商务规则计算。不同企业指标口径不一样，很多指标不是简单加减，比如客户活跃度要综合近 30 天登录次数、下单次数、客单价加权；毛利率要把折扣、退货、运费都摊进去。这类逻辑我一般写成独立的 SQL 或 Python 函数，落在 DW 汇总层，报表层直接取，绝不在报表工具里重算一遍——两处各算一遍是口径不一致的头号原因。

4.3 清洗规则的配置化：规则表加校验 SQL

规则如果硬编码在脚本里，每加一条就要改代码发版，维护成本高。比较稳妥的做法是把规则落成表，脚本读规则动态生成校验语句。

-- 清洗规则配置表：一条规则一行，脚本按表驱动执行 create table etl_clean_rule ( rule_id serial primary key, target_table varchar(64) not null, -- 作用的表，如 ods_biz_customer column_name varchar(64) not null, -- 作用的字段 rule_type varchar(16) not null, -- not_null / regex / enum / range / unique rule_expr varchar(256), -- 正则、枚举值列表或数值区间 action varchar(16) not null, -- reject 拒绝 / fix 修复 / warn 仅告警 owner_dept varchar(64), -- 异常数据的责任部门 enabled boolean default true );

脚本按rule_type拼出对应的校验 SQL，命中的行按action分流：reject写入rej_前缀的拒绝表，fix用rule_expr里定义的替换逻辑原地修正（比如去首尾空格、全角转半角），warn只计数并进告警汇总。规则表加上owner_dept字段之后，每天的拒绝数据能自动按部门分组发邮件，省掉大量人工分发。

这套机制的好处是清洗规则可审计：某条数据为什么被过滤，查规则表就知道是哪条规则、什么条件、谁负责的。

4.4 加载策略与三类日志表

加载分三种策略，按表选用：

维表用 Upsert 最稳，写法如下：

-- 维表加载：主键冲突则更新，否则插入 insert into dwd_customer(customer_id, customer_name, region_code, etl_batch_id) values (:customer_id, :customer_name, :region_code, :batch_id) on conflict (customer_id) do update set customer_name = excluded.customer_name, region_code = excluded.region_code, etl_batch_id = excluded.etl_batch_id;

日志是 ETL 的黑匣子，分三张表记：批次总表、步骤明细表、错误表。

-- 1) 批次总日志：一次作业运行一行 create table etl_job_log ( batch_id varchar(32) primary key, job_name varchar(64), start_time timestamp, end_time timestamp, status varchar(16), -- running / success / failed row_count bigint -- 本批次影响行数 ); -- 2) 步骤日志：批次内每一步一行，记录起止时间与影响行数 create table etl_step_log ( id bigserial primary key, batch_id varchar(32), step_name varchar(128), start_time timestamp, end_time timestamp, row_count bigint, status varchar(16) ); -- 3) 错误日志：出错时记录模块、错误码和错误信息 create table etl_error_log ( id bigserial primary key, batch_id varchar(32), step_name varchar(128), error_time timestamp default current_timestamp, error_code varchar(32), error_msg text );

执行过程日志是流水账，用来回答「这一步跑了多久、处理了多少行」；错误日志回答「错在哪、为什么错」；批次总日志回答「昨晚这一批到底成没成」。三张表缺一张，排查时就得多花半小时翻应用日志。

告警走邮件或企业消息通道，内容至少要包含作业名、批次号、失败步骤、错误摘要，附上批次号之后运维可以直接拿它去查三张日志表，不用来回问「是哪个批次」。

5. 避坑与排查：五个让 ETL 半夜翻车的点

5.1 全量重跑把 ODS 冲了

现象：为了修一个字段，手工重跑了某个作业，第二天发现 ODS 表里只剩当天的数据，历史全没。

原因：作业写成了先truncate再insert，全量覆盖的语义被用在了本该增量追加的表上。

解决：按表定策略并写进表级配置，事实表与流水表一律只追加；确实需要重跑时，用批次号做条件删除（delete from ... where etl_batch_id = :batch_id）再重灌，而不是清表。Truncate 权限单独收口，不放在日常调度账号里。

5.2 时间戳增量漏数据

现象：对账时发现某天的订单数比源系统少，但作业日志全是成功。

原因：源系统批量修数时绕过了应用层，updated_at没更新；或者抽取用了「大于水位」的开区间，边界上同一时间戳的多条记录被截断。

解决：抽取区间改成「大于水位且小于等于批次启动时间」，批次启动时间在作业开始时固定；对关键表每天跑一次主键比对，把新增和缺失的行数记进对账表，数量对不上就告警。对账比日志更能发现问题，因为日志只证明作业跑完了，不证明数据齐了。

5.3 清洗过滤把有效数据干掉

现象：报表某个区域的数据突然少了一大截，业务方找上门。

原因：新加的一条正则规则写得太严，把带括号的客户名称全判成了非法字符，直接进了拒绝表。

解决：每条规则上线前用一批已知有效数据做反向验证，确认误杀率为零；规则表加enabled开关，新规则先以warn模式跑三天，只统计不拦截，观察命中量正常后再切reject。拒绝表每天要有数量趋势，某天突然涨十倍，基本就是规则写错了。

5.4 日志只写成功不写行数

现象：作业显示成功，但下游报表数据明显偏少，谁也说不清是哪一步少的。

原因：日志只记了状态，没记每一步的影响行数，无法定位是抽取少抽了还是转换过滤多了。

解决：步骤日志强制记录row_count，并在关键节点之间做数量校验：抽取行数与加载行数之差应该等于拒绝表增加的行数，对不上就判定异常。这条校验加上去之后，大部分数据缺失问题在当天就能定位到具体步骤。

5.5 指标口径在 ETL 和报表里各算一遍

现象：同一张毛利率报表，两个页面数字不一样，业务方两边都截图来问。

原因：ETL 在汇总层算了一遍毛利率，报表工具里又用明细字段算了一遍，两边对折扣和退货的处理规则不同。

解决：口径只允许有一处实现，统一落在 DW 汇总层，报表层只做筛选和展示；指标定义写成文档，注明分子分母、过滤条件、时间口径，任何人要改口径必须改 ETL 并同步更新文档。这条规则执行起来有点笨，但它是唯一能让数字长期对得上的办法。

6. 验收与答辩：一份能被追问的 ETL 交付清单

交付前我会走一遍固定清单，这份清单同时也是答辩时最容易被追问的点，逐条准备好答案，现场基本不会被问住。

答辩时最常被问的三个问题，我一般提前准备好答案。第一个是「为什么选增量而不是全量」，答案要落到数据量和调度窗口上，而不是「大家都这么做」。第二个是「脏数据怎么保证没被误杀」，答案就是规则表的 warn 模式和反向验证记录。第三个是「作业失败了怎么办」，答案要具体到批次号、日志表、重跑命令三步，能当场演示最好。

还有一个细节值得单独准备：把对账结果做成一张每天自动生成的表，行数差异、拒绝量、耗时趋势都在里面。答辩时打开这张表，比讲十页架构图都管用，因为它证明这套 ETL 是真的在跑，而不是画出来的。

从那以后我每次交付 ETL 相关的东西，都强制走一遍「人为制造一次失败，看日志和告警能不能定位」这个动作，很多问题只有真的失败一次才会暴露。希望这套结构和清单能帮到你，把手上那份方案补成一个能跑、能查、能验收的完整工程。

本文还有配套的精品资源，点击获取