mzanferrari/imdb-sentiment-terraform-pyspark-emr

# IMDB 情感分析 Pipeline - 基于 Amazon EMR 的 PySpark [](https://github.com/mzanferrari/imdb-sentiment-terraform-pyspark-emr/actions/workflows/ci.yml) [](https://www.terraform.io/) [](https://www.python.org/) [](https://aws.amazon.com/emr/) [](LICENSE) ## 目录 - [关于](#about) - [架构](#architecture) - [技术栈](#tech-stack) - [项目结构](#project-structure) - [前置条件](#prerequisites) - [快速开始](#quickstart) - [工作原理](#how-it-works) - [成本估算](#cost-estimate) - [测试](#testing) - [故障排除](#troubleshooting) - [已知局限性](#known-limitations) - [洞察与经验总结](#insights--lessons-learned) - [路线图](#roadmap) - [许可证](#license) ## 关于 **问题。** 使用分布式计算引擎在 IMDB 电影评论数据集（约 50k 条评论，约 63 MB CSV）上训练一个二元情感分类器（正面/负面），并将整个基础设施声明为代码。 **为什么选择这个范围。** 该数据集足够小，可以运行在配置了 DuckDB 或 pandas 的笔记本电脑上。对于 63 MB 的数据使用 Spark + EMR 本身就是**过度设计** —— 这个项目明确承认这一点。重点不在于需要 Spark；重点在于展示一名数据工程师需要端到端掌握的完整技能体系：IaC、云身份、分布式计算、ML 特征工程、可观测性以及 FinOps 思维。 **为什么这不是“另一个普通的 DSA 课程项目”。** 除了最初的范围，本仓库还添加了：架构决策记录 (ADR)、带有安全扫描的 GitHub Actions CI、自动化数据摄取、支持 FinOps 的集群规模调整（包含 EMR Serverless 选项）、符合 GDPR 的设计说明，以及将相同的 pipeline 架构应用于国际贸易数据 (UN Comtrade) 以构建领域迁移叙事的路线图。 ## 架构 ``` flowchart LR subgraph "Local / Dev" Dev[Developer] -->|terraform apply| TF[Terraform CLI] Dev -->|docker compose up| Dock[Dev Container] end subgraph "AWS - eu-west-1" TF -->|bootstrap| StateBkt[(S3 Terraform State
encrypted + versioned)] TF -->|data-platform| AppBkt[(S3 Project Bucket)] TF --> IAM[IAM Roles
EMR Service + EC2 Profile] TF --> EMR[EMR Cluster
release 7.13.0] AppBkt -->|raw CSV| EMR AppBkt -.->|scripts| EMR EMR -->|features parquet| AppBkt EMR -->|models| AppBkt EMR -->|logs| AppBkt SSM[SSM Parameter Store
config & paths] --> EMR end subgraph "External" IMDB[IMDB Reviews Dataset
50k labeled reviews] IMDB -->|automated download| Dev end classDef aws fill:#FF9900,stroke:#232F3E,color:#000 classDef storage fill:#3F8624,stroke:#232F3E,color:#fff class EMR,IAM,SSM aws class StateBkt,AppBkt storage ``` **两阶段 Terraform 部署。** Bootstrap 创建远程状态桶；主堆栈从中读取数据。这种模式避免了“状态桶的状态存放在哪里”的鸡生蛋还是蛋生鸡问题。 **通过 SSM Parameter Store 进行配置。** 桶名称和 S3 路径在运行时从 SSM 中读取 - Python 代码中没有硬编码的 ARN。只需更改 `project_id` 变量，相同的 pipeline 即可在任何 AWS 账户中运行。 ## 技术栈 | 层级 | 工具 | 原因 | |---|---|---| | **IaC** | Terraform 1.15+ | 行业标准，模块化，原生 S3 状态锁定 (1.10+)。考虑过 Pulumi/CDK - 见 ADR-001。 | | **云** | AWS (eu-west-1) | 选择 EU 区域以符合数据驻留叙事。见 ADR-002。 | | **计算** | Amazon EMR 7.13.0 (Spark 3.5) | 托管 Spark 集群。提供 Serverless 变体 - 见 ADR-003。 | | **存储** | Amazon S3 + Parquet | 带有生命周期策略的对象存储；列式格式用于分析读取。 | | **身份** | IAM Instance Profiles | 无长期访问密钥。工作负载通过 EC2 元数据担任角色。 | | **配置** | SSM Parameter Store | 集中化配置，代码中不包含机密。 | | **ML** | PySpark MLlib | 在三种特征表示上进行逻辑回归：HashingTF, TF-IDF, Word2Vec。 | | **容器** | Docker + docker-compose | 开发环境隔离；在不同的机器上保持相同的 Terraform/AWS CLI 版本。 | | **CI** | GitHub Actions | Lint (ruff)、类型检查、测试、IaC 验证 (terraform fmt, tflint, trivy)、机密扫描。 | | **质量** | Pre-commit hooks | 每次提交前运行 ruff, mypy, terraform fmt, gitleaks。 | | **可观测性** | 结构化 JSON 日志 -> CloudWatch | 分布式运行的 Trace ID。 | ## 项目结构 ``` . ├── infra/ # All infrastructure as code │ ├── bootstrap/ # Stage 1: creates remote state bucket │ │ ├── main.tf │ │ ├── variables.tf │ │ └── terraform.tfvars.example │ └── data-platform/ # Stage 2: project resources │ ├── main.tf │ ├── variables.tf │ ├── terraform.tfvars.example │ └── modules/ │ ├── s3/ # Storage + SSM parameters │ ├── iam/ # Service & instance roles │ ├── emr/ # Cluster + security groups │ └── finops/ # Budget + SNS cost alerts ├── src/ │ └── pipeline/ # Spark application code │ ├── main.py # Entry point │ ├── config.py # SSM Parameter Store reader │ ├── logging_setup.py # Structured JSON logger │ ├── processing.py # Feature engineering │ ├── ml.py # Model training & evaluation │ └── s3_io.py # S3 upload helpers ├── scripts/ │ ├── bootstrap_emr.sh # Conda + Python deps on EMR cluster │ └── ingest_data.py # Automated dataset download ├── tests/ │ ├── conftest.py # Shared pytest fixtures │ ├── test_config.py # Unit tests on config + SSM loading │ ├── test_ingest.py # Unit tests on data ingestion │ ├── test_ml.py # Unit tests on training helpers │ ├── test_processing.py # Unit tests on transform logic │ └── test_s3_io.py # Unit tests on S3 path + write helpers ├── docs/ │ ├── ARCHITECTURE.md # System design rationale │ ├── DEPLOYMENT.md # Step-by-step deployment guide │ └── adrs/ # Architecture Decision Records │ ├── 0001-iac-tool.md │ ├── 0002-aws-region.md │ ├── 0003-emr-deployment-mode.md │ ├── 0004-no-pii-no-pseudonymisation.md │ ├── 0005-cost-guardrails.md │ └── 0006-containerized-dev-environment.md ├── data/ # gitignored; populated by ingest_data.py ├── .github/ │ └── workflows/ │ ├── ci.yml # Lint + test + IaC validate on PR │ └── security.yml # Scheduled trivy + gitleaks ├── .gitignore ├── .pre-commit-config.yaml ├── pyproject.toml # Python deps & tool config ├── requirements.txt # Fallback for non-uv users ├── .devcontainer/ │ └── devcontainer.json # VS Code dev container config ├── .dockerignore ├── Dockerfile ├── docker-compose.yml ├── Makefile # Common commands └── README.md ``` ## 前置条件 | 要求 | 版本 | 说明 | |---|---|---| | AWS 账户 | - | **不**足以使用免费套餐；集群会产生费用。建议设置预算告警。 | | AWS CLI | v2.15+ | 需在 `~/.aws/credentials` 中配置凭证 | | Terraform | 1.15+ | 原生 S3 状态锁定所需 | | Docker | 24+ | 本地安装的替代方案 | | Python | 3.11 | 本地 pipeline 运行和数据摄取。不能使用 3.12 - PySpark 3.5 需要 `distutils`（在 3.12 中已移除）。 | | Java (JRE) | 17 | PySpark 在本地运行 Spark（测试 + 本地运行）所需。`openjdk-17-jre-headless` | | `uv` (推荐) | 0.5+ | 现代 Python 包管理器。`pip install uv` | ## 快速开始 ### 1. 克隆并配置 ``` git clone https://github.com/mzanferrari/imdb-sentiment-terraform-pyspark-emr.git cd imdb-sentiment-terraform-pyspark-emr cp infra/bootstrap/terraform.tfvars.example infra/bootstrap/terraform.tfvars cp infra/data-platform/terraform.tfvars.example infra/data-platform/terraform.tfvars # 使用你的 project_id（AWS 账户 ID）编辑 .tfvars 文件 # 这些文件已被 gitignore - 它们不会被提交。 ``` ### 2. 下载数据集 ``` # 如果不存在，则创建 ./data/dataset.csv（约 63 MB） python scripts/ingest_data.py ``` ### 3. 使用 Docker 部署（推荐） ``` docker compose up -d docker compose exec mz-p2 bash # 在 container 内： make deploy # runs bootstrap + data-platform Terraform stacks ``` ### 4. 或者直接从宿主机部署 ``` make deploy ``` ### 5. 监控运行 EMR 集群将出现在 AWS Console 中 (EMR -> Clusters)。步骤按顺序执行；日志流式传输到 `s3:///logs/`。集群将在最后一个步骤完成后自动终止。 ### 6. 检查结果 ``` aws s3 ls s3:///output/ --recursive aws s3 cp s3:///output/ ./output/ --recursive ``` ### 7. 销毁 ``` make destroy ``` ## 工作原理 Pipeline 在 2 个 EMR 步骤中运行： 1. **同步脚本 S3 -> EC2。** `aws s3 cp` 从 `s3:///pipeline/` 到 `/home/hadoop/pipeline/`（源文件 + 构建好的 `pipeline.zip`）。 2. **运行 Spark 应用程序。** `spark-submit main.py ` 触发完整的 ML pipeline： - 从 S3 读取原始 CSV。 - 验证 schema（`review`, `sentiment` 列）。 - 空值报告；删除包含空值的行（带有审计日志条目）。 - 类别平衡审计（正面 vs 负面计数）。 - 文本清理：去除 HTML 标签、非字母字符、多余空格；转为小写。 - Tokenization (RegexTokenizer) + 去除停用词。 - 三种并行特征表示：HashingTF（250 个特征）、TF-IDF、Word2Vec + MinMaxScaler。 - 训练/测试集划分（70/30，固定种子）。 - 逻辑回归结合 2 折 CrossValidator，遍历 `maxIter in {10, 15, 20}`。 - 记录准确率；模型构件持久化到 S3。 集群在最后一步完成后自动终止（`keep_job_flow_alive_when_no_steps = false`），并设置了 10 分钟的空闲超时作为第二道防线（`auto_termination_policy`）。 ## 成本估算 ### 默认配置 (EMR-on-EC2, 合理配置规格) | 资源 | 规格 | $/小时 | 运行时长 | 成本/次 | |---|---|---|---|---| | Master | m5.xlarge (4 vCPU, 16 GB) 按需 | ~$0.214 | ~30 分钟 | ~$0.11 | | Core (×2) | m5.large (2 vCPU, 8 GB) 竞价型 (~70% 折扣) | 各 ~$0.066 | ~25 分钟 | ~$0.06 | | S3 存储 | ~100 MB，价格 $0.023/GB-月 | - | 每月 | <$0.01/月 | | 数据传输 | EU 区域内 | $0 | - | $0 | | **每次运行总计** | | | | **~$0.17** | 每月运行 10 次用于作品集演示：**~$1.70/月**。 ### EMR Serverless 变体（推荐用于零星工作负载） | 资源 | 费率 | 每次运行 | |---|---|---| | vCPU | $0.052624 / vCPU-小时 | ~$0.10 | | 内存 | $0.0057785 / GB-小时 | ~$0.04 | | **每次运行总计** | | **~$0.14** | 此外：零空闲成本，零集群管理开销。见 [`docs/adrs/0003-emr-deployment-mode.md`](docs/adrs/0003-emr-deployment-mode.md)。 ### 预算和空闲集群告警 - 由 Terraform 配置 成本防护机制以代码形式提供，位于 `modules/finops`（账户月度预算，在预测达到 80% 和实际达到 100% 时发送 SNS 告警），外加 `modules/emr` 中的 CloudWatch `IsIdle` 告警，如果集群空闲 15 分钟就会触发（这是 `auto_termination_policy` 之后的第二层防线）。要接收告警，请在被 gitignore 忽略的 `terraform.tfvars` 中设置 `alert_email`；Terraform 会创建 SNS 订阅，AWS 将发送一次性的确认链接。 ## 测试 ``` # 所有检查 make ci # 单独 make lint # ruff check + ruff format --check make typecheck # mypy --strict src/ make test # pytest -v --cov=src tests/ make iac-fmt-check # terraform fmt -check (no rewrite) make iac-validate # terraform validate on both stacks make iac-security # trivy config scan ``` 运行本地 pipeline 进行端到端冒烟测试（在执行 `pip install -e .` 或 `uv sync` 之后）： ``` EXECUTION_ENV=local python -m pipeline.main ``` `EXECUTION_ENV=local` 会跳过 S3 写入；其他所有操作均针对 `data/dataset.csv` 中的本地 CSV 运行。 ## 故障排除 | 症状 | 可能原因 | 解决方法 | |---|---|---| | `terraform init` 失败并提示 `NoSuchBucket` | 尚未应用 Bootstrap 堆栈 | 先运行 `terraform -chdir=infra/bootstrap apply` | | EMR 集群提示 `BOOTSTRAP_FAILURE` 并立即终止 | 集群上的 Bootstrap 脚本出错 | 检查 `s3:///logs//node//setup-devel.gz` | | `NameError: path_output is not defined` | 旧代码，修复前版本 | 从 main 分支拉取最新代码 | | 提示找不到 `dataset.csv` | 忘记运行 `ingest_data.py` | `python scripts/ingest_data.py` | | Spark 作业卡在 `RegexTokenizer` | Driver 可能出现 OOM - 输入超出预期 | 检查 `master_instance_group.instance_type` 并查看 Spark UI | | `ssm:GetParameter` 出现 `AccessDeniedException` | 实例配置文件缺少 SSM 读取策略 | 重新检查 `modules/iam/main.tf` - 必须附加 `AmazonSSMManagedInstanceCore` | ## 已知局限性 - **单一模型家族。** 仅训练了逻辑回归。路线图包括用于比较的 Random Forest、Gradient Boosted Trees。 - **无增量处理。** Pipeline 在设计上是全量刷新的（数据集是静态的）。对于不断变化的数据集，请参阅关于 CDC 模式的路线图项目。 - **无数据质量框架。** Schema 检查是通过断言手动进行的。路线图包括集成 `dbt tests` 或 `great-expectations`。 - **无数据血缘追踪。** 路线图包括集成 OpenLineage。 - **仅限英文文本。** 清理 pipeline 假设使用 ASCII；国际评论需要进行适配。 - **GDPR。** 项目使用不含 PII 的公开数据集。有关假设的 PII 感知变体的设计说明，请参见 [`docs/adrs/0004-no-pii-no-pseudonymisation.md`](docs/adrs/0004-no-pii-no-pseudonymisation.md)。 ## 洞察与经验总结 **关于针对合适的规模进行工程设计。** 63 MB 的数据集并不需要 Spark。这里的练习在于**模式 (pattern)**，而不是吞吐量。ADR-003 中记录了真正的数据工程 (DE) 决策 - 何时使用 EMR 是合理的，而何时在单台 VM 上使用 DuckDB 才是诚实的答案。 **关于通过 Parameter Store 进行配置。** 在运行时从 SSM 中提取桶名称和路径（而不是通过 Spark 参数或 Terraform 中设置的环境变量注入），意味着 Python 代码是**基础设施无关**。同一个 `main.py` 在开发、预发布和生产环境中运行无需更改代码 - 只有 SSM 目录结构发生变化。 **关于 Instance Profile 与访问密钥。** 代码中包含长期的 AWS 访问密钥是公共 GitHub 仓库中凭证泄露的头号原因。工作负载使用 Instance Profiles + STS，人类使用 IAM Identity Center (SSO)。按照设计，此仓库中不存在任何访问密钥。 **关于可观测性。** 普通的 `print()` 输出到 stdout 足以进行本地开发。结构化 JSON 日志（使用 `correlation_id` 将单个 pipeline 运行的所有日志行关联起来）是在分布式系统中进行调试的关键。CloudWatch Logs Insights 原生支持查询 JSON。 **关于将 FinOps 作为一门学科。** 你忘记关掉的集群比你随意配置规格的集群成本更高。`auto_termination_policy` + 预算告警 + 成本分配标签不是“平台团队的责任” - 它们应该包含在每一次 IaC 提交中。 ## 路线图 有关详细的演进路径，请参见 [`docs/ROADMAP.md`](docs/ROADMAP.md)。重点内容： - [ ] **集成 dbt**，用于在精选层上进行 SQL 优先的转换。 - [ ] **迁移到 EMR Serverless** 作为默认部署模式。 - [ ] 使用 **Apache Iceberg** 作为表格式，以支持多引擎读取。 - [ ] 加入 **OpenLineage** 指标监控以实现自动化血缘追踪。 - [ ] 使用 **Apache Airflow**（或 Dagster）实现超越单次运行的调度。 - [ ] **领域替换**：基于 **UN Comtrade** 国际贸易数据复制该 pipeline 架构 - 这是作者拥有 15 年实操经验的领域（对外贸易、清关、国际物流）。 - [ ] 结合 Kinesis Data Streams + Spark Structured Streaming 的**流处理变体**。 ## 许可证 基于 MIT 许可证分发。详情请参见 [`LICENSE`](LICENSE)。 ## 作者 **mzanferrari** - 正在向数据工程转型的外贸专家，目标是欧盟远程岗位。 - GitHub: [github.com/mzanferrari](https://github.com/mzanferrari) - LinkedIn: [linkedin.com/in/mzanferrari](https://www.linkedin.com/in/mzanferrari)

标签：Amazon EMR, ECS, FinOps, IMDB情感分析, PySpark, Terraform, 数据工程, 请求拦截, 逆向工具