★ FEATURED ARTICLE 简介这是一份围绕数据工程全流程的Jupyter Notebook项目资料面向数据工程初学者、数据分析师和准备进入大数据开发领域的读者系统覆盖数据采集、清洗、转换、存储、集成和可视化等核心环节。压缩包内共有332个文件主要包含ipynb交互式笔记、py脚本、json配置、csv数据、sql查询、cfg配置及png图表整体大小约23.85MB目录结构清晰便于按主题查阅并动手复现。目前已有114人参与学习下载。内容提供dwh.cfg数据仓库配置、part-0000系列结果文件及日志数据并涉及Hadoop、Spark、Kafka等大数据处理工具与AWS/GCP云服务场景能够帮助读者理解真实ETL流程、数据建模思路与常见排错方法。通过项目练习还可以掌握Pandas数据清洗、SQLAlchemy数据库操作、数据可视化等关键技能快速获得可运行的代码与完整实验参考。1. Data Engineering 不是工具堆叠先搞清楚这条管线到底要解决什么问题很多人一提数据工程第一反应就是先上一套 Spark、Kafka、Hadoop好像组件越多越有工程感。但实际做下来你会发现绝大多数业务场景根本用不到那套重型基础设施真正决定数据工程质量的反而是那些看似不起眼的细节字段类型在源头是不是一致的、调度任务失败了能不能安全重跑、时区在不同数据源之间是不是统一。数据工程解决的不是“能不能存下大数据”的问题而是“数据能不能被稳定、准时、正确地交付给下游使用”的问题。这篇文章按我自己的落地习惯把端到端数据管线的四个核心环节——采集、清洗、建模、调度——拆开讲清楚用一套纯 Python 加关系型数据库的方案就能跑通适合数据开发新手建立全局观也适合后端转数仓的同学快速找到动手路径。2. 选型先把原理立住为什么是纯 Python 加关系型数据库而不是重型组件栈2.1 数据工程的最小闭环采集、清洗、建模、调度各环节的取舍数据工程不管用哪套技术栈核心闭环永远是固定的四条链路先把数据从源头取回来然后做清洗让数据可被信任再按业务逻辑建模组织成可查询的结构最后用调度让这一切按频率自动发生。这四个环节里最容易被轻视的是链路设计和数据契约设计。我见过很多团队把精力全花在清洗函数写得多花哨上结果源头采集阶段连字段类型都不稳定换一天的数据直接就崩。在选型这件事上我的判断标准很简单如果单机处理的量级是每天几十万到几百万行关系型数据库加 Python 的常规生态完全够用。计算下推给数据库清洗留在 Python 里做灵活性处理调度用系统级工具整条链路没有黑匣子。等到数据量上升到写入成为瓶颈或者下游需要秒级响应再考虑引入队列和分布式引擎。一上来就堆重型组件最大的问题不是性能过剩而是排查问题时你根本不知道数据在哪一环丢了这对数据工程是致命的。2.2 ETLe 与 ELT 选择的实际依据什么时候该先落地再转换是先把数据清洗好再加载到数仓还是先把原始数据原封不动落地再做转换这个争论在团队里永远存在。ETL 的优势是下游拿到的表直接能用对查询引擎的压力小但代价是清洗逻辑一旦写死源头加字段或者改格式你的清洗脚本就要连带改一轮。ELT 的思路是先把原始数据完整落下来转换交给查询层的 SQL 去处理上游变更的容忍度高很多。我从实际项目里的体会是优先选 ELT 更稳妥。原因很现实原始数据是最可靠的后悔药你清洗错了、建模建错了至少还能从原始层重来。而纯 ETL 链路里中间层一旦被覆盖想找回原来的样子就只能依赖备份。但是 ELT 也有前提条件你的存储和查询引擎得能承载原始数据膨胀的体量。关系型数据库对大字段和超宽表的支持有限所以实践中我一般做的是混合策略——分层设计每一层做适量的裁剪转换。原始数据落地成基础层清洗后的数据进入核心层业务聚合放在应用层层层递进每一层的变更影响范围都可控。2.3 轻量存储的真正瓶颈关系型数据库在什么规模下会拖垮你很多读者的担心是关系型数据库能扛多大并发。从单机使用场景讲写入并发和查询并发是两回事。SQLite 适合完全单机的场景但多进程写同一份库文件时锁竞争会非常明显我实测到每秒几十笔写入时就会出现卡顿。切换到 PostgreSQL 或 MySQL 之后并发能力会好一个数量级但代价是部署和运维复杂度上来了。数据量上来之后真正拖垮关系的往往不是行数而是没有做分区和索引设计。一张千万行的表如果每次查询都全表扫描再好的数据库也会被拖垮。所以我的建议是起步阶段用 SQLite 把链路跑通模型和调度稳定后再把连接串换成 PostgreSQL代码层的改动通常只有一行连接配置。先跑通再上强度这个策略能帮你省掉大量前期搭建时间。3. 落地一条端到端数据管线从采集到调度的可复现代码3.1 先搭骨架项目目录与数据库初始化做数据工程最容易犯的错是把所有脚本堆在一个目录里跑一段时间后自己都分不清哪份脚本对应哪层数据。我一般建议按分层结构组织目录project/ ├── data/ │ ├── raw/ # 原始数据落地 │ ├── staging/ # 清洗后的中间数据 │ └── warehouse/ # 建模后的数据文件 ├── scripts/ │ ├── fetch.py # 采集 │ ├── clean.py # 清洗 │ ├── model.py # 建模 │ └── validate.py # 校验 ├── schemas/ # 建表语句和字段定义 └── logs/ # 运行日志然后初始化数据库连接我习惯用一个独立的 db.py 统一管理连接和路径不把连接串散落到各脚本里import os import sqlite3 from pathlib import Path BASE_DIR Path(__file__).resolve().parent.parent DATA_DIR BASE_DIR / data DB_PATH DATA_DIR / analytics.db def get_conn(): DATA_DIR.mkdir(exist_okTrue) conn sqlite3.connect(DB_PATH) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA foreign_keysON;) return conn这里两个 PRAGMA 值得说明journal_mode 设为 WAL 能显著减少读写锁冲突多进程场景下比默认的 delete 模式稳定很多外键约束打开是为了在建模型层时保证维度表和事实表的引用完整性。路径统一从 BASE_DIR 推导也是避免你写死绝对路径后换机器就跑不了。3.2 数据采集层从外部数据源拉取并落盘的完整逻辑采集层写起来不难但稳定性是玄学。最常见的翻车点不是请求写不出来而是源数据的格式说变就变。所以我采集层的第一原则不信任源头的格式强制做一层落地快照。下面是拉取某电商平台订单导出的简化版本通过 HTTP 接口拉取后按日期归档成 JSONimport json import requests from datetime import datetime, timedelta from pathlib import Path def fetch_orders(date_str: str, save_dir: Path) - Path: 拉取指定日期的订单数据并落盘为 JSON 快照 # 实际接口地址按你的数据源替换 api_url fhttps://example-api.internal/orders?date{date_str} resp requests.get(api_url, timeout30) resp.raise_for_status() save_dir save_dir / date_str save_dir.mkdir(parentsTrue, exist_okTrue) out_file save_dir / forders_{date_str}.json out_file.write_text(json.dumps(resp.json(), ensure_asciiFalse, indent2), encodingutf-8) return out_file if __name__ __main__: today datetime.now().strftime(%Y-%m-%d) fetch_orders(today, Path(data/raw))一个参数值得注意ensure_asciiFalse是专门给非英文内容用的不设置的话中文会被转成 \uXXXX后续清洗时你看到的数据全是转义符排查乱码十分痛苦。另一个是超时参数 timeout 必须带不然源站假死时你的采集脚本会一直挂在那里占用连接。我一般还会在采集脚本里加一个重试装饰器状态码 500 时等待几秒重拉一次这个看你的源站稳定性决定要不要加。如果源数据不是 HTTP 接口而是本地文件采集逻辑就变成文件监听与增量比对。核心思路是维护一张记录已处理文件名的清单每次只处理清单里没有出现过的文件避免重复解析造成数据翻倍。3.3 数据清洗层处理缺失值、类型推断与字符串规范化的要点采集落盘后的数据不能直接进数仓。我用 pandas 做清洗最常遇到的三类问题是日期字段格式不统一、金额字段带着货币符号、同一家公司名在不同来源里写法不一致。下面是清洗层的标准动作import pandas as pd def clean_orders(raw_path: str, out_path: str) - pd.DataFrame: df pd.read_json(raw_path) # 1. 统一日期格式errorscoerce 让无法解析的日期变成 NaT df[order_date] pd.to_datetime(df[order_date], errorscoerce) # 2. 金额字段的字符串清洗去掉货币符号和千分位逗号 df[amount] ( df[amount].astype(str) .str.replace(¥, , regexFalse) .str.replace(,, , regexFalse) .astype(float) ) # 3. 公司名去除首尾空格并统一大小写避免同一主体被拆成多行 df[company_name] df[company_name].str.strip().str.title() # 4. 缺失值处理金额缺失直接剔除备注缺失填充占位符 df df.dropna(subset[amount]) df[remark] df[remark].fillna(N/A) df.to_parquet(out_path, indexFalse) return df清洗逻辑的要点不是处理得越多越好而是每一步都要有明确业务依据。比如金额缺失剔除是因为下游聚合时缺失金额会导致统计偏差备注字段填充占位符是因为这个字段只是辅助信息填充不影响分析结论。日期解析用errorscoerce而不是默认的报错模式是为了让单条脏数据不会拖垮整批任务清洗完再看有多少 NaT 就知道这批数据质量如何。如果你不想引入 parquet 格式存成 CSV 也可以但要记得指定 utf-8 编码和 indexFalse不然每次写入文件都会多出一列无意义的行号。3.4 数据建模层星型模型的事实表和维度表的建表语句清洗后的数据要变成可分析的结构。我对大多数业务场景默认推荐星型模型——一张事实表记录业务过程多张维度表记录业务实体的属性。事实表只存外键和度量值维度表存描述属性。以订单为例建表语句如下-- 订单事实表每行代表一条订单业务过程 CREATE TABLE fact_order ( order_id TEXT PRIMARY KEY, order_date_id INTEGER NOT NULL, user_id INTEGER NOT NULL, amount REAL NOT NULL, FOREIGN KEY (order_date_id) REFERENCES dim_date(date_id), FOREIGN KEY (user_id) REFERENCES dim_user(user_id) ); -- 日期维度表预填日期方便按周/月/季度聚合 CREATE TABLE dim_date ( date_id INTEGER PRIMARY KEY, date TEXT NOT NULL, year INTEGER NOT NULL, month INTEGER NOT NULL, day INTEGER NOT NULL, is_weekend INTEGER NOT NULL DEFAULT 0 ); -- 用户维度表 CREATE TABLE dim_user ( user_id INTEGER PRIMARY KEY, user_name TEXT NOT NULL, region TEXT, created_date TEXT );用 SQLite 建表时有个容易忽略的细节SQLite 不支持 ALTER TABLE ADD CONSTRAINT所以表结构设计阶段就要想清楚外键关系。另外 SQLite 默认不校验外键需要每次连接都执行PRAGMA foreign_keysON这也是我把它写进 get_conn 的原因。事实表的主键设置成业务订单号而不是自增 ID目的有两个一是天然防止重复导入二是将来做增量同步时可以直接用主键判断某条记录是否存在。如果你后续要切换到 PostgreSQL这套建表语句只需要把 TEXT 换成 VARCHAR 即可迁移成本很低。3.5 调度与重跑机制幂等设计让每次调度都安全可回溯调度是整个数据工程里最不像技术、但最容易出事故的环节。没有调度的脚本只是一堆手工命令有了调度之后你才敢说这是一条管线。我先用最简单的 cron 方式做定时触发写一个总入口脚本#!/bin/bash # pipeline.sh - 按顺序执行采集、清洗、建模 set -euo pipefail cd $(dirname $0)/.. PYTHONvenv/bin/python echo [$(date %F %T)] 开始采集 $PYTHON scripts/fetch.py --date $(date -d yesterday %F) echo [$(date %F %T)] 开始清洗 $PYTHON scripts/clean.py --input data/raw --output data/staging echo [$(date %F %T)] 开始建模 $PYTHON scripts/model.py --table fact_orderset -euo pipefail这行很短但非常重要-e 让脚本在某一步失败时立刻停止-u 防止变量未定义导致静默错误pipefail 保证管道中任何一条命令失败都会让整个管道返回失败状态。没有这行你的调度经常会出现“命令执行了但数据没更新”的假成功现象。真正的关键在一个设计原则每个脚本都要幂等。同一份数据重复执行两次结果不应该变化。实现幂等最可靠的办法是重跑前先清理本次涉及的数据范围。比如建模脚本在写事实表之前先把对应日期范围内的旧数据删掉再写入这样即使调度重复触发或者你手工重跑都不会产生重复记录。有了这个机制兜底你才敢放心把调度频率调高也不会因为一次失败的半成品数据污染整张表。4. 数据工程避坑记录时区、重复键、快照与并发覆盖4.1 时区不一致导致按天分区错位现象业务方反馈某天的订单量数据比后台系统统计少了一截而且少的量每天不固定。排查发现采集到的时间字段带时区偏移而清洗时用pd.to_datetime直接解析没有统一时区结果一部分订单的日期被归到了前一天。原因源头接口返回的时间是 UTC而本机环境是东八区解析出的时间直接用了本地时区展示。解决在采集层统一把时间字段转成 ISO 格式并带时区偏移在清洗层统一用tz_convert转到业务时区之后再截取日期。数据工程里最忌讳每个环节各自解读时间格式和时区必须在源头定死。4.2 重新清洗时覆盖了原始快照现象清洗脚本跑完后发现某个字段清洗规则写错了想回头重新处理结果发现原始文件已经被覆盖了只能重新拉取。原因清洗脚本把结果写回了原路径相当于把 raw 层的原始数据覆盖成清洗后的数据等于没有留后悔药。解决强制约定采集层目录 data/raw 是只读的清洗结果一律写往 data/staging用两层目录隔离原始数据和中间数据。这个约定要写进开发文档里我后来还加了一个启动检查清洗脚本如果检测到输出路径等于输入路径就直接报错退出。4.3 主键重复导致事实表数据翻倍现象某天调度日志显示订单事实表行数比前一天多了一倍但业务量并无明显增长。原因接口支持分页拉取采集脚本在翻页时因网络超时重试了最后一页而我没有对同一订单号做去重判断重复数据直接入库。解决事实表建表时把订单号设为主键写入用INSERT OR IGNORE的方式重复键直接跳过。不要依赖清洗阶段手工去重数据库层面的约束才是最后一道保险而且这个保险是免费的。4.4 多任务并发写同一个 SQLite 文件导致锁等待现象调度同时跑了三个任务日志里频繁出现 database is locked 的错误任务耗时从几分钟涨到半小时。原因SQLite 默认的日志模式在并发写入时锁粒度太大多个进程抢同一份文件的写锁。解决连接时执行PRAGMA journal_modeWAL把读写锁分开配合 busy_timeout 设置等待时间。WAL 模式解决了大部分并发写问题但如果你的数据量继续涨到每天千万行级别就该认真考虑切换 PostgreSQL 而不是继续跟锁斗争。4.5 第三方库依赖升级导致清洗结果变化现象一次例行维护中升级了 pandas 版本之后几天数据结果和之前对不上部分字段类型变了。原因pandas 对 CSV 自动类型推断的规则在不同版本间有调整之前被识别为整数的字段在升级后变成了浮点数进而影响了下游的聚合结果。解决pandas 读取 CSV 时强制指定 dtype 和 parse_dates 列表不要依赖自动推断。自动推断在开发阶段很方便但生产环境里它就是个黑匣子类型必须显式声明升级依赖后才能快速发现影响。5. 验证管线没白建先建数据血缘反查表再谈链路稳定5.1 数据血缘反查每条数据来自哪个原始文件哪一天管线运行一段时间后最大的困惑是面对一个异常结果不知道从哪查起。我的做法是在采集层给每批数据打上来源标记血缘反查会方便很多。具体实现很简单清洗后的每张表都加一个 source_file 字段记录这行数据来自哪个原始文件。SELECT source_file, COUNT(1) FROM fact_order WHERE order_date_id BETWEEN 20250101 AND 20250131 GROUP BY source_file;这样当业务方问“1 月份订单为什么少了”你只需要用上面这条 SQL 看哪一天没有数据然后直接去对应日期的原始快照排查不用漫无目的地翻日志。先解决“数据到底有没有来”再讨论“为什么不对”这是数据工程排查问题最顺的顺序。5.2 全链路自检的三条 SQL空值率、主键唯一性与行列数勾稽每次调度跑完我喜欢在建模层后面加一份自检报告用几条 SQL 快速判断这轮数据质量是否过关。不需要额外工具三条 SQL 基本够用分别对应高频故障模式可用于自检-- 检查关键字段空值率超过阈值就告警 SELECT COUNT(1) AS total_rows, COUNT(order_id) AS valid_ids, ROUND(1.0 * (COUNT(1) - COUNT(order_id)) / COUNT(1), 4) AS null_rate FROM fact_order; -- 检查主键唯一性大于 0 说明有重复数据 SELECT order_id, COUNT(1) AS c FROM fact_order GROUP BY order_id HAVING c 1 LIMIT 10; -- 行列数勾稽比较事实表维度外键与维度表主键是否完全匹配 SELECT COUNT(1) AS orphan_count FROM fact_order f LEFT JOIN dim_date d ON f.order_date_id d.date_id WHERE d.date_id IS NULL;第一条查空值率关键字段超过几个百分点就说明采集出了问题越早发现越省事。第二条查重复键正常跑批任务不应产出重复数据。第三条是查孤儿记录事实表的外键在维度表里找不到匹配说明建模阶段的维度表没有完全覆盖业务范围关联查询时这些数据会被静默丢弃。把这三条 SQL 加上阈值判断写成一个独立自检脚本放进调度链路的最后一步比遇到问题手忙脚乱翻数据强太多。5.3 耗时监控与调度频率的调整策略数据链路稳定运转之后下一步值得优化的是耗时和资源使用。用 Python 内建的 time 模块或者直接在看调度日志里的时间戳就能统计每段的耗时我习惯每次跑批都记录一个耗时指标采集多久、清洗多久、建模多久。判断要不要优化先看清洗耗时是否随数据量线性增长。如果一个月前的数据 10 分钟能跑完现在数据量翻倍但耗时涨了三倍大概率是清洗逻辑里有循环逐行处理或者 pandas 的类型推断在做重复劳动这时候值得检查一下有没有数据量级变化的误区。另一种情况是耗时完全跑平数据量涨了但耗时几乎没变这种不用过度优化说明瓶颈在固定开销上比如任务启动和连接建立不值得为它折腾。最后说一个我自己的习惯。数据工程里最能保护你的不是多么高深的技术而是对于“数据从哪来、经过什么处理、变成什么结果”保持透明。每次调度跑完我会看一眼当天的自检报告就像每天上班先看邮件一样几秒钟就能确认链路健康。数据工程很多问题不会马上炸开而是悄悄错上几天之后才被业务发现那时候的排查成本往往是当场的十倍不止。养成低频但稳定的自检习惯能帮你躲开大多数这类事故希望这些实践能帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
