79 lines
4.6 KiB
Markdown
79 lines
4.6 KiB
Markdown
# 台球场 ETL 系统
|
||
|
||
用于台球门店业务的数据采集与入湖：从上游 API 拉取订单、支付、会员、库存等数据，先落地 ODS，再清洗写入事实/维度表，并提供运行追踪、增量游标、数据质量检查与测试脚手架。
|
||
|
||
## 核心特性
|
||
- **两阶段链路**：ODS 原始留痕 + DWD/事实表清洗，支持回放与重跑。
|
||
- **任务注册与调度**：`TaskRegistry` 统一管理任务代码，`ETLScheduler` 负责游标、运行记录和失败隔离。
|
||
- **统一底座**：配置（默认值 + `.env` + CLI 覆盖）、分页/重试的 API 客户端、批量 Upsert 的数据库封装、SCD2 维度处理、质量检查。
|
||
- **测试与回放**：ONLINE/OFFLINE 模式切换，`run_tests.py`/`test_presets.py` 支持参数化测试；`MANUAL_INGEST` 可将归档 JSON 重灌入 ODS。
|
||
- **可安装**：`setup.py` / `entry_point` 提供 `etl-billiards` 命令，或直接 `python -m cli.main` 运行。
|
||
|
||
## 仓库结构（摘录）
|
||
- `etl_billiards/config`：默认配置、环境变量解析、配置加载。
|
||
- `etl_billiards/api`：HTTP 客户端，内置重试/分页。
|
||
- `etl_billiards/database`：连接管理、批量 Upsert。
|
||
- `etl_billiards/tasks`：业务任务（ORDERS、PAYMENTS…）、ODS 任务、DWD 任务、人工回放；`base_task.py`/`base_dwd_task.py` 提供模板。
|
||
- `etl_billiards/loaders`：事实/维度/ODS Loader；`scd/` 为 SCD2。
|
||
- `etl_billiards/orchestration`：调度器、任务注册表、游标与运行追踪。
|
||
- `etl_billiards/scripts`：测试执行器、数据库连通性检测、预置测试指令。
|
||
- `etl_billiards/tests`：单元/集成测试与离线 JSON 归档。
|
||
|
||
## 支持的任务代码
|
||
- **事实/维度**：`ORDERS`、`PAYMENTS`、`REFUNDS`、`INVENTORY_CHANGE`、`COUPON_USAGE`、`MEMBERS`、`ASSISTANTS`、`PRODUCTS`、`TABLES`、`PACKAGES_DEF`、`TOPUPS`、`TABLE_DISCOUNT`、`ASSISTANT_ABOLISH`、`LEDGER`、`TICKET_DWD`、`PAYMENTS_DWD`、`MEMBERS_DWD`。
|
||
- **ODS 原始采集**：`ODS_ORDER_SETTLE`、`ODS_TABLE_USE`、`ODS_ASSISTANT_LEDGER`、`ODS_ASSISTANT_ABOLISH`、`ODS_GOODS_LEDGER`、`ODS_PAYMENT`、`ODS_REFUND`、`ODS_COUPON_VERIFY`、`ODS_MEMBER`、`ODS_MEMBER_CARD`、`ODS_PACKAGE`、`ODS_INVENTORY_STOCK`、`ODS_INVENTORY_CHANGE`。
|
||
- **辅助**：`MANUAL_INGEST`（将归档 JSON 回放到 ODS）。
|
||
|
||
## 快速开始
|
||
1. **环境要求**：Python 3.10+、PostgreSQL。推荐在 `etl_billiards/` 目录下执行命令。
|
||
2. **安装依赖**
|
||
```bash
|
||
cd etl_billiards
|
||
pip install -r requirements.txt
|
||
# 开发模式：pip install -e .
|
||
```
|
||
3. **配置 `.env`**
|
||
```bash
|
||
cp .env.example .env
|
||
# 核心项
|
||
PG_DSN=postgresql://user:pwd@host:5432/LLZQ
|
||
API_BASE=https://api.example.com
|
||
API_TOKEN=your_token
|
||
STORE_ID=2790685415443269
|
||
EXPORT_ROOT=/path/to/export
|
||
LOG_ROOT=/path/to/logs
|
||
```
|
||
配置的生效顺序为 “默认值” < “环境变量/.env” < “CLI 参数”。
|
||
4. **运行任务**
|
||
```bash
|
||
# 运行默认任务集
|
||
python -m cli.main
|
||
|
||
# 按需选择任务（逗号分隔）
|
||
python -m cli.main --tasks ODS_ORDER_SETTLE,ORDERS,PAYMENTS
|
||
|
||
# Dry-run 示例（不提交事务）
|
||
python -m cli.main --tasks ORDERS --dry-run
|
||
|
||
# Windows 批处理
|
||
..\\run_etl.bat --tasks PAYMENTS
|
||
```
|
||
5. **查看输出**：日志目录与导出目录分别由 `LOG_ROOT`、`EXPORT_ROOT` 控制；运行追踪与游标记录写入数据库 `etl_admin.*` 表。
|
||
|
||
## 数据与运行流转
|
||
- CLI 解析参数 → `AppConfig.load()` 组装配置 → `ETLScheduler` 创建 DB/API/游标/运行追踪器。
|
||
- 调度器按任务代码实例化任务，读取/推进游标，落盘运行记录。
|
||
- 任务模板：确定时间窗口 → 调用 API/ODS 数据 → 解析校验 → Loader 批量 Upsert/SCD2 → 质量检查 → 提交事务并回写游标。
|
||
|
||
## 测试与回放
|
||
- 单元/集成测试：`pytest` 或 `python scripts/run_tests.py --suite online`。
|
||
- 预置组合：`python scripts/run_tests.py --preset offline_realdb`（见 `scripts/test_presets.py`）。
|
||
- 离线模式：`TEST_MODE=OFFLINE TEST_JSON_ARCHIVE_DIR=... pytest tests/unit/test_etl_tasks_offline.py`。
|
||
- 数据库连通性：`python scripts/test_db_connection.py --dsn postgresql://... --query "SELECT 1"`。
|
||
|
||
## 其他提示
|
||
- `.env.example` 列出了所有常用配置；`config/defaults.py` 记录默认值与任务窗口配置。
|
||
- `loaders/ods/generic.py` 支持定义主键/列名即可落 ODS；`tasks/manual_ingest_task.py` 可将归档 JSON 快速灌入对应 ODS 表。
|
||
- 需要新增任务时，在 `tasks/` 中实现并在 `orchestration/task_registry.py` 注册即可复用调度能力。
|
||
|