358 lines
18 KiB
Markdown
358 lines
18 KiB
Markdown
# 技术设计：团购详情接口整合 ETL 数据流
|
||
|
||
> 对应需求文档：#[[file:.kiro/specs/etl-coupon-detail/requirements.md]]
|
||
|
||
## 1. 架构概览
|
||
|
||
```
|
||
ODS_GROUP_PACKAGE 任务（BaseOdsTask + OdsTaskSpec）
|
||
│
|
||
│ 阶段 1：列表拉取（UnifiedPipeline #1）
|
||
│ QueryPackageCouponList → ods.group_buy_packages
|
||
│
|
||
▼
|
||
ods.group_buy_packages ──写入完成──▶ 阶段 2：详情拉取自动启动
|
||
│ │
|
||
│ ┌───────┴───────┐
|
||
│ │ UnifiedPipeline│
|
||
│ │ #2 │
|
||
│ │ (detail_mode) │
|
||
│ └───────┬───────┘
|
||
│ │
|
||
│ ┌───────▼───────┐
|
||
│ │ 串行请求线程 │
|
||
│ │ (主线程) │
|
||
│ │ RateLimiter │
|
||
│ └───────┬───────┘
|
||
│ │ 响应提交
|
||
│ ┌───────▼───────┐
|
||
│ │ 处理队列 │
|
||
│ │ N 个工作线程 │
|
||
│ └───────┬───────┘
|
||
│ │ 处理完成
|
||
│ ┌───────▼───────┐
|
||
│ │ 写入队列 │
|
||
│ │ 单线程写库 │
|
||
│ └───────┬───────┘
|
||
│ │
|
||
▼ ▼
|
||
DWD dim_groupbuy_package ods.group_buy_package_details
|
||
DWD dim_groupbuy_package_ex ◀── SCD2 合并 ──┘
|
||
```
|
||
|
||
> 不再新建独立的 `DetailFetcher` 类。详情拉取完全复用 `BaseOdsTask` 已有的 `detail_endpoint` 二级拉取模式，通过 `OdsTaskSpec` 声明式配置即可驱动。
|
||
|
||
## 2. 调研结论与设计决策
|
||
|
||
### 2.1 ODS 层表方案 → 选项 A：新建独立表
|
||
|
||
决策依据：
|
||
- 现有 `ods.group_buy_packages` 使用 `OdsTaskSpec` 驱动，payload 存储列表接口的原始 JSON，PK 为 `id + content_hash`
|
||
- 详情接口返回嵌套结构（子数组 `packageCouponAssistants`、`grouponSiteInfos` 等），与列表接口的扁平结构差异大
|
||
- 两个接口的写入时机不同（列表先写完，详情后写），混在同一表会导致 `SnapshotMode.FULL_TABLE` 的软删除逻辑冲突
|
||
- 独立表可以独立演进字段，不影响现有列表数据的稳定性
|
||
|
||
新表：`ods.group_buy_package_details`，PK = `coupon_id`（BIGINT），全量快照覆盖
|
||
|
||
### 2.2 DWD 层表方案 → 选项 A：在现有扩展表新增字段
|
||
|
||
决策依据：
|
||
- `dim_groupbuy_package_ex` 当前 21 个业务字段（不含 SCD2），密度适中
|
||
- 详情接口的增量价值字段仅 4 个 JSONB 列（`table_area_ids`、`table_area_names`、`assistant_services`、`groupon_site_infos`）
|
||
- 新建独立表会增加 SCD2 版本同步复杂度，且下游查询需要额外 JOIN
|
||
- 扩展表的 SCD2 合并已与主表同步，新增字段自动纳入变更检测
|
||
|
||
### 2.3 取消信号 → 复用现有 `CancellationToken`
|
||
|
||
`CancellationToken`（封装 `threading.Event`）已在 etl-unified-pipeline 中实现。详情阶段的 `UnifiedPipeline` 实例共享同一个 `cancel_token`，在请求循环和 `RateLimiter.wait()` 中检查取消状态。
|
||
|
||
## 3. 组件设计
|
||
|
||
### 3.1 OdsTaskSpec 详情配置（声明式，无需新建类）
|
||
|
||
> `RateLimiter`（`api/rate_limiter.py`）、`CancellationToken`（`utils/cancellation.py`）和 `UnifiedPipeline`（`pipeline/unified_pipeline.py`）已在 etl-unified-pipeline 中实现并上线。`BaseOdsTask.execute()` 内置了 `detail_endpoint` 二级详情拉取模式，本 spec 通过 `OdsTaskSpec` 声明式配置驱动，不新建独立类。
|
||
|
||
在 `tasks/ods/ods_tasks.py` 的 `ODS_GROUP_PACKAGE` 任务 spec 中添加以下配置：
|
||
|
||
```python
|
||
OdsTaskSpec(
|
||
code="ODS_GROUP_PACKAGE",
|
||
# ... 现有列表拉取配置 ...
|
||
|
||
# ── Detail_Mode 配置 ──
|
||
detail_endpoint="/PackageCoupon/QueryPackageCouponInfo",
|
||
detail_param_builder=lambda rec: {"couponId": rec["id"]},
|
||
detail_target_table="ods.group_buy_package_details",
|
||
detail_data_path=("data",),
|
||
detail_id_column="id",
|
||
)
|
||
```
|
||
|
||
配置字段说明：
|
||
|
||
| 字段 | 值 | 说明 |
|
||
|------|-----|------|
|
||
| `detail_endpoint` | `/PackageCoupon/QueryPackageCouponInfo` | 详情接口 endpoint |
|
||
| `detail_param_builder` | `lambda rec: {"couponId": rec["id"]}` | 将列表表的 `id` 映射为详情接口的 `couponId` 参数 |
|
||
| `detail_target_table` | `ods.group_buy_package_details` | 详情数据写入的目标表 |
|
||
| `detail_data_path` | `("data",)` | 详情响应的数据路径 |
|
||
| `detail_id_column` | `id` | 从 `ods.group_buy_packages` 提取 couponId 列表的列名 |
|
||
|
||
### 3.2 执行流程（BaseOdsTask.execute() 内置）
|
||
|
||
`BaseOdsTask.execute()` 在列表拉取全部完成后，自动检测 `spec.detail_endpoint` 是否配置，若已配置则启动详情拉取阶段：
|
||
|
||
```python
|
||
# BaseOdsTask.execute() 内置逻辑（已实现，无需修改）：
|
||
if spec.detail_endpoint:
|
||
# 1. 创建独立的 UnifiedPipeline 实例（共享 cancel_token）
|
||
detail_pipeline = UnifiedPipeline(
|
||
api_client=self.api,
|
||
db_connection=self.db,
|
||
logger=self.logger,
|
||
config=pipeline_config, # PipelineConfig.from_app_config()
|
||
cancel_token=cancel_token, # 与列表阶段共享
|
||
)
|
||
|
||
# 2. 从 ODS 目标表查询 ID 列表，生成详情请求序列
|
||
detail_requests = self._build_detail_requests(spec)
|
||
# → SELECT DISTINCT {detail_id_column} FROM {table_name}
|
||
# → 对每个 ID 调用 detail_param_builder 构造参数
|
||
# → yield PipelineRequest(is_detail=True, detail_id=record_id)
|
||
|
||
# 3. 构建处理和写入函数
|
||
detail_process_fn = self._build_detail_process_fn(spec)
|
||
detail_write_fn = self._build_detail_write_fn(spec, source_file)
|
||
# → 写入 detail_target_table，使用 _insert_records_schema_aware()
|
||
|
||
# 4. 执行管道
|
||
detail_result = detail_pipeline.run(
|
||
detail_requests, detail_process_fn, detail_write_fn,
|
||
)
|
||
self.db.commit()
|
||
```
|
||
|
||
### 3.3 详情响应处理（需自定义 process_fn）
|
||
|
||
默认的 `_build_detail_process_fn` 从 `response.get("records", [])` 提取记录。对于团购详情接口，需要自定义字段提取逻辑：
|
||
|
||
- 从 `data.groupPurchasePackage` 提取结构化字段（`package_name`、`duration`、`start_time`、`end_time` 等）
|
||
- 从 `data.groupPurchasePackage.tableAreaId` / `tableAreaNameList` 提取台区数组为 JSONB
|
||
- 从 `data.packageCouponAssistants` 提取助教服务关联数组为 JSONB
|
||
- 从 `data.grouponSiteInfos` 提取关联门店数组为 JSONB
|
||
- 从 `data.packagePackageService` 和 `data.packageCouponDetailsList` 提取为 JSONB
|
||
- 计算 `content_hash`，保留完整原始响应为 `payload`
|
||
|
||
实现方式：在 `ODS_GROUP_PACKAGE` 任务中覆盖 `_build_detail_process_fn`，或在 `OdsTaskSpec` 中扩展 `detail_process_fn` 回调。
|
||
|
||
### 3.4 详情数据写入（复用 _insert_records_schema_aware）
|
||
|
||
`_build_detail_write_fn` 已内置调用 `_insert_records_schema_aware()`，按目标表结构动态写入，支持 ON CONFLICT UPSERT。写入目标为 `ods.group_buy_package_details`，PK = `coupon_id`。
|
||
|
||
### 3.5 DWD 加载扩展
|
||
|
||
文件：`apps/etl/connectors/feiqiu/tasks/dwd/dwd_load_task.py`
|
||
|
||
在 `_merge_dim_scd2()` 处理 `dim_groupbuy_package_ex` 时，需要额外从 `ods.group_buy_package_details` 读取详情字段并合并到 ODS 快照中：
|
||
|
||
```python
|
||
# 伪代码：在 DWD 加载 dim_groupbuy_package_ex 时
|
||
def _load_groupbuy_package_ex(self, cur, now):
|
||
# 1. 从 ods.group_buy_packages 读取列表数据（现有逻辑）
|
||
# 2. 从 ods.group_buy_package_details 读取详情数据
|
||
# 3. 通过 coupon_id = id 关联，将详情字段合并到 ODS 快照
|
||
# 4. 执行 SCD2 合并（现有 _merge_dim_scd2 逻辑）
|
||
```
|
||
|
||
新增字段映射（`ods.group_buy_package_details` → `dwd.dim_groupbuy_package_ex`）：
|
||
|
||
| ODS 详情字段 | DWD 扩展表字段 | 类型 |
|
||
|-------------|---------------|------|
|
||
| `table_area_ids` | `table_area_ids` | JSONB |
|
||
| `table_area_names` | `table_area_names` | JSONB |
|
||
| `assistant_services` | `assistant_services` | JSONB |
|
||
| `groupon_site_infos` | `groupon_site_infos` | JSONB |
|
||
|
||
## 4. 数据库变更
|
||
|
||
### 4.1 新建 ODS 表
|
||
|
||
```sql
|
||
-- db/etl_feiqiu/ods/group_buy_package_details.sql
|
||
CREATE TABLE IF NOT EXISTS ods.group_buy_package_details (
|
||
coupon_id BIGINT NOT NULL,
|
||
package_name TEXT,
|
||
duration INTEGER, -- 台费计时时长（秒）
|
||
start_time TIMESTAMPTZ, -- 可用日期开始
|
||
end_time TIMESTAMPTZ, -- 可用日期结束
|
||
add_start_clock TEXT, -- 可用时段开始
|
||
add_end_clock TEXT, -- 可用时段结束
|
||
is_enabled INTEGER,
|
||
is_delete INTEGER,
|
||
site_id BIGINT,
|
||
tenant_id BIGINT,
|
||
create_time TIMESTAMPTZ,
|
||
creator_name TEXT,
|
||
-- JSONB 数组字段
|
||
table_area_ids JSONB, -- [2791960001957765, ...]
|
||
table_area_names JSONB, -- ["A区", ...]
|
||
assistant_services JSONB, -- [{skillId, assistantLevel, assistantDuration}, ...]
|
||
groupon_site_infos JSONB, -- [{siteId, siteName}, ...]
|
||
package_services JSONB, -- 待调研，可能为空
|
||
coupon_details_list JSONB, -- 待调研，可能为空
|
||
-- ETL 元数据
|
||
content_hash TEXT,
|
||
payload JSONB, -- 完整原始响应
|
||
fetched_at TIMESTAMPTZ DEFAULT now(),
|
||
-- 主键
|
||
CONSTRAINT pk_group_buy_package_details PRIMARY KEY (coupon_id)
|
||
);
|
||
|
||
COMMENT ON TABLE ods.group_buy_package_details IS '团购套餐详情 ODS：QueryPackageCouponInfo 原始数据';
|
||
```
|
||
|
||
### 4.2 DWD 扩展表 ALTER
|
||
|
||
```sql
|
||
-- db/etl_feiqiu/migrations/2026-03-05__add_detail_fields_to_dim_groupbuy_package_ex.sql
|
||
ALTER TABLE dwd.dim_groupbuy_package_ex
|
||
ADD COLUMN IF NOT EXISTS table_area_ids JSONB,
|
||
ADD COLUMN IF NOT EXISTS table_area_names JSONB,
|
||
ADD COLUMN IF NOT EXISTS assistant_services JSONB,
|
||
ADD COLUMN IF NOT EXISTS groupon_site_infos JSONB;
|
||
|
||
COMMENT ON COLUMN dwd.dim_groupbuy_package_ex.table_area_ids IS '可用台区 ID 列表（来自详情接口 tableAreaId）';
|
||
COMMENT ON COLUMN dwd.dim_groupbuy_package_ex.table_area_names IS '可用台区名称列表（来自详情接口 tableAreaNameList）';
|
||
COMMENT ON COLUMN dwd.dim_groupbuy_package_ex.assistant_services IS '助教服务关联（来自详情接口 packageCouponAssistants）';
|
||
COMMENT ON COLUMN dwd.dim_groupbuy_package_ex.groupon_site_infos IS '关联门店信息（来自详情接口 grouponSiteInfos）';
|
||
```
|
||
|
||
## 5. 线程模型详细设计
|
||
|
||
详情阶段复用 `UnifiedPipeline` 的三层并发架构，与列表阶段完全一致：
|
||
|
||
```
|
||
UnifiedPipeline #2（detail_mode）
|
||
│
|
||
│ 主线程（_request_loop）
|
||
│ ┌─────────────────────────────────────────┐
|
||
│ │ for req in _build_detail_requests(spec): │
|
||
│ │ if cancel_token.is_cancelled: break │
|
||
│ │ resp = api.post(req.endpoint, params) │
|
||
│ │ processing_queue.put((req, resp)) │
|
||
│ │ rate_limiter.wait(cancel_token.event) │
|
||
│ └─────────────────┬───────────────────────┘
|
||
│ │
|
||
│ processing_queue.put(SENTINEL × worker_count)
|
||
│ 等待所有 worker 完成
|
||
│ write_queue.put(SENTINEL)
|
||
│ 等待 writer 完成
|
||
│
|
||
├──▶ Worker Thread 1 ──┐
|
||
├──▶ Worker Thread 2 ──┤
|
||
│ │
|
||
│ processing_queue │
|
||
│ ┌─────────────┐ │
|
||
│ │ (req, resp) │───▶ detail_process_fn(resp)
|
||
│ │ (req, resp) │ → 提取字段、计算 content_hash
|
||
│ │ SENTINEL │ write_queue.put(records)
|
||
│ └─────────────┘ │
|
||
│ │
|
||
│ ▼
|
||
│ Write Thread
|
||
│ ┌──────────────────┐
|
||
│ │ write_queue │
|
||
│ │ batch=batch_size │──▶ _insert_records_schema_aware()
|
||
│ │ timeout=5s │ → UPSERT ods.group_buy_package_details
|
||
│ │ SENTINEL │
|
||
│ └──────────────────┘
|
||
│
|
||
▼
|
||
PipelineResult（detail_success / detail_failure / detail_skipped）
|
||
```
|
||
|
||
关键设计点（均由 `PipelineConfig` 统一控制，支持任务级覆盖）：
|
||
- `processing_queue`：`queue.Queue(maxsize=queue_size)`，满时阻塞主线程（背压机制）
|
||
- `write_queue`：`queue.Queue(maxsize=queue_size * 2)`
|
||
- Worker 数量：`PipelineConfig.workers`（默认 2）
|
||
- Writer 批量写入：累积 `batch_size` 条或超时 `batch_timeout` 秒后执行
|
||
- SENTINEL：`None` 对象，用于通知线程退出
|
||
- 取消信号：主线程检查 `cancel_token.is_cancelled`，`RateLimiter.wait()` 轮询 `cancel_token.event`
|
||
|
||
## 6. 取消信号
|
||
|
||
详情阶段的 `UnifiedPipeline` 实例与列表阶段共享同一个 `CancellationToken`。取消信号的上游传递链（Admin-web → Backend → Orchestration → CancellationToken）属于 etl-unified-pipeline 的架构范畴，本 spec 仅关注详情阶段收到取消信号后的行为：
|
||
|
||
1. `_request_loop` 检测到 `cancel_token.is_cancelled`，停止发送新请求
|
||
2. `RateLimiter.wait()` 在 0.5s 轮询周期内检测到取消，立即返回 `False`
|
||
3. 主线程发送 SENTINEL 到处理队列，等待已入队数据处理完成
|
||
4. `PipelineResult.cancelled = True`，`BaseOdsTask` 据此设置任务状态
|
||
|
||
## 7. 错误处理策略
|
||
|
||
错误处理由 `UnifiedPipeline` 统一管理，各阶段行为如下：
|
||
|
||
| 场景 | 处理方式 |
|
||
|------|---------|
|
||
| 单个 couponId 请求超时/HTTP 错误 | `_request_loop` 捕获异常，`request_failures++`，继续下一个 |
|
||
| 单个 couponId 返回 `code != 0` | 同上（API 层异常） |
|
||
| 连续失败超过阈值 | `_request_loop` 中断，`PipelineResult.status = "FAILED"` |
|
||
| Worker 线程处理异常 | `_process_worker` 捕获异常，`processing_failures++`，继续消费队列 |
|
||
| Writer 线程写入失败 | `_write_worker` 捕获异常，`write_failures++`，继续消费队列 |
|
||
| 取消信号到达 | 停止新请求，等待已入队数据处理完成，`cancelled = True` |
|
||
|
||
`BaseOdsTask.execute()` 在详情阶段完成后，将 `detail_result` 的统计信息合并到任务结果中，并记录每个失败项的错误日志。
|
||
|
||
连续失败阈值：`PipelineConfig.max_consecutive_failures`（默认 10，支持 `pipeline.ods_group_package.max_consecutive_failures` 任务级覆盖）。
|
||
|
||
## 8. 配置参数
|
||
|
||
详情阶段复用 `PipelineConfig` 统一配置体系，支持三级回退：`pipeline.ods_group_package.<key>` → `pipeline.<key>` → 硬编码默认值。
|
||
|
||
| 配置键 | 默认值 | 说明 |
|
||
|--------|--------|------|
|
||
| `pipeline.workers` | 2 | 处理线程数 |
|
||
| `pipeline.queue_size` | 100 | 处理队列容量 |
|
||
| `pipeline.batch_size` | 100 | 写入批量阈值 |
|
||
| `pipeline.batch_timeout` | 5.0 | 写入等待超时（秒） |
|
||
| `pipeline.rate_min` | 5.0 | RateLimiter 最小间隔（秒） |
|
||
| `pipeline.rate_max` | 20.0 | RateLimiter 最大间隔（秒） |
|
||
| `pipeline.max_consecutive_failures` | 10 | 连续失败中断阈值 |
|
||
|
||
如需为详情阶段单独调参，可通过 `pipeline.ods_group_package.*` 任务级覆盖（列表和详情阶段共享同一 `PipelineConfig` 实例）。
|
||
|
||
> 不再需要独立的 `DETAIL_FETCH_*` 配置参数。
|
||
|
||
## 9. 实施任务清单
|
||
|
||
### Task 1：新建 ODS 详情表 DDL
|
||
- 创建 `db/etl_feiqiu/ods/group_buy_package_details.sql`
|
||
- 执行 DDL 到测试库
|
||
- 需求覆盖：需求 3 验收标准 1-4
|
||
|
||
### Task 2：扩展 ODS_GROUP_PACKAGE 任务 — 配置详情拉取
|
||
- 在 `tasks/ods/ods_tasks.py` 的 `ODS_GROUP_PACKAGE` OdsTaskSpec 中添加 `detail_endpoint` 等配置
|
||
- 实现自定义的 `_build_detail_process_fn` 字段提取逻辑
|
||
- 实现自定义的 `_build_detail_write_fn` 写入逻辑
|
||
- 复用 `BaseOdsTask.execute()` 已有的详情拉取流程（`UnifiedPipeline` + `RateLimiter` + `CancellationToken`）
|
||
- 需求覆盖：需求 1 验收标准 1-8，需求 2 验收标准 1-6，需求 5 验收标准 1-4
|
||
|
||
### Task 3：DWD 扩展表 ALTER + 加载逻辑
|
||
- 执行 ALTER TABLE 到测试库
|
||
- 修改 DWD 加载逻辑，从详情 ODS 表读取并合并到扩展表
|
||
- 需求覆盖：需求 4 验收标准 1-5
|
||
|
||
### Task 4：数据调研 — 获取全部团购详情并分析未标注字段
|
||
- 编写一次性脚本调用详情接口获取全部数据
|
||
- 分析未标注字段的值分布
|
||
- 确认 `packagePackageService` 和 `packageCouponDetailsList` 是否有数据
|
||
- 根据分析结果调整 ODS/DWD 字段定义
|
||
- 需求覆盖：需求 3 验收标准 6，附录 B 调研 3、4
|
||
|
||
### Task 5：文档同步更新
|
||
- 更新 ODS DDL 文档、字段映射文档
|
||
- 更新 BD Manual
|
||
- 更新 DWD 全景文档
|
||
- 更新 README 任务清单
|
||
- 需求覆盖：需求 6 验收标准 1-4
|