简介：这份文档面向BI项目中的数据仓库建设者与ETL初学者，系统讲解数据抽取、清洗转换与加载的完整设计思路，帮助读者理清ETL在项目中的定位与实施要点。内容围绕数据抽取阶段的数据源调研问题、同构与异构数据库及文件型数据源的抽取方法、增量更新策略展开，并详细说明不完整数据、错误数据与重复数据三类清洗对象及对应处理方式，同时涵盖数据转换中的编码统一、粒度聚合与商务规则计算，以及ETL日志分类与警告发送机制。资源包共1个docx文件，约20KB，以文字讲解为主，结构按抽取、清洗转换、加载三部分组织，便于按模块查阅。已有1536人学习，适合需要快速建立ETL设计框架、对照实际项目梳理流程的数据开发与BI实施人员参考。

1. 从一次数据对不上账说起：ETL 方案到底要解决什么

业务方拿着报表问「为什么这个月销售额和财务系统差了三万」，你打开数据库一看，源系统里客户区域字段是空的、日期存成了全角字符、同一个供应商在结算系统和 CRM 里挂着两个编码。这类问题不是 SQL 写得好不好，而是整条数据链路没有设计。ETL 设计详解要解决的就是这件事：把分散在多个业务系统、文件、手工表里的数据，按一套可复现的规则抽到 ODS，清洗转换后加载进 DW，并且让每一步都能查、能重跑、能追责。这套方案适合正在做 BI 课设、数据仓库毕设的学生，也适合要交付一份数据集成方案的在职工程师。通常 ETL 会吃掉整个项目三分之一的时间，其中清洗转换又占掉这三分之一里的三分之二，先把结构立住，后面才不至于天天救火。

2. 需求调研与总体架构：三库分层和四个必须先答的问题

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

动手写第一行抽取脚本之前，调研结论要能直接决定后面的技术选型。我一般会强制把下面四个问题写成一张表，答不上来就不进入设计：

这四个问题答完，抽取方式基本就定了。很多方案翻车不是技术不行，是调研阶段漏了「手工数据」这一项，上线后才发现每月还有一批 Excel 要人肉导。

2.2 分层架构：ODS、DW 与清洗转换的职责边界

整体按三层走：业务系统层、ODS 层、DW 层。业务系统到 ODS 这一段做抽取加初步清洗，把脏数据和不完整数据挡在外面；ODS 到 DW 这一段做转换，包括编码统一、粒度聚合和业务指标计算。这个边界划清楚很关键，因为清洗规则和转换规则混在一起写，后期排查时根本分不清是源数据脏还是业务口径错。

业务系统(多DBMS) ──抽取──> ODS(贴源+清洗标记) ──转换──> DW(聚合+指标) │ │ │ └──────── 文件/手工表 ────┘ 日志与告警贯穿全链路

ODS 层建议保留源系统的原始字段，额外加etl_load_time、etl_batch_id、clean_status三个审计字段。这样任何一条进 DW 的数据都能反查到它来自哪一批、当时清洗判定是什么。DW 层则按分析粒度建表，不再保留明细冗余。

2.3 实现路线选型：工具、SQL 还是两者结合

原文给了三条路：纯 ETL 工具（如 SSIS、OWB、Informatica 这类）、纯 SQL、工具加 SQL 结合。我的经验是中小型项目直接选第三种。纯工具上手快、调度和日志现成，但复杂业务规则用图形化组件拼会非常别扭；纯 SQL 灵活、跑得快，但调度、依赖、失败重试全得自己写，等于重造半个轮子。结合的做法是：抽取和加载用工具或调度框架管，清洗转换里那些多表关联、窗口函数、指标计算用 SQL 写，工具只负责按顺序调用。

-- ODS 到 DW 的转换示例：统一供应商编码 + 按日聚合销售额 INSERT INTO dw_sales_daily (stat_date, supplier_code, region, total_amount, etl_batch_id) SELECT t.stat_date, m.unified_code AS supplier_code, -- 映射后的统一编码 t.region, SUM(t.amount) AS total_amount, -- 粒度从明细聚合到日 '${batch_id}' AS etl_batch_id FROM ods_order_detail t JOIN dim_supplier_map m -- 编码映射维表，解决跨系统编码不一致 ON t.supplier_code = m.source_code AND m.source_system = t.source_system WHERE t.clean_status = 'PASS' -- 只放行清洗通过的数据 AND t.stat_date = '${biz_date}' GROUP BY t.stat_date, m.unified_code, t.region;

