Nachossrd/etl-data-quality-pipeline

# 🧹 ETL 数据质量 Pipeline

## 🎯 问题描述 现实世界中的数据总是充满各种问题：同一列中存在五种不同格式的日期，价格表现为 `$850.000`、`350k` 或 `45425`（伪装成金额的 Excel 序列号），不一致的 ID（`TRX-001`、`002`、`id_003`），混杂的单位，以及出现在数字单元格中的人工备注。 本 pipeline 接收一个原始文件（`.csv`、`.xlsx` 或 SQL Server 的 `.bak` 备份），并生成一个**经过清洗、验证和审计的**数据集 —— 它会自动判断每一列的类型以及哪些行不可靠。 ## ✨ 功能特性 - **自动 schema 检测** —— 无需配置即可推断每列的类型、单位和值域。 - **语义标准化** —— 专用的解析器用于处理日期、货币/价格、ID、货币单位以及交易分类。 - **业务规则引擎** —— 根据声明式规则进行验证；违规操作不会中断运行，而是被隔离处理。 - **智能隔离机制** —— 可疑行会被分离到 `quarantine/` 目录并注明原因；如果超过阈值（默认为 `5%`），pipeline 将**以非 0 的 exit code 失败**，而不是带着可疑数据虚报成功。 - **质量报告** —— 生成 `quality_report.json`、运行清单以及人类可读的运行报告。 - **分析层** —— 生成维度模型、SQL schema、OLAP 元数据和 SSAS/XMLA 构件；并加载至 SQL Server（包含临时的 Docker 环境）。 - **可选的 Web UI**（基于 FastAPI）用于上传文件和下载结果，并支持 Bearer token 身份验证。 - **自动引导** —— 首次启动时自动安装依赖项。 ## 🏗️ 架构 ``` ┌─────────────────────────────────────────────┐ archivo crudo ──▶ │ extraction → schema detection → cleaning │ (.csv/.xlsx/.bak) │ → semantic enrichment → rule validation │ │ → quarantine → export → quality report │ └───────────────┬─────────────────────────────┘ │ ┌──────────────────────┼──────────────────────┐ ▼ ▼ ▼ dataset limpio cuarentena + modelo dimensional + manifest.json motivos + SQL / OLAP / SSAS ``` | 模块 | 职责 | |---------------|----------------------------------------------------------------| | `core/` | 数据抽取、schema 检测、清洗引擎、隔离区、数据导出、报告生成 | | `semantic/` | 语义解析器（日期、金额、单位、领域分类） | | `rules/` | 声明式业务规则（日期、货币、ID、价格、产品） | | `config/` | 通过 `.env` 进行集中式配置 | | `tests/` | 测试套件（基于 pytest 并使用合成测试数据） | ## 🚀 用法 ``` # 安装依赖 pip install -r requirements.txt # 复制并调整环境变量 cp .env.example .env # 在 demo 数据集上运行 pipeline python auto_pipeline.py --input infierno.csv --output ./output # (可选) 启动 Web UI python auto_pipeline.py --ui ``` **Flags:** | Flag | 描述 | |-----------------------------|------------------------------------------------------------| | `--input ` | 输入文件（`.csv`, `.xlsx`, `.bak`） | | `--output ` | 输出目录（本次运行结果将存放在 `

//`） | | `--quarantine-threshold `| 导致运行失败的隔离阈值 | | `--ui` | 启动 Web 界面 (FastAPI) | ## 📁 示例数据集 — `infierno.csv` 本仓库包含 `infierno.csv`，这是一个 **100% 合成**的数据集，旨在展示各种类型的数据脏乱情况：混合的日期格式、格式混乱的价格、不一致的 ID、伪装成金额的 Excel 序列号以及数据列中的人工备注。*该数据集不包含任何个人或企业的真实数据。* ## 🧪 测试 ``` python -m pytest tests/ -v ``` ## 🔒 配置与机密信息 所有敏感配置都存放在 `.env` 文件中（该文件永远不会被纳入版本控制）。请参阅 [`.env.example`](.env.example) 获取完整的变量列表：SQL 凭据、告警 webhook、Web UI token、chunk 大小以及各类阈值。 ## 🛠️ 技术栈 `Python` · `pandas` · `NumPy` · `SQLAlchemy` · `pyodbc` · `PyArrow` · `FastAPI` · `Uvicorn` · `openpyxl` · `rapidfuzz` ## ✍️ 作者 **Nacho** — [@Nachossrd](https://github.com/Nachossrd) 用 ☕ 和对脏数据的深切鄙视制作。 ## 📄 许可证 基于 MIT 许可证分发。请参阅 [`LICENSE`](LICENSE)。

标签：AV绕过, ETL, FastAPI, JavaCC, OLAP, 数据仓库, 数据清洗, 数据质量, 请求拦截, 逆向工具