# 面向机器学习的数据工程 _作者：Joe Alongi_ 欢迎来到我的工作笔记本和配套代码库。这里记录了我即将出版的书籍《面向机器学习的数据工程》（_Data Engineering for Machine Learning_）的活跃笔记、系统设计草案和代码原型。 我的目标很简单：我想带你踏上一段旅程。我们将共同构建一个生产级、全栈的遥测和机器学习平台。我们将从一个全新的 Mac 安装环境开始，一步步构建出部署完毕、安全且可观测的系统。我们会将现代数据工程实践与机器学习工作流直接整合，最终，你将掌握如何构建、观测并保护一个由 ML 驱动的应用程序。 要深入了解该系统的架构和概念设计，请阅读 [白皮书](WHITEPAPER.md)。 ## 快速链接 - [白皮书](WHITEPAPER.md) - [致谢与技术](#acknowledgements--technologies) ## 第 1 章：全新安装与环境设置 在构建全栈系统时——在我们的案例中是前端的 Angular 和后端的 Python (Django)——你需要一个坚实的基础。让我们从一个干净的开发环境开始吧。 ### 前端设置 如果你和我一样使用 Mac，我强烈建议使用 Apple 生态系统的包管理工具。打开终端并安装 Homebrew： ``` /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)" ``` 对于 Apple Silicon 芯片的 Mac，请确保安装 Rosetta 2 以保证与旧版二进制文件的兼容性： ``` softwareupdate --install-rosetta ``` Homebrew 准备就绪后，让我们安装 Node.js 和 Angular CLI 来启动前端引导过程： ``` brew install node npm install -g @angular/cli ``` 现在，让我们搭建一个新的 Angular 项目并启动开发服务器： ``` ng new frontend cd frontend npm start ``` 访问 `http://localhost:4200` 以验证开发服务器是否正在运行。为了尽早落实我对这个代码库的样式和质量标准，我配置了 ESLint 和 Prettier： ``` # 运行 ESLint 以检查代码问题 npm run lint # 使用 Prettier 格式化 codebase npx prettier --write . ``` 为了将这个前端推向生产环境，我们将使用多阶段 Dockerfile 对其进行容器化，并通过 NGINX 提供服务： ``` docker build -t frontend-app . docker run -p 8080:8080 frontend-app ``` ### 后端设置 在后端方面，我们使用的是 Python 和 Django。在 macOS 上，我发现管理和处理 Python 版本及依赖项绝对最快、最干净的方法是使用 [Astral uv](https://github.com/astral-sh/uv)。通过 Homebrew 安装它： ``` brew install uv ``` 一旦安装了 `uv`，我就用它来初始化环境、安装 Django 并搭建后端骨架： ``` mkdir backend && cd backend # 使用 uv 创建 virtual environment uv venv # 激活 virtual environment source .venv/bin/activate # 使用 uv 的 lightning-fast pip 接口安装 Django uv pip install django # Scaffold Django 项目 django-admin startproject config . python manage.py runserver ``` 访问 `http://127.0.0.1:8000` 确保它成功运行。就像前端一样，我们也希望强制执行专业的格式化（Google Python 风格指南）。我使用 Ruff 来实现这一点： ``` # 运行 Ruff lint 检查并自动修复问题 uvx ruff check --fix . # 自动格式化 Python 文件 uvx ruff format . ``` ### 在本地运行全栈应用 要在本地运行整个系统，你有几种选择。 **选项 A：macOS 一键开发者启动脚本（推荐）** 我编写了一个交互式脚本，它会启动基于 Docker 的支撑服务（PostgreSQL 和 Redpanda），并自动打开一个带有前端、后端、worker 和 CMS 标签页的新终端窗口。 ``` ./start_dev.sh ``` **选项 B：通过 Docker Compose 运行整个技术栈** 如果你更喜欢在 Docker 中运行所有内容： ``` docker-compose up --build ``` **选项 C：独立运行各项服务** 有时你需要原生运行某些内容以便于调试： 1. 启动支撑服务：`docker-compose up -d postgres redpanda` 2. 配置你的 `.env` 文件（在 `frontend/` 和 `backend/` 中将 `.env.example` 复制并重命名为 `.env`）。 3. 迁移并启动 Django： cd backend source .venv/bin/activate uv pip install -r requirements.txt python manage.py migrate python manage.py createsuperuser python manage.py runserver 4. 在单独的标签页中启动 worker：`python manage.py telemetry_worker` 和 `python manage.py ml_worker`。 5. 启动前端： cd frontend npm install --legacy-peer-deps npx dotenvx run -- npm start ## 第 2 章：保持代码库整洁 随着代码库的增长，保持高质量标准是重中之重。我为前端配置了 Prettier 和 ESLint，以防止代码风格发生偏移： ``` npm install --save-dev prettier ng add @angular-eslint/schematics ``` 为了在 Angular 之外快速进行 TypeScript 原型设计，我喜欢使用 `tsx`： ``` npm install --save-dev tsx npx tsx --watch your-script.ts ``` ### 自动化代码质量（Pre-commit） 为了节省时间，我配置了 pre-commit 钩子，以便在每次提交前自动检查和格式化 Python 文件、前端文件以及 YAML 配置。我通过 `uvx` 执行它们： ``` uvx pre-commit run --all-files ``` ## 第 3 章：构建接口与集成数据 现代系统设计的基石是解耦客户端和服务器。让我们通过一个简单的 REST API 健康检查来集成它们。 首先，我在 Django 中定义视图： ``` # backend/config/views.py from django.http import JsonResponse def health(request): return JsonResponse({"status": "ok"}) ``` ``` # backend/config/urls.py # 将此添加到你的 urlpatterns 中： # path('api/health', views.health, name='health'), ``` 为了让 Angular 能够与 Django 通信，我配置了 CORS： ``` pip install django-cors-headers ``` ``` # backend/config/settings.py import os from dotenv import load_dotenv load_dotenv() cors_origins = os.getenv('CORS_ALLOWED_ORIGINS', '') CORS_ALLOWED_ORIGINS = [o.strip() for o in cors_origins.split(',')] if cors_origins else [] ``` 现在，我配置 Angular 使用 Signals 来简洁地管理响应式状态并调用该接口： ``` // frontend/src/app/app.config.ts import { provideHttpClient, withFetch } from "@angular/common/http"; export const appConfig = { providers: [provideHttpClient(withFetch())] }; ``` ``` // frontend/src/app/app.component.ts import { Component, inject, signal, OnInit } from "@angular/core"; import { HttpClient } from "@angular/common/http"; @Component({ selector: "app-root", standalone: true, template: ``, }) export class AppComponent implements OnInit { backendStatus = signal<"checking" | "ok" | "error">("checking"); private http = inject(HttpClient); ngOnInit() { this.http.get<{ status: string }>("/api/health").subscribe({ next: (res) => this.backendStatus.set(res.status === "ok" ? "ok" : "error"), error: () => this.backendStatus.set("error"), }); } } ``` 这种模式——暴露 JSON、处理 CORS 并进行响应式消费——是我们应用程序的核心心跳。 ## 第 4 章：设计数据库 一个健壮的数据库对于存储训练机器学习模型所需的历史遥测数据至关重要。我们使用的是 PostgreSQL，我推荐使用 DBeaver 来轻松可视化 schema： ``` brew install --cask dbeaver-community ``` 让我们把健康检查演变为一个持久化的数据点。首先，创建一个 Django 应用： ``` python manage.py startapp monitor ``` 然后，定义一个模型来表示健康检查记录： ``` # monitor/models.py import uuid from django.db import models class Endpoints(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) url = models.URLField() last_tested = models.DateTimeField(auto_now=True) status_code = models.IntegerField() response_time = models.DurationField() is_active = models.BooleanField(default=True) ``` 运行迁移： ``` python manage.py makemigrations monitor python manage.py migrate ``` 最后，更新视图，使得每次该 endpoint 被请求时都会记录一条日志： ``` # config/views.py import time from datetime import timedelta from django.http import JsonResponse from monitor.models import Endpoints def health(request): start_time = time.time() # ... perform healthcheck logic ... duration = timedelta(seconds=time.time() - start_time) Endpoints.objects.create( url=request.build_absolute_uri(), status_code=200, response_time=duration, is_active=True ) return JsonResponse({'status': 'ok'}) ``` ## 第 5 章：数据可视化 一旦有活跃的遥测数据流入 PostgreSQL，下一个合乎逻辑的步骤就是进行可视化。 我创建了一个 endpoint 来暴露数据： ``` # monitor/views.py from django.http import JsonResponse from .models import Endpoints def get_all_endpoints(request): endpoints = list(Endpoints.objects.values()) return JsonResponse(endpoints, safe=False) ``` 在前端，我使用 `ag-charts` 构建响应式、交互式的图表： ``` npm install ag-charts-angular ag-charts-community ``` 在仪表盘组件中，我获取遥测数据并将其绑定到图表配置中： ``` // frontend/src/app/pages/dashboard/dashboard.ts import { Component, OnInit, inject } from "@angular/core"; import { AgCharts } from "ag-charts-angular"; import { HttpClient } from "@angular/common/http"; @Component({ selector: "app-dashboard", standalone: true, imports: [AgCharts], template: ``, }) export class Dashboard implements OnInit { private http = inject(HttpClient); public chartOptions = { title: { text: "Application Stability" }, data: [], series: [{ type: "line", xKey: "time", yKey: "statusCode" }], }; ngOnInit() { this.http.get("/api/monitor/endpoints").subscribe((data) => { this.chartOptions = { ...this.chartOptions, data: data.map((ep) => ({ time: new Date(ep.last_tested).toLocaleTimeString(), statusCode: ep.status_code, })), }; }); } } ``` ## 第 6 章：智能化（建模与训练） 既然我们正在跟踪 endpoint 健康数据，我们就可以开始构建模型来预测系统性能下降。我们引入了 PyTorch 和 Polars 以实现快速的数据操作： ``` pip install torch polars skops scikit-learn python manage.py startapp ml ``` 我们没有在 Web 服务器上同步运行高强度的训练循环，而是勾勒了一个基础的多层感知机。我们通过一个 API endpoint 暴露这个模型训练过程，以便可以异步触发它： ``` # backend/ml/ml_api.py import torch import torch.nn as nn from django.http import JsonResponse from monitor.models import Endpoints class SLAModel(nn.Module): def __init__(self): super().__init__() self.fc1 = nn.Linear(3, 16) self.fc2 = nn.Linear(16, 1) def forward(self, x): return self.fc2(torch.relu(self.fc1(x))) def train_model(request): endpoints = Endpoints.objects.all() # ... prepare X and Y tensors from endpoint data ... model = SLAModel() optimizer = torch.optim.Adam(model.parameters(), lr=0.01) optimizer.zero_grad() # loss = criterion(model(X), Y) # loss.backward() # optimizer.step() return JsonResponse({'status': 'training_initiated'}) ``` ## 第 7 章：保护计算资源安全 训练机器学习模型在计算上代价高昂，因此我们必须保护这些 endpoint 的安全。我们将用户凭证交给客户端上的 **Firebase Authentication** 处理，并在后端验证 token。 在前端，我们在 Angular 服务中跟踪身份验证状态： ``` // frontend/src/app/services/auth.service.ts import { Injectable, signal } from "@angular/core"; import { initializeApp } from "firebase/app"; import { getAuth, onAuthStateChanged } from "firebase/auth"; @Injectable({ providedIn: "root" }) export class AuthService { public isAuthenticated = signal(false); public currentUserId = signal(null); public auth: any; constructor() { const app = initializeApp(environment.firebase); this.auth = getAuth(app); onAuthStateChanged(this.auth, async (user) => { if (user) { const token = await user.getIdToken(); this.http .get("/api/v1/auth/user", { headers: { Authorization: `Bearer ${token}` }, }) .subscribe((res: any) => { this.isAuthenticated.set(res.status === "success"); this.currentUserId.set(res.user_id); }); } else { this.isAuthenticated.set(false); this.currentUserId.set(null); } }); } } ``` 在后端，我们使用自定义的 Django middleware 拦截调用，并通过 Firebase Admin SDK 验证 token： ``` # backend/config/middleware.py from django.contrib.auth.models import AnonymousUser, User from django.utils.deprecation import MiddlewareMixin from firebase_admin import auth class FirebaseAuthenticationMiddleware(MiddlewareMixin): def process_request(self, request): request.user = AnonymousUser() auth_header = request.META.get("HTTP_AUTHORIZATION") if not auth_header or not auth_header.startswith("Bearer "): return None token = auth_header.split(" ")[1] try: decoded_token = auth.verify_id_token(token) user, created = User.objects.get_or_create(username=decoded_token.get("uid")) request.user = user except Exception: pass return None ``` 我们还通过 SMS 实现了多因素身份验证 (MFA)，并使用 **Firebase App Check** 和 reCAPTCHA Enterprise 保护我们的 endpoint，从而阻止恶意的机器人流量。 ## 第 8 章：增强可观测性 为了在不阻塞主 Web 服务器的情况下扩展遥测数据接入，我引入了 **Redpanda**——一种快速、轻量级的 Kafka 替代方案。 首先，Angular 前端捕获异常并将其发布到我们的接入 endpoint。在后端，我们使用 `django-ninja` 和 `aiokafka` 编写了一个异步视图，它会立即分发事件： ``` # backend/telemetry/views.py import json from aiokafka import AIOKafkaProducer from ninja import Router router = Router() producer = AIOKafkaProducer(bootstrap_servers="localhost:9092") @router.post("/telemetry/endpoints") async def post_telemetry(request, payload: dict): await producer.start() await producer.send("app-events", json.dumps(payload).encode("utf-8")) await producer.stop() return {"status": "accepted"} ``` 一个后台 worker 订阅该 topic，拉取消息并使用 Polars 处理它们。我还集成了 Sentry 进行全栈错误跟踪，并集成了 Semgrep 来实现漏洞检查自动化。 ### OpenTelemetry 与 ClickHouse 集成 为了标准化我们的追踪和指标，我们在 **ClickHouse** 旁边集成了 **OpenTelemetry (OTel)**。 OpenTelemetry Collector 通过 gRPC/HTTP 接收来自我们的应用程序服务和基础设施的原生 OTLP 遥测数据。它处理、批处理并直接将这些大容量数据导出到 ClickHouse——这是一个针对 OLAP 工作负载进行了优化的闪电般快速的列式数据库。这使我们能够扩展可观测性并高效查询分布式追踪，而不会给我们的主 PostgreSQL 事务数据库带来负担。 ## 第 9 章：应用实践（状态页面） 随着遥测数据在 Redpanda 中流转，让我们构建一个实用的应用案例：一个公开的状态仪表盘。 1. **遥测处理**：worker 从 Redpanda 拉取健康检查指标，并通过 Polars 对它们进行批量处理。 2. **SLA 计算**：计算实时的 SLA 合规性。 3. **事件运营**：操作员登录我们的无头 CMS (Sanity.io) 以声明事件。Angular 前端通过 Signals 监听，实时更新警报横幅。 4. **历史可视化**：聚合遥测数据以渲染一个 90 天的交互式健康图表。 ## 第 10 章：数据加密与密钥管理 我们实现了端到端加密。在传输过程中，所有内容都采用 TLS 1.3。在静态状态下，PostgreSQL 中的 API 密钥和凭证通过 AES-256 数据加密密钥 (DEK) 进行加密。我们不在本地存储密钥，而是使用 Google Cloud KMS 进行信封加密，并每 90 天自动轮换一次密钥。 ## 第 11 章：模型调优 为了确保我们的 PyTorch 神经网络保持准确，我们利用 scikit-learn 的网格搜索和交叉验证实现了一个超参数调优 pipeline。这能找到最佳的学习率和隐藏层，并通过 `skops` 保存模型，以便租户 SLA 预测能够动态适应。 ## 第 12 章：收集非结构化数据 遥测告诉我们发生了_什么_故障；非结构化的反馈则告诉我们它_如何_影响了用户。我构建了一个利用 LangChain、LangGraph 和 Google Gemini 的 AI 增强 pipeline。 系统将自然语言投诉路由到 AI agent。该 agent 使用 Gemini 解析投诉，将其与最近的遥测上下文进行比较，并确定根本原因。最后，它将结构化的 JSON 分析结果发布回 Redpanda。 ## 第 13 章：利用威胁情报增强数据 为了检测恶意流量，我们集成了来自 Google Analytics 4、Microsoft Clarity、AbuseIPDB 和 AlienVault OTX 的 API。 这些数据直接馈送到 PyTorch 的 `ThreatModel` 中，用于计算访问威胁评分。我们还在本地执行主动网络侦察（如 ICMP ping 延迟和活动端口探测），以丰富上下文并标记异常行为。 ## 第 14 章：使用 Sanity 扩展报告和公告 我们使用 Sanity.io 作为带有边缘缓存 API CDN 的无头 CMS。这将我们的状态通信与后端数据库解耦。即使我们的主要基础设施崩溃，边缘缓存的 API CDN 也能确保用户能瞬间加载公共状态更新和横幅。 ## 第 15 章：集成简报订阅 为了保留对客户数据的控制权，我们将简报订阅存储在 PostgreSQL 数据库中，并通过 Resend 发送电子邮件。 ``` # 通过 Resend 发送电子邮件 send_resend_email( to_email=payload.email, subject="Welcome to Our Innovations Newsletter!", html_content="