这段 SQL 把三件事一次做完：通过dim_supplier_map做不一致数据转换，通过GROUP BY做粒度转换，通过SUM做业务指标计算。${batch_id}和${biz_date}是调度框架传入的参数，保证每次跑批可追溯、可重跑同一天。clean_status = 'PASS'这个条件不能省，它是清洗层和转换层之间的契约，少了它脏数据会直接漏进 DW。

3. 数据抽取模块：同构直连、异构 ODBC 与增量时间戳

3.1 按数据源类型分通道抽取

抽取不是写一个脚本通吃，而是按源类型分通道。与 DW 同构的库，直接在 DW 服务器和业务库之间建数据库链接，写 SELECT 就能取；异构库（比如一边 SQL Server 一边 Oracle）优先走 ODBC 建链接，建不了链接就退化成导出 txt/xls 再导入 ODS，或者走程序接口。文件类数据源（.txt、.xls）有两条路：培训业务人员用数据库工具先导进一个指定库再抽，或者用工具的平面数据源组件直接进 ODS。我一般倾向前者，因为把「文件怎么进来」的责任前移到业务侧，ETL 这边只面对库表，链路更干净。

# 增量抽取核心逻辑：基于业务时间戳取增量 import pandas as pd def extract_incremental(src_conn, ods_conn, table, ts_col): # 1. 取 ODS 中该表已有的最大业务时间，作为本次增量起点 max_ts = pd.read_sql( f"SELECT MAX({ts_col}) AS m FROM ods_{table}", ods_conn )["m"][0] # 2. 首次抽取无历史水位时，回退到全量或约定起始日 if max_ts is None: max_ts = "1970-01-01 00:00:00" # 3. 只取大于水位的数据，避免全表扫描 sql = f"SELECT * FROM {table} WHERE {ts_col} > %(ts)s" df = pd.read_sql(sql, src_conn, params={"ts": max_ts}) # 4. 打上批次与加载时间，写入 ODS df["etl_load_time"] = pd.Timestamp.now() df.to_sql(f"ods_{table}", ods_conn, if_exists="append", index=False) return len(df)

这里的关键是「水位」概念：每次抽取前先查 ODS 里已有的最大业务时间，只取大于它的记录。ts_col必须是业务系统真实记录业务发生时间的字段，不能用自增 ID 代替，因为自增 ID 在有回滚、补录的场景下会乱序。首次运行没有水位时回退到全量或约定起始日，这个分支一定要写，否则第一次跑批直接报空指针。

3.2 增量更新的现实约束

增量抽取的前提是业务系统有时间戳。现实里经常遇到「部分表有时间戳、部分表没有」的情况。对没有时间戳的表，通常的做法是退化成全量覆盖加比对，或者让业务方补一个更新时间字段。这一步要在调研阶段就确认，别等开发到一半才发现某张核心表没有时间字段，那时候改源系统已经来不及了。数据量大的系统必须走增量，全量每天跑一遍迟早把窗口跑爆。

4. 清洗与转换模块：三类脏数据的处理规则和编码统一

4.1 不完整、错误、重复三类数据的处置

清洗的任务是过滤不符合要求的数据，但过滤不是删掉了事，而是分类处置。不完整的数据（供应商名称缺失、客户区域为空、主表与明细表匹配不上）要按缺失内容分别导出成不同 Excel 交给业务主管部门，限期补全后再入库。错误的数据要再分两类：全角字符、字段前后有不可见字符这类，用 SQL 找出来要求业务系统修正后重抽；日期格式不对、日期越界这类会直接导致 ETL 跑挂的，必须去源库用 SQL 挑出来限期修正。重复数据在维表里最常见，把所有字段导出来让业务方确认去重规则。

-- 清洗检查：找出错误数据并打标记，不直接删除 UPDATE ods_customer SET clean_status = 'ERROR_DATE' WHERE clean_status = 'PENDING' AND (order_date IS NULL OR order_date < '2000-01-01' OR order_date > CURRENT_DATE); -- 日期越界判定 -- 找出含全角数字或首尾空白的记录 SELECT customer_id, customer_name FROM ods_customer WHERE customer_name ~ '[０-９]' -- 全角数字 OR customer_name <> TRIM(customer_name);

