简介本资源是一份面向数据工程师、BI开发人员及ETL初学者的系统性实践指南聚焦ETL核心环节——数据抽取、清洗与转换的设计原理与落地要点。文档深入剖析ODS层数据抽取策略含同构/异构数据库、文件类数据源的适配方案、增量更新机制详解三类脏数据不完整、错误、重复的识别逻辑、业务确认流程与修正路径并对比ETL工具如SSIS、OWB、纯SQL及混合方案的适用场景与优劣权衡同时涵盖日志分级管理与告警机制设计。资源为单个20KB的Word文档.docx内容结构清晰、术语准确、案例贴合企业级数仓建设实际已获1532人学习下载适合需要夯实ETL底层逻辑、规避常见设计陷阱、提升数据质量管控能力的中初级从业者快速掌握关键方法论。1. ETL 不是管道而是数据可信度的守门人很多人把 ETL 理解成“把 A 表的数据搬进 B 表”结果上线后 BI 报表天天报错、指标对不上、运营反复质疑“数怎么又变了”。真实项目里ETL 的核心矛盾从来不是“能不能跑通”而是“跑通之后谁敢信”——ODS 层里混着手工 Excel 补录的空值、CRM 系统里供应商编码用全角数字、订单时间戳被业务系统写成2023-02-30这些不是异常是常态。ETL 设计的本质是构建一套可验证、可回溯、可干预的数据质量控制链抽取阶段锁定源头可信边界清洗阶段定义“什么是脏”并留下修正证据转换阶段把业务规则固化为不可绕过的计算逻辑。它服务的对象不是数据库而是下游分析师、风控模型、管理层决策——当财务总监指着仪表盘问“为什么上月营收少了 37%”你得能立刻定位到是清洗环节漏掉了某类退款单的负向冲销而不是重跑一遍整个链路。本文聚焦实战中高频卡点如何设计增量抽取避免全量扫库、怎样用 SQL 精准识别全角字符和隐形换行符、为什么维表去重必须保留原始主键而非直接DISTINCT、日志结构如何支撑分钟级故障定位。所有方案均基于 PostgreSQL Python Airflow 组合验证适配中小型企业数据平台现状。2. 数据抽取从源头锁定可信数据边界2.1 抽取策略选型三类数据源对应三种技术路径ETL 抽取不是“越快越好”而是“在可控成本下获取最小必要数据集”。实际项目中需根据数据源类型选择底层机制避免盲目使用全量拉取同构数据库如 PostgreSQL → PostgreSQL优先采用dblink或物化视图避免中间文件落地。例如在目标库创建远程连接CREATE EXTENSION IF NOT EXISTS dblink; SELECT * FROM dblink(host192.168.1.10 port5432 dbnamecrm useretl passwordxxx, SELECT id, name, created_at FROM customers WHERE created_at 2024-01-01) AS t(id INT, name TEXT, created_at TIMESTAMP);提示dblink查询需显式声明返回字段类型否则易因类型不匹配导致任务失败WHERE条件必须包含时间范围禁止无条件SELECT *。异构数据库如 Oracle → PostgreSQL禁用 ODBC 直连生产库。采用ora2pg工具导出为 SQL 文件后在目标库执行# 在 Oracle 服务器执行导出需安装 ora2pg ora2pg -t COPY -o customers.sql -b /tmp/ -c oracle.conf # 传输至 PostgreSQL 服务器并导入 psql -U etl -d dw -f /tmp/customers.sql参数说明-t COPY启用高效批量插入-b指定输出目录oracle.conf中需配置ORACLE_DSN和PG_DSN连接串。文件类数据源.xlsx/.csv拒绝业务人员手动拖拽上传。部署inotifywait监控指定目录触发校验脚本# 监控脚本片段/opt/etl/watcher.sh inotifywait -m -e create,move_to /data/incoming | while read path action file; do if [[ $file ~ \.(xlsx|csv)$ ]]; then python3 /opt/etl/validate_file.py --path /data/incoming/$file --schema customers fi done校验逻辑必须包含文件编码检测UTF-8 with BOM、列数一致性检查、首行字段名白名单比对、空行率阈值5% 则告警。2.2 增量抽取用业务时间戳构建可审计的变更窗口全量抽取在千万级表上耗时超 2 小时且无法定位单条记录变更。正确做法是建立“双时间戳”机制last_modified业务系统更新时间etl_batch_idETL 执行批次号。字段名类型说明last_modifiedTIMESTAMP WITH TIME ZONE业务系统最后更新时间要求非 NULLetl_batch_idVARCHAR(32)格式为20240520_142305_abc123含日期、时间、随机码抽取 SQL 示例PostgreSQL-- 获取上次最大时间戳 WITH last_run AS ( SELECT COALESCE(MAX(last_modified), 1970-01-01::TIMESTAMP) AS max_time FROM etl_log WHERE table_name orders AND status success ) -- 抽取增量数据 INSERT INTO ods_orders (id, amount, status, last_modified, etl_batch_id) SELECT id, amount, status, last_modified, 20240520_142305_abc123 FROM dblink(hostprod-db port5432 dbnameerp, SELECT id, amount, status, last_modified FROM orders WHERE last_modified 2024-05-19 14:23:05) AS t(id INT, amount NUMERIC, status TEXT, last_modified TIMESTAMP) WHERE t.last_modified (SELECT max_time FROM last_run);注意last_modified必须有索引否则WHERE条件将触发全表扫描etl_batch_id需全局唯一避免跨任务覆盖。2.3 抽取可靠性保障断点续传与幂等性设计网络中断或目标库临时不可用时传统脚本会丢失已处理数据。解决方案是引入状态表etl_checkpointCREATE TABLE etl_checkpoint ( table_name TEXT PRIMARY KEY, last_max_timestamp TIMESTAMP WITH TIME ZONE, last_batch_id TEXT, updated_at TIMESTAMP DEFAULT NOW() );每次抽取前先查询该表获取last_max_timestamp抽取完成后用ON CONFLICT DO UPDATE写入新状态INSERT INTO etl_checkpoint (table_name, last_max_timestamp, last_batch_id) VALUES (orders, 2024-05-20 14:23:0508, 20240520_142305_abc123) ON CONFLICT (table_name) DO UPDATE SET last_max_timestamp EXCLUDED.last_max_timestamp, last_batch_id EXCLUDED.last_batch_id, updated_at NOW();此设计确保同一batch_id可重复执行而不产生重复数据且故障恢复时自动从断点继续。3. 数据清洗用 SQL 构建可解释的脏数据过滤器3.1 不完整数据识别缺失模式分类与修复路径绑定“缺失”不是单一问题需按业务影响分级处理。以客户表为例清洗规则需明确每类缺失的处置方式缺失字段影响等级处置方式SQL 示例mobile手机号P1拦截入库生成mobile_missing告警表WHERE mobile IS NULL OR LENGTH(TRIM(mobile)) 0region_code区域编码P2用上级区域补全记录region_filled日志COALESCE(region_code, (SELECT parent_code FROM regions WHERE code province_code))created_by创建人P3允许 NULL但需在 DW 层标注source_system legacyCASE WHEN created_by IS NULL THEN legacy ELSE crm END关键点所有清洗逻辑必须生成可追溯的中间表。例如ods_customers_cleaned表结构应包含CREATE TABLE ods_customers_cleaned ( id BIGINT, mobile TEXT, region_code TEXT, created_by TEXT, -- 清洗标记字段 mobile_status TEXT CHECK (mobile_status IN (valid, missing, invalid)), region_status TEXT CHECK (region_status IN (original, filled, unknown)), etl_batch_id TEXT, cleaned_at TIMESTAMP DEFAULT NOW() );3.2 错误数据精准捕获全角字符、隐形换行符、非法日期的 SQL 检测业务系统录入常引入不可见字符直接TRIM()无法清除。需用正则表达式逐项扫描全角数字检测如替代123SELECT id, name FROM ods_customers WHERE name ~ [\uFF10-\uFF19\uFF21-\uFF3A\uFF41-\uFF5A];说明\uFF10-\uFF19匹配全角数字 -\uFF21-\uFF3A匹配全角大写字母 -。隐形换行符与制表符\r\n\tSELECT id, address FROM ods_customers WHERE address ~ E[\\r\\n\\t] OR LENGTH(address) ! LENGTH(TRIM(address));非法日期格式如2024-02-30SELECT id, order_date FROM ods_orders WHERE order_date !~ ^\d{4}-\d{2}-\d{2}$ OR NOT (order_date::DATE BETWEEN 1900-01-01 AND NOW()::DATE);注意::DATE强制转换会抛出异常故先用正则校验格式再用范围判断有效性。3.3 重复数据治理维表去重必须保留业务主键维表如dim_product去重若直接SELECT DISTINCT *将丢失原始系统主键erp_product_id导致后续无法关联业务单据。正确做法是分组聚合并保留首次出现记录-- 创建清洗后维表 CREATE TABLE dim_product_clean AS SELECT MIN(id) AS id, -- 保留最小ID作为代理键 erp_product_id, -- 业务主键必须保留 product_name, category, MAX(updated_at) AS latest_update, -- 记录最新更新时间 COUNT(*) AS duplicate_count -- 标记重复次数 FROM ods_products GROUP BY erp_product_id, product_name, category HAVING COUNT(*) 1; -- 生成去重映射表供事实表关联使用 CREATE TABLE product_dedupe_map AS SELECT erp_product_id, MIN(id) AS clean_id FROM ods_products GROUP BY erp_product_id;此方案确保事实表通过erp_product_id关联时始终指向清洗后的唯一记录且duplicate_count字段可用于监控数据质量问题趋势。4. 数据转换将业务规则固化为不可绕过的计算逻辑4.1 不一致数据统一多源编码映射表驱动转换不同系统对同一实体使用不同编码如 CRM 中供应商编码CRM-SUP-001ERP 中为ERP_SUP_001硬编码映射易出错。应建立code_mapping主数据表CREATE TABLE code_mapping ( source_system TEXT NOT NULL, -- crm, erp, legacy source_code TEXT NOT NULL, -- 原始编码 target_domain TEXT NOT NULL, -- supplier, customer target_code TEXT NOT NULL, -- 统一编码 status TEXT CHECK (status IN (active, deprecated)), PRIMARY KEY (source_system, source_code, target_domain) ); -- 转换SQL示例将CRM和ERP供应商编码映射为统一ID SELECT COALESCE(c.target_code, e.target_code) AS supplier_id, c.source_system AS source_system, o.order_amount FROM ods_orders o LEFT JOIN code_mapping c ON o.supplier_code c.source_code AND c.source_system crm AND c.target_domain supplier LEFT JOIN code_mapping e ON o.supplier_code e.source_code AND e.source_system erp AND e.target_domain supplier;提示code_mapping表需每日校验完整性缺失映射时触发告警而非默认填充空值。4.2 数据粒度聚合从明细订单到月度销售汇总的 SQL 实现业务系统存储每笔订单明细而分析需求常需月度汇总。转换层需预计算并存储聚合结果避免报表层实时计算-- 创建月度销售汇总表 CREATE TABLE fact_sales_monthly AS SELECT DATE_TRUNC(month, order_date)::DATE AS sales_month, supplier_id, product_id, SUM(order_amount) AS total_amount, COUNT(*) AS order_count, AVG(order_amount) AS avg_order_value, -- 业务规则大客户定义为单月消费 10万元 CASE WHEN SUM(order_amount) 100000 THEN 1 ELSE 0 END AS is_key_customer FROM ods_orders_cleaned WHERE order_status IN (completed, shipped) GROUP BY DATE_TRUNC(month, order_date), supplier_id, product_id; -- 添加分区按月份 ALTER TABLE fact_sales_monthly ADD COLUMN sales_month_partition TEXT GENERATED ALWAYS AS (TO_CHAR(sales_month, YYYYMM)) STORED; CREATE INDEX idx_sales_monthly_partition ON fact_sales_monthly(sales_month_partition);此设计使 BI 工具查询“2024年5月各供应商销售额”时直接命中fact_sales_monthly表响应时间从秒级降至毫秒级。4.3 商务规则计算动态折扣率与阶梯返利的 SQL 表达复杂业务规则如“采购额满100万返5%满500万返8%”不能依赖应用层计算。需在 ETL 中固化为 SQL 函数-- 创建阶梯返利函数 CREATE OR REPLACE FUNCTION calculate_rebate(total_amount NUMERIC) RETURNS NUMERIC AS $$ BEGIN IF total_amount 5000000 THEN RETURN total_amount * 0.08; ELSIF total_amount 1000000 THEN RETURN total_amount * 0.05; ELSE RETURN 0; END IF; END; $$ LANGUAGE plpgsql; -- 在转换SQL中调用 SELECT customer_id, SUM(order_amount) AS annual_purchase, calculate_rebate(SUM(order_amount)) AS rebate_amount FROM ods_orders_cleaned WHERE order_date 2024-01-01 GROUP BY customer_id;注意函数需在目标库中创建且calculate_rebate必须声明为IMMUTABLE若参数不变则结果不变否则无法被查询优化器内联。5. ETL 日志与告警构建分钟级故障定位能力5.1 三层日志结构从流水账到根因分析ETL 日志不是记录“是否成功”而是提供“哪里失败、为何失败、如何修复”的线索。必须实现三类日志分离存储日志类型存储位置关键字段使用场景执行过程日志etl_step_log表step_name,start_time,end_time,rows_affected,sql_text定位慢 SQL查end_time - start_time 00:05:00的步骤错误日志etl_error_log表error_code,error_message,stack_trace,failed_row_data开发调试error_code映射到具体清洗规则如ERR_MOBILE_INVALID总体日志etl_batch_log表batch_id,status,duration,success_rate运维看板status failed AND success_rate 0.95触发告警建表语句示例CREATE TABLE etl_step_log ( id SERIAL PRIMARY KEY, batch_id TEXT NOT NULL, step_name TEXT NOT NULL, start_time TIMESTAMP WITH TIME ZONE, end_time TIMESTAMP WITH TIME ZONE, rows_affected BIGINT DEFAULT 0, sql_text TEXT, duration INTERVAL GENERATED ALWAYS AS (end_time - start_time) STORED ); CREATE INDEX idx_step_batch ON etl_step_log(batch_id); CREATE INDEX idx_step_duration ON etl_step_log(duration) WHERE duration 00:05:00;5.2 告警分级与邮件模板让运维一眼抓住重点告警邮件不能只写“ETL 失败”需结构化呈现关键信息。采用 Markdown 格式生成邮件正文【ETL 告警】订单清洗任务失败批次20240520_142305_abc123 ■ 故障定位 ▸ 失败步骤validate_mobile_format ▸ 错误代码ERR_MOBILE_INVALID ▸ 影响行数127 条占总数据 0.3% ■ 根因分析 检测到 127 条手机号含全角数字例 业务系统未做前端校验需协调 CRM 团队修复录入逻辑 ■ 临时措施 已将问题数据转入 ods_orders_invalid 表不影响主流程 下次运行将自动重试预计 2024-05-20 15:00 ■ 修复建议 ✅ 短期在清洗脚本中添加全角转半角函数见 PR#221 ✅ 长期推动 CRM 系统增加手机号格式校验提示邮件标题必须含batch_id便于在日志表中快速关联影响行数和占比是判断故障严重性的核心指标。5.3 日志驱动的自动化修复用 SQL 定位并隔离问题数据当清洗发现 127 条手机号异常时不应人工导出 Excel。应通过 SQL 自动生成修复指令-- 生成问题数据隔离语句 SELECT INSERT INTO ods_orders_invalid SELECT * FROM ods_orders WHERE id || id || ; FROM ods_orders WHERE mobile ~ [\uFF10-\uFF19] LIMIT 10; -- 生成修复建议SQL半角转换 SELECT UPDATE ods_orders SET mobile regexp_replace(mobile, E[\\uFF10-\\uFF19], (ascii(substring(mobile from position( in mobile))) - 65248)::TEXT, g) WHERE id || id || ; FROM ods_orders WHERE mobile ~ [\uFF10-\uFF19] LIMIT 1;此方案使运维人员复制粘贴即可执行隔离与修复将平均故障处理时间MTTR从小时级压缩至 5 分钟内。本文还有配套的精品资源点击获取
PostgreSQL+Airflow实战：构建高可信ETL数据质量链
简介本资源是一份面向数据工程师、BI开发人员及ETL初学者的系统性实践指南聚焦ETL核心环节——数据抽取、清洗与转换的设计原理与落地要点。文档深入剖析ODS层数据抽取策略含同构/异构数据库、文件类数据源的适配方案、增量更新机制详解三类脏数据不完整、错误、重复的识别逻辑、业务确认流程与修正路径并对比ETL工具如SSIS、OWB、纯SQL及混合方案的适用场景与优劣权衡同时涵盖日志分级管理与告警机制设计。资源为单个20KB的Word文档.docx内容结构清晰、术语准确、案例贴合企业级数仓建设实际已获1532人学习下载适合需要夯实ETL底层逻辑、规避常见设计陷阱、提升数据质量管控能力的中初级从业者快速掌握关键方法论。1. ETL 不是管道而是数据可信度的守门人很多人把 ETL 理解成“把 A 表的数据搬进 B 表”结果上线后 BI 报表天天报错、指标对不上、运营反复质疑“数怎么又变了”。真实项目里ETL 的核心矛盾从来不是“能不能跑通”而是“跑通之后谁敢信”——ODS 层里混着手工 Excel 补录的空值、CRM 系统里供应商编码用全角数字、订单时间戳被业务系统写成2023-02-30这些不是异常是常态。ETL 设计的本质是构建一套可验证、可回溯、可干预的数据质量控制链抽取阶段锁定源头可信边界清洗阶段定义“什么是脏”并留下修正证据转换阶段把业务规则固化为不可绕过的计算逻辑。它服务的对象不是数据库而是下游分析师、风控模型、管理层决策——当财务总监指着仪表盘问“为什么上月营收少了 37%”你得能立刻定位到是清洗环节漏掉了某类退款单的负向冲销而不是重跑一遍整个链路。本文聚焦实战中高频卡点如何设计增量抽取避免全量扫库、怎样用 SQL 精准识别全角字符和隐形换行符、为什么维表去重必须保留原始主键而非直接DISTINCT、日志结构如何支撑分钟级故障定位。所有方案均基于 PostgreSQL Python Airflow 组合验证适配中小型企业数据平台现状。2. 数据抽取从源头锁定可信数据边界2.1 抽取策略选型三类数据源对应三种技术路径ETL 抽取不是“越快越好”而是“在可控成本下获取最小必要数据集”。实际项目中需根据数据源类型选择底层机制避免盲目使用全量拉取同构数据库如 PostgreSQL → PostgreSQL优先采用dblink或物化视图避免中间文件落地。例如在目标库创建远程连接CREATE EXTENSION IF NOT EXISTS dblink; SELECT * FROM dblink(host192.168.1.10 port5432 dbnamecrm useretl passwordxxx, SELECT id, name, created_at FROM customers WHERE created_at 2024-01-01) AS t(id INT, name TEXT, created_at TIMESTAMP);提示dblink查询需显式声明返回字段类型否则易因类型不匹配导致任务失败WHERE条件必须包含时间范围禁止无条件SELECT *。异构数据库如 Oracle → PostgreSQL禁用 ODBC 直连生产库。采用ora2pg工具导出为 SQL 文件后在目标库执行# 在 Oracle 服务器执行导出需安装 ora2pg ora2pg -t COPY -o customers.sql -b /tmp/ -c oracle.conf # 传输至 PostgreSQL 服务器并导入 psql -U etl -d dw -f /tmp/customers.sql参数说明-t COPY启用高效批量插入-b指定输出目录oracle.conf中需配置ORACLE_DSN和PG_DSN连接串。文件类数据源.xlsx/.csv拒绝业务人员手动拖拽上传。部署inotifywait监控指定目录触发校验脚本# 监控脚本片段/opt/etl/watcher.sh inotifywait -m -e create,move_to /data/incoming | while read path action file; do if [[ $file ~ \.(xlsx|csv)$ ]]; then python3 /opt/etl/validate_file.py --path /data/incoming/$file --schema customers fi done校验逻辑必须包含文件编码检测UTF-8 with BOM、列数一致性检查、首行字段名白名单比对、空行率阈值5% 则告警。2.2 增量抽取用业务时间戳构建可审计的变更窗口全量抽取在千万级表上耗时超 2 小时且无法定位单条记录变更。正确做法是建立“双时间戳”机制last_modified业务系统更新时间etl_batch_idETL 执行批次号。字段名类型说明last_modifiedTIMESTAMP WITH TIME ZONE业务系统最后更新时间要求非 NULLetl_batch_idVARCHAR(32)格式为20240520_142305_abc123含日期、时间、随机码抽取 SQL 示例PostgreSQL-- 获取上次最大时间戳 WITH last_run AS ( SELECT COALESCE(MAX(last_modified), 1970-01-01::TIMESTAMP) AS max_time FROM etl_log WHERE table_name orders AND status success ) -- 抽取增量数据 INSERT INTO ods_orders (id, amount, status, last_modified, etl_batch_id) SELECT id, amount, status, last_modified, 20240520_142305_abc123 FROM dblink(hostprod-db port5432 dbnameerp, SELECT id, amount, status, last_modified FROM orders WHERE last_modified 2024-05-19 14:23:05) AS t(id INT, amount NUMERIC, status TEXT, last_modified TIMESTAMP) WHERE t.last_modified (SELECT max_time FROM last_run);注意last_modified必须有索引否则WHERE条件将触发全表扫描etl_batch_id需全局唯一避免跨任务覆盖。2.3 抽取可靠性保障断点续传与幂等性设计网络中断或目标库临时不可用时传统脚本会丢失已处理数据。解决方案是引入状态表etl_checkpointCREATE TABLE etl_checkpoint ( table_name TEXT PRIMARY KEY, last_max_timestamp TIMESTAMP WITH TIME ZONE, last_batch_id TEXT, updated_at TIMESTAMP DEFAULT NOW() );每次抽取前先查询该表获取last_max_timestamp抽取完成后用ON CONFLICT DO UPDATE写入新状态INSERT INTO etl_checkpoint (table_name, last_max_timestamp, last_batch_id) VALUES (orders, 2024-05-20 14:23:0508, 20240520_142305_abc123) ON CONFLICT (table_name) DO UPDATE SET last_max_timestamp EXCLUDED.last_max_timestamp, last_batch_id EXCLUDED.last_batch_id, updated_at NOW();此设计确保同一batch_id可重复执行而不产生重复数据且故障恢复时自动从断点继续。3. 数据清洗用 SQL 构建可解释的脏数据过滤器3.1 不完整数据识别缺失模式分类与修复路径绑定“缺失”不是单一问题需按业务影响分级处理。以客户表为例清洗规则需明确每类缺失的处置方式缺失字段影响等级处置方式SQL 示例mobile手机号P1拦截入库生成mobile_missing告警表WHERE mobile IS NULL OR LENGTH(TRIM(mobile)) 0region_code区域编码P2用上级区域补全记录region_filled日志COALESCE(region_code, (SELECT parent_code FROM regions WHERE code province_code))created_by创建人P3允许 NULL但需在 DW 层标注source_system legacyCASE WHEN created_by IS NULL THEN legacy ELSE crm END关键点所有清洗逻辑必须生成可追溯的中间表。例如ods_customers_cleaned表结构应包含CREATE TABLE ods_customers_cleaned ( id BIGINT, mobile TEXT, region_code TEXT, created_by TEXT, -- 清洗标记字段 mobile_status TEXT CHECK (mobile_status IN (valid, missing, invalid)), region_status TEXT CHECK (region_status IN (original, filled, unknown)), etl_batch_id TEXT, cleaned_at TIMESTAMP DEFAULT NOW() );3.2 错误数据精准捕获全角字符、隐形换行符、非法日期的 SQL 检测业务系统录入常引入不可见字符直接TRIM()无法清除。需用正则表达式逐项扫描全角数字检测如替代123SELECT id, name FROM ods_customers WHERE name ~ [\uFF10-\uFF19\uFF21-\uFF3A\uFF41-\uFF5A];说明\uFF10-\uFF19匹配全角数字 -\uFF21-\uFF3A匹配全角大写字母 -。隐形换行符与制表符\r\n\tSELECT id, address FROM ods_customers WHERE address ~ E[\\r\\n\\t] OR LENGTH(address) ! LENGTH(TRIM(address));非法日期格式如2024-02-30SELECT id, order_date FROM ods_orders WHERE order_date !~ ^\d{4}-\d{2}-\d{2}$ OR NOT (order_date::DATE BETWEEN 1900-01-01 AND NOW()::DATE);注意::DATE强制转换会抛出异常故先用正则校验格式再用范围判断有效性。3.3 重复数据治理维表去重必须保留业务主键维表如dim_product去重若直接SELECT DISTINCT *将丢失原始系统主键erp_product_id导致后续无法关联业务单据。正确做法是分组聚合并保留首次出现记录-- 创建清洗后维表 CREATE TABLE dim_product_clean AS SELECT MIN(id) AS id, -- 保留最小ID作为代理键 erp_product_id, -- 业务主键必须保留 product_name, category, MAX(updated_at) AS latest_update, -- 记录最新更新时间 COUNT(*) AS duplicate_count -- 标记重复次数 FROM ods_products GROUP BY erp_product_id, product_name, category HAVING COUNT(*) 1; -- 生成去重映射表供事实表关联使用 CREATE TABLE product_dedupe_map AS SELECT erp_product_id, MIN(id) AS clean_id FROM ods_products GROUP BY erp_product_id;此方案确保事实表通过erp_product_id关联时始终指向清洗后的唯一记录且duplicate_count字段可用于监控数据质量问题趋势。4. 数据转换将业务规则固化为不可绕过的计算逻辑4.1 不一致数据统一多源编码映射表驱动转换不同系统对同一实体使用不同编码如 CRM 中供应商编码CRM-SUP-001ERP 中为ERP_SUP_001硬编码映射易出错。应建立code_mapping主数据表CREATE TABLE code_mapping ( source_system TEXT NOT NULL, -- crm, erp, legacy source_code TEXT NOT NULL, -- 原始编码 target_domain TEXT NOT NULL, -- supplier, customer target_code TEXT NOT NULL, -- 统一编码 status TEXT CHECK (status IN (active, deprecated)), PRIMARY KEY (source_system, source_code, target_domain) ); -- 转换SQL示例将CRM和ERP供应商编码映射为统一ID SELECT COALESCE(c.target_code, e.target_code) AS supplier_id, c.source_system AS source_system, o.order_amount FROM ods_orders o LEFT JOIN code_mapping c ON o.supplier_code c.source_code AND c.source_system crm AND c.target_domain supplier LEFT JOIN code_mapping e ON o.supplier_code e.source_code AND e.source_system erp AND e.target_domain supplier;提示code_mapping表需每日校验完整性缺失映射时触发告警而非默认填充空值。4.2 数据粒度聚合从明细订单到月度销售汇总的 SQL 实现业务系统存储每笔订单明细而分析需求常需月度汇总。转换层需预计算并存储聚合结果避免报表层实时计算-- 创建月度销售汇总表 CREATE TABLE fact_sales_monthly AS SELECT DATE_TRUNC(month, order_date)::DATE AS sales_month, supplier_id, product_id, SUM(order_amount) AS total_amount, COUNT(*) AS order_count, AVG(order_amount) AS avg_order_value, -- 业务规则大客户定义为单月消费 10万元 CASE WHEN SUM(order_amount) 100000 THEN 1 ELSE 0 END AS is_key_customer FROM ods_orders_cleaned WHERE order_status IN (completed, shipped) GROUP BY DATE_TRUNC(month, order_date), supplier_id, product_id; -- 添加分区按月份 ALTER TABLE fact_sales_monthly ADD COLUMN sales_month_partition TEXT GENERATED ALWAYS AS (TO_CHAR(sales_month, YYYYMM)) STORED; CREATE INDEX idx_sales_monthly_partition ON fact_sales_monthly(sales_month_partition);此设计使 BI 工具查询“2024年5月各供应商销售额”时直接命中fact_sales_monthly表响应时间从秒级降至毫秒级。4.3 商务规则计算动态折扣率与阶梯返利的 SQL 表达复杂业务规则如“采购额满100万返5%满500万返8%”不能依赖应用层计算。需在 ETL 中固化为 SQL 函数-- 创建阶梯返利函数 CREATE OR REPLACE FUNCTION calculate_rebate(total_amount NUMERIC) RETURNS NUMERIC AS $$ BEGIN IF total_amount 5000000 THEN RETURN total_amount * 0.08; ELSIF total_amount 1000000 THEN RETURN total_amount * 0.05; ELSE RETURN 0; END IF; END; $$ LANGUAGE plpgsql; -- 在转换SQL中调用 SELECT customer_id, SUM(order_amount) AS annual_purchase, calculate_rebate(SUM(order_amount)) AS rebate_amount FROM ods_orders_cleaned WHERE order_date 2024-01-01 GROUP BY customer_id;注意函数需在目标库中创建且calculate_rebate必须声明为IMMUTABLE若参数不变则结果不变否则无法被查询优化器内联。5. ETL 日志与告警构建分钟级故障定位能力5.1 三层日志结构从流水账到根因分析ETL 日志不是记录“是否成功”而是提供“哪里失败、为何失败、如何修复”的线索。必须实现三类日志分离存储日志类型存储位置关键字段使用场景执行过程日志etl_step_log表step_name,start_time,end_time,rows_affected,sql_text定位慢 SQL查end_time - start_time 00:05:00的步骤错误日志etl_error_log表error_code,error_message,stack_trace,failed_row_data开发调试error_code映射到具体清洗规则如ERR_MOBILE_INVALID总体日志etl_batch_log表batch_id,status,duration,success_rate运维看板status failed AND success_rate 0.95触发告警建表语句示例CREATE TABLE etl_step_log ( id SERIAL PRIMARY KEY, batch_id TEXT NOT NULL, step_name TEXT NOT NULL, start_time TIMESTAMP WITH TIME ZONE, end_time TIMESTAMP WITH TIME ZONE, rows_affected BIGINT DEFAULT 0, sql_text TEXT, duration INTERVAL GENERATED ALWAYS AS (end_time - start_time) STORED ); CREATE INDEX idx_step_batch ON etl_step_log(batch_id); CREATE INDEX idx_step_duration ON etl_step_log(duration) WHERE duration 00:05:00;5.2 告警分级与邮件模板让运维一眼抓住重点告警邮件不能只写“ETL 失败”需结构化呈现关键信息。采用 Markdown 格式生成邮件正文【ETL 告警】订单清洗任务失败批次20240520_142305_abc123 ■ 故障定位 ▸ 失败步骤validate_mobile_format ▸ 错误代码ERR_MOBILE_INVALID ▸ 影响行数127 条占总数据 0.3% ■ 根因分析 检测到 127 条手机号含全角数字例 业务系统未做前端校验需协调 CRM 团队修复录入逻辑 ■ 临时措施 已将问题数据转入 ods_orders_invalid 表不影响主流程 下次运行将自动重试预计 2024-05-20 15:00 ■ 修复建议 ✅ 短期在清洗脚本中添加全角转半角函数见 PR#221 ✅ 长期推动 CRM 系统增加手机号格式校验提示邮件标题必须含batch_id便于在日志表中快速关联影响行数和占比是判断故障严重性的核心指标。5.3 日志驱动的自动化修复用 SQL 定位并隔离问题数据当清洗发现 127 条手机号异常时不应人工导出 Excel。应通过 SQL 自动生成修复指令-- 生成问题数据隔离语句 SELECT INSERT INTO ods_orders_invalid SELECT * FROM ods_orders WHERE id || id || ; FROM ods_orders WHERE mobile ~ [\uFF10-\uFF19] LIMIT 10; -- 生成修复建议SQL半角转换 SELECT UPDATE ods_orders SET mobile regexp_replace(mobile, E[\\uFF10-\\uFF19], (ascii(substring(mobile from position( in mobile))) - 65248)::TEXT, g) WHERE id || id || ; FROM ods_orders WHERE mobile ~ [\uFF10-\uFF19] LIMIT 1;此方案使运维人员复制粘贴即可执行隔离与修复将平均故障处理时间MTTR从小时级压缩至 5 分钟内。本文还有配套的精品资源点击获取