Thank you for subscribing!

" ) ``` ## 第 16 章：开发者工作流与版本管理 我们使用自定义的 Python CLI 工具 (`scripts/git_flow.py`) 自动化我们的分支创建、PR 提交以及语义化发布标签。 ``` # 创建 feature branch python scripts/git_flow.py feature "user auth changes" # 生成 PR python scripts/git_flow.py pr # Semantic versioning bump python scripts/git_flow.py release patch ``` ## 第 17 章：无障碍合规审计 无障碍功能至关重要。我们使用 `scripts/run_axe.js` 封装了 `@axe-core/cli`，并将扫描 HTML 模板作为我们的 git pre-commit 钩子的一部分，确保在合并任何代码之前满足 WCAG 2.1 AA 标准要求。 ``` node scripts/run_axe.js frontend/src/index.html ``` ## 第 18 章：客户端内容同步 为了让我们的 LLM 上下文文件和前端文档保持完美同步，我们运行了一个自动化 pipeline (`scripts/sync_content.py`)，它会从这个 README 中提取章节，并将它们直接输送到我们的 Angular 资产中。 ## 第 19 章：第三方遥测与 Cloudflare 集成 我们通过添加 Cloudflare Web Analytics 扩展了遥测能力。我们安全地存储 Cloudflare API token（静态加密），并将 Cloudflare 信标动态注入到租户页面中，以收集全面的流量洞察。 ## 第 20 章：团队工作流与漏洞管理 为了高效解决安全问题，我们构建了一个集成的看板工作流。我们还优化了 UI，创建了受 Boneyard 启发的骨架屏加载器，并采用了奢华的“保时捷喷气绿金属色”调色板，赋予应用真正的尊贵质感。 ## 第 21 章：在 Railway 上进行生产部署 在整本书中，我们构建了一个为生产发布做好准备的平台。我们在 [Railway](https://railway.app/) 上无缝部署了 Web 前端、Django API 和遥测 Worker。详细的扩展配置和 CI/CD 触发器记录在 `RAILWAY.md` 文件中。 ## 第 22 章：企业级安全（SOC 2 与 CMMC 2.0） 为了推动该平台向企业级成熟度迈进，我们实施了严格的控制措施： 1. 带有 WebAuthn 加密密钥的 Google SSO (Firebase Auth)。 2 通过 Google Cloud Logging 实现不可变的 SIEM 日志记录。 3. 强化的、无发行版（distroless）的 Docker 镜像。 4. 通过 Socket.dev、Checkov、Trivy 和 Renovate 进行持续的依赖扫描和代码检查。 5. 通过 **Infisical** 委派密钥编排。 ## 第 23 章：自动化与维护计划 我们的平台旨在实现自给自足，依靠自主后台 worker 和 GitHub Actions 的结合来最大限度地降低维护开销： ### 自主应用 Worker 我们的 Django 后端运行长生命周期的异步 worker，原生地自主管理系统健康、安全和 ML 任务： - **每小时：** `security_worker` 持续获取并更新威胁情报数据。 - **每日：** `ml_worker` 根据最新的遥测数据，自动重新训练所有的机器学习威胁模型和 SLA 预测。 - **每 30 天：** `security_worker` 通过清理过期的数据库遥测数据并自主轮换活动的数据加密密钥 (DEK) 来强制执行合规性。 ### GitHub Actions 工作流 我们利用 GitHub Actions 和外部机器人，严格用于代码级审计、静态分析和依赖项维护： - **每周：** Renovate Bot 创建自动化的 Pull Request，以更新过时的包和依赖项。 - **每月（30 天周期）：** 计划任务工作流运行 Semgrep 安全扫描并验证依赖项锁定文件（`npm audit` 和 `uv lock`）。 - **每季度（90 天周期）：** 我们使用 `ruff` 运行前端 bundle 性能审计和严格的后端静态分析检查。 ## 致谢与技术 我要感谢那些为我这本书的架构提供动力的、令人难以置信的开源工具、平台和 AI 助手： - **前端**: [Angular](https://angular.dev/), [Prettier](https://prettier.io/), [ESLint](https://eslint.org/) - **后端与 API**: [Django](https://www.djangoproject.com/) ([Django Ninja](https://django-ninja.dev/)), [Gunicorn](https://gunicorn.org/), [NGINX](https://nginx.org/), [cryptography](https://cryptography.io/en/latest/) - **数据与消息代理**: [PostgreSQL](https://www.postgresql.org/), [Redpanda](https://redpanda.com/), [Polars](https://pola.rs/) - **机器学习与 AI**: [PyTorch](https://pytorch.org/), [Scikit-learn](https://scikit-learn.org/), [Skops](https://skops.readthedocs.io/), [LangChain](https://www.langchain.com/), [LangGraph](https://langchain-ai.github.io/langgraph/), [Google Gemini](https://deepmind.google/technologies/gemini/), [Antigravity AI Agent (Google DeepMind)](https://deepmind.google/) - **可观测性、安全性与 CMS**: [Sentry](https://sentry.io/), [OpenTelemetry](https://opentelemetry.io/), [ClickHouse](https://clickhouse.com/), [Semgrep](https://semgrep.dev/), [Renovate](https://docs.renovatebot.com/), [FOSSA](https://fossa.com/), [Checkov](https://www.checkov.io/), [Trivy](https://trivy.dev/), [Socket.dev](https://socket.dev/), [Gitleaks](https://gitleaks.io/), [detect-secrets](https://github.com/Yelp/detect-secrets), [Sanity.io](https://www.sanity.io/), [AbuseIPDB](https://www.abuseipdb.com/), [ipify](https://www.ipify.org/), [IPinfo](https://ipinfo.io/), [Google Analytics](https://analytics.google.com/), [Microsoft Clarity](https://clarity.microsoft.com/), [Cloudflare Web Analytics](https://www.cloudflare.com/web-analytics/), [Resend](https://resend.com/) - **DevOps、基础设施与工具**: [Docker](https://www.docker.com/), [Railway](https://railway.app/), [Infisical](https://infisical.com/), [pre-commit](https://pre-commit.com/), [Ruff](https://docs.astral.sh/ruff/), [Django Migration Linter](https://github.com/3YOURMIND/django-migration-linter) [](https://railway.com/deploy/deml?referralCode=BpTk0g&utm_medium=integration&utm_source=template&utm_campaign=generic) [](https://app.fossa.com/projects/git%2Bgithub.com%2Fdataengineeringformachinelearning%2Fdataengineeringformachinelearning?ref=badge_large&issueType=license) [](https://semgrep.dev)     

标签：Angular, Apex, Grype, Python, 凭据扫描, 学习笔记, 数据工程, 无后门, 机器学习, 测试用例, 自动化攻击, 请求拦截, 负责任AI, 逆向工具