清洗结果用clean_status标记而不是物理删除，这是血泪经验：直接删掉的数据没法回溯，业务方回头说「这条不该过滤」时你拿不出证据。标记之后，过滤掉的数据可以每天邮件发给业务单位促其修正，同时作为后期验证数据质量的依据。清洗是个反复的过程，不可能几天做完，规则每加一条都要用户确认，最怕的是把有用数据误过滤了。

4.2 转换的三类工作：编码、粒度、业务规则

转换做三件事。不一致数据转换是整合，同一个供应商在结算系统是 XX0001、在 CRM 是 YY0001，抽过来之后通过映射维表统一成一个编码，第 2 章的dim_supplier_map就是干这个的。粒度转换是把业务系统的明细数据按 DW 的分析粒度聚合，明细进 ODS、聚合进 DW。业务规则计算是把那些不能简单加减的指标在 ETL 里算好存进 DW，供分析直接用，比如客户分层、同比环比、复合费率。这三类里最容易出问题的是业务规则，因为口径经常变，建议把规则参数化配置，别硬编码在 SQL 里。

5. 加载、日志与告警：让每次跑批都有黑匣子

5.1 加载策略与幂等

清洗完直接写 DW，但要保证幂等：同一批次重跑不能产生重复数据。常用做法是按stat_date + etl_batch_id先删后插，或者用 MERGE。加载失败要能回滚到批次前状态，所以每批加载前记录目标表行数，失败时按批次 ID 清理。

5.2 三类日志的分工

日志分三类，缺一不可。执行过程日志是流水账，记录每步的起止时间、影响行数；错误日志在模块出错时记录出错时间、模块名、错误信息；总体日志只记开始时间、结束时间、成功与否。用 ETL 工具的话工具会自带一部分日志，但业务层的清洗统计（过滤了多少条、各类错误各多少）还得自己写。

# 统一日志写入：过程日志 + 错误日志 + 总体日志 import logging, time def run_step(step_name, func, batch_id): start = time.time() try: rows = func() logging.info(f"[PROCESS] batch={batch_id} step={step_name} " f"rows={rows} cost={time.time()-start:.1f}s") return rows except Exception as e: logging.error(f"[ERROR] batch={batch_id} step={step_name} " f"time={time.strftime('%Y-%m-%d %H:%M:%S')} msg={e}") raise # 向上抛，由调度层决定是否告警

run_step把每个步骤包一层，成功写过程日志、失败写错误日志并继续上抛。上抛而不是吞掉，是为了让调度层统一判断整批是否失败，避免某一步静默失败、后面步骤拿着半截数据继续跑。

5.3 告警发送

出错不仅要写日志，还要主动通知。最常用的是给系统管理员发邮件并附上错误信息，方便排查。告警要分级：致命错误（整批失败）立即发，数据质量告警（过滤比例异常升高）每天汇总发一次，别什么都实时发，否则管理员很快就把告警邮件当垃圾邮件忽略了。

6. 避坑与验收：上线前必须走一遍的检查清单

6.1 四个最容易翻车的点

现象：重跑后 DW 数据翻倍。原因：加载没做幂等，重跑直接 append。解决：按批次先删后插或 MERGE，重跑前校验批次是否已存在。

现象：增量抽取漏数据。原因：用水位字段取了自增 ID 或非业务时间字段，补录数据时间早于水位被跳过。解决：水位必须用业务发生时间，并约定补录数据的处理方式。

现象：清洗把有用数据过滤了。原因：规则没经业务确认，边界条件写死。解决：每条规则先跑影子模式只标记不拦截，对比样本后再启用。

现象：跑批越来越慢。原因：ODS 表没建索引、全量抽取没改增量、日志表无限增长。解决：按时间字段建索引，日志表按月归档。

6.2 验收与答辩怎么讲

验收不看代码写得多漂亮，看四件事能不能当场演示。第一，拿一批故意构造的脏数据（缺字段、全角字符、重复记录）跑一遍，展示每类数据被正确标记和分流；第二，同一天重跑两次，展示 DW 行数不变，证明幂等；第三，人为让某一步失败，展示错误日志和告警邮件都到位；第四，随机抽一条 DW 记录，反查到 ODS 原始行和批次号。答辩时把这四步做成一张验收表，比讲架构图有说服力得多。

从那以后我每次交付 ETL 方案，都会在开发环境先把这张验收表走一遍再谈上线，因为跑批这东西平时看着没事，一出问题就是整条链路的数据对不上账，后悔药没处买。希望这套结构能帮到你，把课设或交付方案一次立住。

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