udayjoshi-captech/chi311-ml-platform

# 🏙️ 芝加哥 311 服务请求智能平台 [](https://github.com/udayjoshi-captech/chi311-ml-platform/actions/workflows/ci.yml) [](https://azure.microsoft.com) [](https://databricks.com) [](https://mlflow.org) [](https://python.org) [](https://terraform.io) ## 📋 目录 1. [项目概述](#-project-overview) 2. [架构](#-architecture) 3. [技术栈](#-technology-stack) 4. [项目结构](#-project-structure) 5. [快速开始](#-quick-start) 6. [CI/CD Pipeline](#-cicd-pipeline) 7. [Databricks Asset Bundle](#-databricks-asset-bundle) 8. [监控与可观测性](#-monitoring--observability) 9. [数据工程实践](#-data-engineering-practices) 10. [成本管理](#-cost-management) 11. [关键发现](#-key-findings) 12. [ML 作品集框架对齐](#-ml-portfolio-framework-alignment) ## 🎯 项目概述 ### 问题陈述 芝加哥的 311 服务每天处理约 3,000 个非紧急服务请求。运营团队缺乏在需求激增压垮资源之前主动检测到这些激增的工具，导致响应时间延长和人员安排效率低下。 ### 解决方案 一个端到端的 ML 平台，具有以下功能： - **预测** 7 天的服务请求量，以优化人员配置 - **检测异常** 使用统计阈值（平均值 + 2σ） - **跟踪请求生命周期** 通过 SCD Type 2 进行状态停留时间分析 - **验证数据质量** 在每个 pipeline 阶段使用 Great Expectations - **监控 pipeline 健康** 包含每个任务的行数、持续时间和漂移检测 ### 架构概述 ``` ┌──────────────────────────────────────────────────────────────────┐ │ DATA INGESTION │ │ Chicago 311 API (Socrata) → Chi311APIClient → ADLS Gen2 Volume │ └──────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ DATA QUALITY (Great Expectations) │ │ Bronze expectations → Silver expectations │ │ Results persisted to gold.dq_checkpoint_results + ADLS docs │ └──────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ LAKEFLOW PIPELINE (Medallion + SCD Type 2) │ │ Bronze (Autoloader) → Silver (SCD2 history) → Gold (aggregates) │ │ CONSTRAINT expectations on every table │ └──────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ ML PIPELINE (MLflow) │ │ Feature Engineering → Prophet Training → Anomaly Detection │ │ Experiments tracked in MLflow, metrics logged per run │ └──────────────────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ SERVING & MONITORING │ │ Databricks SQL Dashboard ← Batch Predictions ← PredictionLogger │ │ PipelineMetrics → gold.pipeline_run_log (observability table) │ │ Azure Monitor alert → email on Databricks job failure │ └──────────────────────────────────────────────────────────────────┘ ``` ## 🛠️ 技术栈 | 层级 | 技术 | 用途 | |---|---|---| | **Cloud** | Microsoft Azure | 基于订阅的沙盒环境 | | **Compute** | Azure Databricks (Premium SKU) | 托管 Spark, Unity Catalog, MLflow | | **Storage** | ADLS Gen2 + Delta Lake | ACID 事务, time travel, medalion 架构 | | **ETL** | Lakeflow (SQL Declarative Pipelines) | 通过 `APPLY CHANGES INTO` 实现声明式 SCD2 | | **数据质量** | Great Expectations 0.18 | 带有持久化数据文档的声明式验证 | | **ML Tracking** | MLflow | 实验管理, model registry | | **Forecasting** | Prophet | 处理季节性、节假日、缺失数据 | | **Feature Store** | 自定义 feature engineering 库 | 时间、滞后、滚动特征 | | **Dashboard** | Databricks SQL Dashboard | 具备自动刷新和权限的原生监控 | | **IaC** | Terraform (azurerm ~3.80) | 可重现的 Azure 资源配置 | | **CI/CD** | GitHub Actions | Linting, 单元测试, 自动化 bundle 部署 | | **Orchestration** | Databricks Asset Bundle (DAB) | Jobs + Lakeflow pipeline 生命周期管理 | | **Alerting** | Azure Monitor scheduled query rules | 通过 Log Analytics 在 job 失败时发送电子邮件 | ## 📁 项目结构 ``` chi311-ml-platform/ │ ├── README.md ├── databricks.yml # Databricks Asset Bundle (jobs + pipeline) ├── requirements.txt # Production dependencies ├── requirements-dev.txt # Dev dependencies (ruff, black, mypy, pytest) ├── setup.py ├── pytest.ini ├── .gitignore │ ├── notebooks/ │ ├── 01_setup/ │ │ └── 00_setup_exploration.py # Catalog, schemas, volumes, EDA │ ├── 02_ingestion/ │ │ ├── 01_api_to_volume.py # Chicago 311 API → ADLS Volume │ │ └── 02_bronze_autoloader.py # Autoloader: Volume → Bronze Delta table │ ├── 03_data_quality/ │ │ └── 01_data_quality_checks.py # GE validation + checkpoint persistence │ └── 04_ml/ │ ├── 01_forecasting.py # Prophet training + MLflow tracking │ └── 02_anomaly_detection.py # Statistical anomaly detection │ ├── pipelines/ │ └── chi311_scd2_pipeline.sql # Lakeflow: Silver SCD2 + Gold aggregates │ # Includes CONSTRAINT expectations on every table │ ├── src/chi311/ │ ├── ingestion/ │ │ └── api_client.py # Paginated Socrata client with retry + logging │ ├── features/ │ │ └── feature_engineering.py # Temporal, lag, rolling features with logging │ ├── models/ │ │ └── prophet_forecaster.py # Prophet wrapper: prepare_data, train, predict │ └── monitoring/ │ ├── prediction_logger.py # Delta MERGE upsert + drift detection │ └── pipeline_metrics.py # Per-task row counts + duration logging │ ├── tests/ │ ├── unit/ │ │ └── test_feature_engineering.py # 7 tests — all passing in CI │ └── integration/ │ └── tests_pipeline_e2e.py │ ├── dashboards/ │ ├── queries.sql # SQL queries for Databricks dashboard │ ├── setup_dashboard.py # Notebook for programmatic setup │ └── docs/databricks-dashboard-setup.md # Manual setup guide │ ├── infrastructure/terraform/ │ ├── main.tf # All Azure resources + Monitor alert │ ├── variables.tf │ ├── outputs.tf │ └── terraform.tfvars # (gitignored) secrets + workspace IDs │ └── .github/workflows/ └── ci.yml # 4-stage CI/CD pipeline ``` ## 🚀 快速开始 ### 前置条件 - 带有 Databricks workspace 的 Azure 订阅 - Python 3.11+ - Terraform CLI ≥ 1.5 - Azure CLI (`az login`) ### 1 — 克隆并安装 ``` git clone https://github.com/udayjoshi-captech/chi311-ml-platform.git cd chi311-ml-platform python -m venv venv && source venv/bin/activate # Windows: venv\Scripts\Activate.ps1 pip install -r requirements-dev.txt pip install -e . ``` ### 2 — 部署 Azure 基础设施 ``` cd infrastructure/terraform az login terraform init terraform apply ``` 最小化的 `terraform.tfvars`: ``` azure_subscription_id = "" owner_email = "" alert_email = "" monthly_budget = 100 databricks_workspace_url = "https://adb-.azuredatabricks.net" databricks_workspace_resource_id = "/subscriptions/.../workspaces/" ``` ### 3 — 部署 Databricks Asset Bundle ``` cd ../.. databricks bundle deploy -t dev ``` 在 dev workspace 中创建 3 个 job 和 1 个 Lakeflow pipeline。 ### 4 — 运行数据接入和 ML job ``` databricks bundle run -t dev daily_ingestion databricks bundle run -t dev data_quality databricks bundle run -t dev ml_training ``` ### 5 — 在本地启动 dashboard ``` cd app pip install -r requirements.txt DATABRICKS_HOST=https://adb-.azuredatabricks.net \ DATABRICKS_TOKEN= \ DATABRICKS_CATALOG=workspace \ streamlit run dashboard.py ``` ## ⚙️ CI/CD Pipeline GitHub Actions 工作流（`.github/workflows/ci.yml`）在每次推送到 `main` 分支时运行： ``` Code Quality → Unit Tests → Deploy to Dev → Deploy to Prod (disabled) ``` | 阶段 | 功能描述 | |---|---| | **代码质量** | 在 `src/` 和 `tests/` 上执行 `ruff check` + `black --check` + `mypy` (仅警告) | | **单元测试** | 运行 `pytest tests/unit` 并带有 `--cov=src/chi311` 覆盖率报告 | | **部署至 Dev** | 使用环境密钥执行 `databricks bundle deploy -t dev` | | **部署至 Prod** | 已禁用 (`if: false`) — 通过向 `production` 环境添加密钥来启用 | ### 所需的 GitHub 环境密钥 (Settings → Environments → dev) | 密钥 | 值 | |---|---| | `DATABRICKS_HOST` | `https://adb-.azuredatabricks.net` | | `DATABRICKS_TOKEN` | Databricks PAT (workspace → Settings → Developer → Access tokens) | ## 📦 Databricks Asset Bundle `databricks.yml` 配置了所有 dev 和 prod 资源。为了提高成本效益，dev 集群使用**单节点 spot 定价**。 ### Dev job | Job | 时间表 | 集群 | |---|---|---| | `[DEV] Chi311 Daily Ingestion` | 每天早上 6 点 (已暂停) | `Standard_DS3_v2`, 单节点, spot | | `[DEV] Chi311 Data Quality` | 按需运行 | `Standard_DS3_v2`, 单节点, spot | | `[DEV] Chi311 ML Training` | 星期一早上 8 点 (已暂停) | `Standard_DS3_v2`, 单节点, spot | ### Lakeflow pipeline | 设置 | 值 | |---|---| | 名称 | `[DEV] Chi311 Lakeflow Pipeline` | | 来源 | `pipelines/chi311_scd2_pipeline.sql` | | 集群 | `Standard_DS3_v2`, 1 个 worker, spot 定价 | | 模式 | Development | ## 🔍 监控与可观测性 ### 1. Pipeline 运行指标 — `gold.pipeline_run_log` 用 `PipelineMetrics` 包装每个 notebook 任务，以记录行数、持续时间和状态： ``` from chi311.monitoring import PipelineMetrics metrics = PipelineMetrics(catalog="workspace") metrics.start(task_name="bronze_autoloader", run_id="") try: # ... transformation logic ... PipelineMetrics.assert_non_empty(df_out, "bronze output") # raises on empty metrics.finish(rows_in=rows_in, rows_out=df_out.count(), spark_session=spark) except Exception as e: metrics.fail(str(e), spark_session=spark) raise ``` 列：`run_id`, `task_name`, `status`, `rows_in`, `rows_out`, `rows_dropped`, `duration_seconds`, `error_message`, `logged_at`。 ### 2. 预测漂移检测 — `gold.gold_prediction_log` `PredictionLogger` 通过在 `(ds, model_version)` 上执行 Delta MERGE 来 upsert 预测结果 —— 重新运行时具有幂等性。`check_drift()` 计算针对实际值的 MAPE，并在超过阈值（默认 20%）时发出 `logger.warning`。 ### 3. 数据质量检查点 — `gold.dq_checkpoint_results` 每次 DQ notebook 运行后，Great Expectations 的结果（已评估 / 已通过 / 已失败 / 通过率 %）会被持久化到可查询的 Delta 表中。HTML 数据文档会被构建到 ADLS 的 `checkpoints` 容器中。 ### 4. Azure Monitor 告警 计划的查询规则每 15 分钟轮询一次 Log Analytics 中的 `DatabricksJobs runFailed` 事件，并在发生任何失败时向 `alert_email` 发送电子邮件。由 Terraform 自动配置。 ### 5. Databricks SQL Dashboard — 三个标签页 | 标签页 | 内容 | |---|---| | 📊 概览 | KPI 卡片（总请求量、日均请求量、MAPE）、每日请求量趋势、星期几的模式、最近的 pipeline 运行 | | 🔮 预测 | 带有置信区间的 7 天预测、预测值与实际值的散点图、随时间变化的模型 MAPE 趋势 | | 🔍 监控 | 数据质量指标、pipeline 健康状态、异常检测结果、漂移监控、任务持续时间趋势 | **设置：** 请参阅 `docs/databricks-dashboard-setup.md` 进行手动设置，或运行 `dashboards/setup_dashboard.py` 进行自动创建。 **相较于 Streamlit 的优势：** - ✅ 无需单独的部署基础设施 - ✅ 原生的 Unity Catalog 权限 - ✅ 内置自动刷新调度 - ✅ 成本更低（使用共享的 SQL Warehouse） - ✅ 包含电子邮件/Slack 订阅 ## 🏗️ 数据工程实践 | 实践 | 实现 | |---|---| | **结构化日志** | 所有模块中使用 `logging.getLogger(__name__)`；在每个阶段记录行数 | | **幂等写入** | 在自然键上执行 Delta MERGE —— 可安全重新运行而不会产生重复数据 | | **Schema 强制校验** | Lakeflow 在 silver/gold 层的 `CONSTRAINT` (`ON VIOLATION DROP ROW` 或 `WARN`) | | **空数据集守卫** | 如果任何阶段产生 0 行数据，`PipelineMetrics.assert_non_empty()` 将引发异常 | | **带退避的重试** | `Chi311APIClient` 使用指数退避进行重试；重试耗尽后引发异常 | | **无硬编码配置** | Catalog 名称、workspace URL、token 全部来自环境变量或密钥 | | **Spot 定价** | 所有 dev 集群使用 `SPOT_WITH_FALLBACK_AZURE` —— 降低 60–80% 的成本 | | **单节点集群** | 使用 `local[*]` Spark 的 `num_workers: 0` —— 消除了 dev 环境中 worker VM 的成本 | ## 💰 成本管理 月度预算告警：**$100**（`chi311-dev-budget`，Azure Cost Management）。 | 组件 | 预估每月成本 | |---|---| | Databricks workspace (空闲开销) | ~$30–50 | | 每次 job 运行的 compute（`DS3_v2` spot, ~10 分钟） | ~$0.05–0.10 每次 | | ADLS Gen2 存储 (~15 GB) | ~$0.30 | | Log Analytics (最低 30 天保留期) | ~$1–3 | | Azure Monitor 告警规则 | ~$0.10 | | **总计（轻度 dev 使用）** | **~$35–60/月** | 与最初的双节点按需配置相比，单节点 spot 集群将每次运行的 compute 成本降低了约 80%。 ## 🔑 数据探索的关键发现 | 发现 | 值 | 影响 | |---|---|---| | 每日服务请求量 | ~3,000（不包括咨询电话） | 预测基准 | | 咨询电话占比 | ~40% (`"311 INFORMATION ONLY CALL"`) | 必须在 ML 之前过滤掉 | | 状态值 | Open, Completed, Canceled | SCD2 跟踪目标 | | 异常阈值 | 4,851 (平均值 + 2σ) | 检测基准 | | 周末下降 | 35–40% | 季节性特征 | | 第 28 选区 (Ward 28) 占主导 | 39% 的请求 | 咨询电话行政选区 —— 已排除 | ## 🏗️ ML 作品集框架对齐 | 框架要素 | 实现 | 状态 | |---|---|---| | **问题定义** | 311 需求预测与异常检测 | ✅ | | **数据获取** | Chicago 311 API (Socrata) 支持分页增量加载 | ✅ | | **数据质量** | Bronze/Silver 层使用 Great Expectations + Silver/Gold 层使用 Lakeflow 约束 | ✅ | | **Feature Engineering** | 时间、滞后、滚动特征，并带有行数日志记录 | ✅ | | **模型开发** | Prophet + MLflow 实验跟踪 | ✅ | | **部署** | 通过 CI 部署 Databricks Asset Bundle（3 个 job + 1 个 Lakeflow pipeline） | ✅ | | **监控** | PipelineMetrics, PredictionLogger, DQ 检查点, Azure Monitor 告警 | ✅ | | **CI/CD** | GitHub Actions: lint → 单元测试 → 部署至 dev | ✅ | | **IaC** | 所有 Azure 资源由 Terraform 管理，计划具备幂等性 ✅ | ## 📄 许可证 本项目用于作品集/教育目的。数据来源于 https://data.cityofchicago.org/，基于开放数据许可。

标签：Azure Databricks, ECS, Kubernetes, MLflow, Terraform, 异常检测, 数据工程, 机器学习工程, 逆向工具, 需求预测