# ODS → DWD 数据处理全流程报告

> 生成日期：2026-02-17
> 代码文件：`tasks/dwd/dwd_load_task.py`（DwdLoadTask，约 1700 行）
> 任务代码：`DWD_LOAD_FROM_ODS`

---

## 一、总体架构

### 1.1 一句话概括

DWD 层的核心任务 `DWD_LOAD_FROM_ODS` 做的事情很简单：**从 ODS 表读取最新数据，清洗后写入 DWD 表**。
维度表（dim_*）用"缓慢变化维度"策略保留历史版本，事实表用"增量追加"策略只写新数据。

### 1.2 类继承关系

```
BaseTask（基类：时间窗口、E/T/L 模板）
  └── DwdLoadTask（DWD 装载：维度合并 + 事实增量）
```

> ⚠️ 代码中还有一个 `BaseDwdTask`（继承 BaseTask），但 `DwdLoadTask` 并不继承它。
> `BaseDwdTask` 是早期遗留的死代码，其 `iter_ods_rows()` 和 `parse_payload()` 方法从未被调用。

### 1.3 处理规模

| 类别 | 数量 | 说明 |
|------|------|------|
| 维度表映射 | 17 对 | 含 `_ex` 扩展表，如 dim_site / dim_site_ex |
| 事实表映射 | 23 对 | 含 `_ex` 扩展表，如 dwd_payment / dwd_refund_ex |
| 合计 | 40 对 | 每对 = 1 个 DWD 表 ← 1 个 ODS 源表 |

### 1.4 事务模型

**每张表一次独立事务**。好处是：某张表失败不影响其他表。

```
表 A → 成功 → commit ✓
表 B → 失败 → rollback，记录错误，继续
表 C → 成功 → commit ✓
最终返回：{tables: [A统计, C统计], errors: [B错误]}
```

---

## 二、主流程（通俗版）

### 2.1 整体流水线

```
┌─────────────────────────────────────────────────────────┐
│  BaseTask.execute()                                     │
│  ┌──────────┐    ┌──────────┐    ┌──────────────────┐   │
│  │ extract  │ →  │transform │ →  │      load        │   │
│  │(返回now) │    │ (透传)   │    │ (遍历TABLE_MAP)  │   │
│  └──────────┘    └──────────┘    └──────────────────┘   │
│                                         │               │
│                          ┌──────────────┼───────────┐   │
│                          ▼              ▼           │   │
│                    dim_* 表?        其他表?         │   │
│                          │              │           │   │
│                    _merge_dim    _merge_fact_        │   │
│                          │       increment          │   │
│                    ┌─────┴─────┐        │           │   │
│                    ▼           ▼        ▼           │   │
│              有SCD2列?    无SCD2列?   增量写入       │   │
│                    │           │        +           │   │
│              _merge_dim   _merge_dim   缺失回补     │   │
│              _scd2        _type1_                    │   │
│                           upsert                    │   │
│                                                     │   │
└─────────────────────────────────────────────────────────┘
```

### 2.2 举个例子：一条会员记录的旅程

假设上游 SaaS 系统中，会员"张三"的手机号从 `138xxxx1234` 改成了 `139xxxx5678`。

**第 1 步：ODS 层已有数据**

ODS 表 `ods.member_profiles` 中可能有多条张三的记录（每次抓取一条）：

| id | name | phone | fetched_at | is_delete | content_hash |
|----|------|-------|------------|-----------|--------------|
| 1001 | 张三 | 138xxxx1234 | 2026-02-15 10:00 | 0 | abc123... |
| 1001 | 张三 | 138xxxx1234 | 2026-02-16 10:00 | 0 | abc123... |
| 1001 | 张三 | 139xxxx5678 | 2026-02-17 10:00 | 0 | def456... |

**第 2 步：DWD 取最新快照**

DwdLoadTask 用 `DISTINCT ON (id) ... ORDER BY fetched_at DESC` 只取最新一条：

| id | name | phone |
|----|------|-------|
| 1001 | 张三 | 139xxxx5678 |

**第 3 步：与 DWD 当前版本对比**

DWD 表 `dwd.dim_member` 当前版本（`scd2_is_current=1`）：

| member_id | name | phone | scd2_version | scd2_is_current |
|-----------|------|-------|-------------|-----------------|
| 1001 | 张三 | 138xxxx1234 | 1 | 1 |

对比发现 `phone` 列不同 → 有变更。

**第 4 步：关闭旧版 + 插入新版**

```sql
-- 关闭旧版本
UPDATE dwd.dim_member
SET scd2_end_time = '2026-02-17 10:30', scd2_is_current = 0
WHERE member_id = 1001 AND scd2_is_current = 1;

-- 插入新版本
INSERT INTO dwd.dim_member (member_id, name, phone, scd2_start_time, scd2_end_time, scd2_is_current, scd2_version)
VALUES (1001, '张三', '139xxxx5678', '2026-02-17 10:30', '9999-12-31', 1, 2);
```

最终 DWD 表中有两条记录，完整保留了变更历史：

| member_id | name | phone | scd2_version | scd2_is_current | scd2_start_time | scd2_end_time |
|-----------|------|-------|-------------|-----------------|-----------------|---------------|
| 1001 | 张三 | 138xxxx1234 | 1 | 0 | 2026-02-15 | 2026-02-17 10:30 |
| 1001 | 张三 | 139xxxx5678 | 2 | 1 | 2026-02-17 10:30 | 9999-12-31 |

---

## 三、维度表处理详解

### 3.1 两种策略的选择

代码根据 DWD 表是否包含 SCD2 列自动选择策略：

| DWD 表列中有 scd2_* 列？ | 策略 | 方法 | 通俗解释 |
|--------------------------|------|------|----------|
| 是 | SCD2 合并 | `_merge_dim_scd2()` | 保留历史：旧版本标记"过期"，新版本插入 |
| 否 | Type1 Upsert | `_merge_dim_type1_upsert()` | 直接覆盖：有变更就更新，不保留历史 |

> SCD2 列集合：`scd2_start_time`、`scd2_end_time`、`scd2_is_current`、`scd2_version`
>
> **实际情况**：经 DDL 验证，当前所有 17 张维度表都有 SCD2 列，所有 23 张事实表都没有。
> 因此 Type1 Upsert 分支在当前架构下不会被触发，属于防御性代码。

### 3.2 ODS 最新快照选取（两种策略共用）

不管哪种策略，第一步都是从 ODS 取"每个业务主键的最新一条记录"：

```sql
SELECT DISTINCT ON (业务主键)
    <列映射表达式>
FROM ods.<表>
WHERE fetched_at IS NOT NULL
ORDER BY 业务主键, fetched_at DESC NULLS LAST
```

**为什么要这样做？** 因为 ODS 是"追加写入"的，同一个会员/商品可能被抓取了很多次。
我们只关心最新状态，所以用 `DISTINCT ON` + `ORDER BY fetched_at DESC` 去重。

### 3.3 列映射机制（FACT_MAPPINGS）

ODS 表的列名和 DWD 表的列名经常不一样（上游 API 用驼峰命名，DWD 用下划线命名），
所以需要一个映射表来"翻译"。这个映射表叫 `FACT_MAPPINGS`（虽然名字里有 FACT，但维度表也用它）。

映射优先级：
1. **显式映射**（FACT_MAPPINGS 中定义的）→ 最高优先级
2. **同名映射**（ODS 和 DWD 列名相同）→ 自动匹配
3. **主键兜底**（DWD 主键在 ODS 中找不到，但 ODS 有 `id` 列）→ 自动映射到 `id`

举例：`dwd.dwd_settlement_head` 的映射

```python
# ODS 源表 settlement_records 用的是驼峰无下划线风格
"dwd.dwd_settlement_head": [
    ("order_settle_id", "id", None),           # 简单重命名
    ("site_name", "sitename", None),           # 驼峰 → 下划线
    ("pay_time", "paytime", None),             # 同上
    ("consume_money", "consumemoney", None),   # 同上
    ...
]
```

支持的映射类型：

| 类型 | 示例 | 说明 |
|------|------|------|
| 简单重命名 | `("table_id", "id", None)` | ODS 的 `id` → DWD 的 `table_id` |
| JSON 提取 | `("shop_name", "siteprofile->>'shop_name'", None)` | 从 JSONB 字段中提取值 |
| 类型转换 | `("longitude", "siteprofile->>'longitude'", "numeric")` | 提取后转数值 |
| 布尔转换 | `("is_first_limit", "is_first_limit", "boolean")` | 转布尔 |
| 日期截断 | `("pay_date", "pay_time", "date")` | 时间戳 → 日期 |
| SQL 表达式 | `("is_leaf", "CASE WHEN ... THEN 1 ELSE 0 END", None)` | 计算列 |

### 3.4 变更检测（_values_equal）

对比 ODS 最新值和 DWD 当前值时，不能简单用 `==`，因为类型经常不一致。
代码做了 4 层归一化：

| 归一化 | 场景 | 举例 |
|--------|------|------|
| 空值归一化 | 空字符串 `""` 和 `None` 视为相同 | `"" == None → True` |
| 日期时间归一化 | 朴素时间和带时区时间统一比较 | `2026-02-17T10:00 == 2026-02-17T10:00+08:00 → True` |
| 布尔归一化 | 字符串/数字/布尔统一 | `"true" == True == 1 → True` |
| 数值归一化 | 字符串数字和 Decimal 统一 | `"3.14" == Decimal("3.14") → True` |

> 归一化顺序：空值 → 日期 → 布尔 → 数值 → 最后才用 `==`

### 3.5 SCD2 批量优化

早期代码是逐行处理（`_upsert_scd2_row`），现在改为批量处理：

1. **预加载**：一次性 `SELECT *` 读取 DWD 当前版本到内存
2. **批量对比**：在 Python 中逐行对比，收集需要关闭/插入的主键
3. **批量关闭**：单主键用 `ANY` 数组，复合主键用 `execute_batch`
4. **批量插入**：用 `execute_values` 一次性插入所有新版本

> 旧的逐行方法 `_upsert_scd2_row` / `_close_current_dim` / `_insert_dim_row` 仍保留在代码中，
> 但已不被主流程调用（属于遗留代码）。

### 3.6 特殊处理：商品分类展开

`dwd.dim_goods_category` 表比较特殊。ODS 源表 `stock_goods_category_tree` 中，
每条记录的 `categoryboxes` 字段是一个 JSON 数组，包含子分类。

代码通过 `_expand_goods_category_rows()` 将父子关系展开为平铺行：

```
原始 ODS 记录：
  { id: 1, category_name: "饮品", categoryboxes: [{id: 11, category_name: "碳酸饮料"}, {id: 12, category_name: "果汁"}] }

展开后：
  { id: 1,  category_name: "饮品",     category_level: 1, is_leaf: 0, pid: 0 }
  { id: 11, category_name: "碳酸饮料", category_level: 2, is_leaf: 1, pid: 1 }
  { id: 12, category_name: "果汁",     category_level: 2, is_leaf: 1, pid: 1 }
```

---

## 四、事实表处理详解

### 4.1 增量机制：水位线（Watermark）

事实表不像维度表那样需要"对比变更"，而是**只写新数据**。
怎么知道哪些是"新数据"？靠水位线。

**统一使用 `fetched_at` 作为增量列**（代码中硬编码 `order_col = "fetched_at"`）。

两种模式：

| 模式 | 触发条件 | SQL 过滤 |
|------|----------|----------|
| 窗口覆盖模式 | 配置了 `run.window_override.start/end` | `WHERE fetched_at >= <start> AND fetched_at < <end>` |
| 自动水位模式 | 未配置窗口覆盖 | `WHERE fetched_at > <watermark>` |

> **通俗理解**：水位线 = "上次处理到哪了"的标记。
>
> **重要区分**：水位线仅在"自动水位模式"下起作用（即没有配置 `run.window_override`）。
> 如果你手动指定了时间窗口（比如 `start=08:00, end=10:00`），代码直接用你给的窗口做
> `WHERE fetched_at >= start AND fetched_at < end`，完全不查水位线。
>
> 在自动水位模式下，水位线之前的数据不参与处理。取出的记录通过 `ON CONFLICT DO UPDATE`
> 写入 — 新记录直接插入，已存在的记录如果有变更就更新。

水位线计算（`_get_fact_watermark`）：

```
1. DWD 表有 fetched_at 列？ → SELECT MAX(fetched_at) FROM dwd表
2. DWD 表无 fetched_at 但有主键？ → JOIN ODS 取 MAX(ods.fetched_at)
3. 都不行？ → 兜底 "1970-01-01"（相当于全量）
```

### 4.2 举个例子：一笔支付记录的写入

假设 ODS 表 `ods.payment_transactions` 新增了一笔支付：

| id | pay_time | amount | site_id | fetched_at |
|----|----------|--------|---------|------------|
| 5001 | 2026-02-17 14:00 | 88.00 | 101 | 2026-02-17 14:05 |

DWD 表 `dwd.dwd_payment` 上次水位线是 `2026-02-17 12:00`。

**第 1 步**：计算水位线 → `MAX(fetched_at) = 2026-02-17 12:00`

**第 2 步**：查询增量数据

```sql
SELECT "id" AS "payment_id", "pay_time"::date AS "pay_date", "tenant_id", ...
FROM ods.payment_transactions
WHERE fetched_at > '2026-02-17 12:00'
  AND fetched_at IS NOT NULL
```

**第 3 步**：写入 DWD（带冲突处理）

```sql
INSERT INTO dwd.dwd_payment (payment_id, pay_date, tenant_id, ...)
SELECT ...
ON CONFLICT (payment_id) DO UPDATE SET pay_date = EXCLUDED.pay_date, ...
WHERE dwd.dwd_payment.pay_date IS DISTINCT FROM EXCLUDED.pay_date
   OR dwd.dwd_payment.tenant_id IS DISTINCT FROM EXCLUDED.tenant_id
   OR ...
RETURNING (xmax = 0) AS inserted
```

### 4.3 快照去重

代码通过 `snapshot_mode = "content_hash" in ods_set` 判断是否启用快照去重。
经过 ODS 去重标准化改造后，所有 23 张 ODS 表都有 `content_hash` 列，
因此 `snapshot_mode` 对所有表都是 True，事实表始终使用 `DISTINCT ON` 取最新快照：

```sql
SELECT DISTINCT ON (主键) ... FROM ods表 WHERE fetched_at > 水位线 ORDER BY 主键, fetched_at DESC
```

### 4.4 写入策略

由于 `snapshot_mode` 始终为 True，`dwd.fact_upsert` 默认为 True，
实际上所有事实表都走同一条路径：

```sql
INSERT INTO dwd表 (列...)
SELECT DISTINCT ON (pk) ... FROM ods表 WHERE fetched_at > 水位线 ORDER BY pk, fetched_at DESC
ON CONFLICT (pk) DO UPDATE SET col1 = EXCLUDED.col1, col2 = EXCLUDED.col2, ...
WHERE dwd表.col1 IS DISTINCT FROM EXCLUDED.col1
   OR dwd表.col2 IS DISTINCT FROM EXCLUDED.col2
   OR ...
RETURNING (xmax = 0) AS inserted
```

> `IS DISTINCT FROM` 确保只有真正有变更的行才会被更新（避免无意义的写放大）。
> `RETURNING (xmax = 0)` 用于区分新增（xmax=0）和更新（xmax≠0），统计用。
>
> 代码中保留了 `dwd.fact_upsert=False` 时走 `DO NOTHING` 的分支，
> 以及 `snapshot_mode=False` 时不做 `DISTINCT ON` 的分支，但在当前架构下都不会触发。

### 4.5 缺失主键回补（_insert_missing_by_pk）

在主增量写入完成后，代码会对**所有事实表**额外执行一次"回补"操作：

```sql
INSERT INTO dwd表 (列...)
SELECT 列... FROM ods表 o
LEFT JOIN dwd表 d ON d.pk = o.pk
WHERE d.pk IS NULL                    -- DWD 中不存在的记录
  AND o.fetched_at > <watermark>      -- 同样受水位线/窗口约束
ON CONFLICT (pk) DO NOTHING
```

#### 什么时候会发生缺失？

**场景 1：主增量写入用了 DISTINCT ON，丢弃了同 ID 的旧版本**

这是最常见的触发场景。假设 ODS 中同一个助教服务记录 id=5001 有两条：

| id | fetched_at | site_id | amount |
|----|------------|---------|--------|
| 5001 | 2026-02-17 10:00 | 101 | 50.00 |
| 5001 | 2026-02-17 10:30 | 101 | 60.00 |

主增量写入的 `DISTINCT ON (id) ORDER BY fetched_at DESC` 只取了 10:30 那条。
如果这条因为某种原因写入失败（比如类型转换异常、NULL 主键等），id=5001 就丢了。
回补的 LEFT JOIN 会发现 DWD 中没有 id=5001，于是从 ODS 补入。

**场景 2：ODS 层收到"迟到"的数据**

```
时间线：
10:00  DWD 装载运行，水位线 = MAX(fetched_at) = 09:00
       → 查 ODS WHERE fetched_at > 09:00
       → 写入了 id=5001, 5002, 5003
       → 新水位线 = 10:00

10:05  ODS 层收到一条迟到数据：
       id=4999, fetched_at=10:02 (在新水位线之后)
       但这条数据的业务时间 pay_time=08:55 (很早的订单)

10:30  DWD 装载再次运行，水位线 = 10:00
       → 主查询 WHERE fetched_at > 10:00 → 取到 id=4999
       → 正常写入，不需要回补
```

但如果迟到数据的 `fetched_at` 恰好在水位线之前（比如手动回灌历史数据时
`fetched_at` 保留了原始值），主查询就会漏掉它。此时回补的 LEFT JOIN 也受
`fetched_at > watermark` 约束，同样捞不到。

**真正能兜底的是** `overlap_seconds`（默认 600 秒）：下次运行时水位线会回退 10 分钟，
覆盖到之前可能遗漏的区间。

**场景 3：并发写入冲突**

如果两个 DWD 装载进程同时运行（虽然不推荐），主写入的 `ON CONFLICT DO UPDATE`
可能因为锁竞争而跳过某些行。回补的 `ON CONFLICT DO NOTHING` 可以在下次运行时补齐。

#### 严格时间窗口下回补基本不触发

如果你的任务严格规范了时间窗口（比如明确获取 8:00-10:00 的数据，ODS 任务从 10:01 启动），
那么：

1. ODS 层在 10:01 抓取的数据，`fetched_at` 都是 10:01 左右
2. DWD 任务配置 `window_override.start=08:00, end=10:00`
3. 主增量写入：`WHERE fetched_at >= 08:00 AND fetched_at < 10:00`
4. 回补也受同样的窗口约束：`WHERE o.fetched_at >= 08:00 AND o.fetched_at < 10:00`

在这种模式下，ODS 和 DWD 处理的数据范围完全一致，主增量写入已经覆盖了所有记录，
回补的 LEFT JOIN 查出来的结果集几乎一定是空的（`d.pk IS NULL` 不成立）。

**结论**：严格时间窗口 + ODS 任务先于 DWD 任务完成 = 回补步骤变成纯安全网，
实际不产生写入。这是最稳健的运行方式。

> ⚠️ 代码中定义了 `FACT_MISSING_FILL_TABLES = {"dwd.dwd_assistant_service_log"}`，
> 注释说"对于会出现回补旧记录的事实表"，但实际上 `_insert_missing_by_pk` 对所有事实表都执行，
> 并没有检查是否在 `FACT_MISSING_FILL_TABLES` 中。这是一个**定义与实际行为不一致**的问题。

### 4.6 类型自动转换

当 DWD 列是数值类型（integer/numeric/bigint 等）但 ODS 列是文本类型时，
代码自动添加类型转换：

```sql
CAST(NULLIF(CAST("col" AS text), '') AS numeric)::integer
```

> `NULLIF(..., '')` 确保空字符串不会导致转换失败，而是变成 NULL。

---

## 五、完整参数清单

### 5.1 代码级常量

| 常量 | 值 | 作用 |
|------|-----|------|
| `TABLE_MAP` | 40 对映射 | DWD 表 → ODS 源表的完整映射 |
| `SCD_COLS` | `{scd2_start_time, scd2_end_time, scd2_is_current, scd2_version}` | SCD2 元数据列集合 |
| `FACT_ORDER_CANDIDATES` | `[pay_time, create_time, update_time, occur_time, settle_time, start_use_time, fetched_at]` | 增量排序候选列（⚠️ 实际未使用，见已知问题） |
| `FACT_MISSING_FILL_TABLES` | `{dwd.dwd_assistant_service_log}` | 需要回补的事实表（⚠️ 实际未检查，见已知问题） |
| `FACT_MAPPINGS` | 大字典 | 每张 DWD 表的列映射规则 |

### 5.2 运行时配置

| 配置项 | 来源 | 默认值 | 作用 |
|--------|------|--------|------|
| `dwd.only_tables` | AppConfig / 环境变量 `DWD_ONLY_TABLES` | 空（处理所有表） | 逗号分隔的表名列表，限定只处理指定表 |
| `dwd.fact_upsert` | AppConfig / 环境变量 `DWD_FACT_UPSERT` | `True` | 事实表冲突时是否更新（False 则 DO NOTHING） |
| `run.window_override.start` | AppConfig | 无 | 手动指定时间窗口起点 |
| `run.window_override.end` | AppConfig | 无 | 手动指定时间窗口终点 |
| `app.timezone` | AppConfig | `Asia/Shanghai` | 时区（影响日期时间归一化） |

### 5.3 BaseTask 继承的配置

| 配置项 | 默认值 | 作用 |
|--------|--------|------|
| `run.window_minutes.default_idle` | 180 | 闲时窗口分钟数 |
| `run.window_minutes.default_busy` | 30 | 忙时窗口分钟数 |
| `run.idle_window.start` / `end` | `04:00` / `16:00` | 闲时判定区间 |
| `run.overlap_seconds` | 600 | 窗口重叠秒数（防漏） |
| `app.store_id` | — | 门店 ID |

---

## 六、已知问题与改进建议

### 6.1 `BaseDwdTask` 是死代码

`tasks/dwd/base_dwd_task.py` 中的 `BaseDwdTask` 继承了 `BaseTask`，
提供了 `iter_ods_rows()` 和 `parse_payload()` 方法。
但 `DwdLoadTask` 直接继承 `BaseTask`，从未使用 `BaseDwdTask`。

**建议**：删除 `base_dwd_task.py`，或在未来重构时决定是否复用。

### 6.2 `_build_column_mapping` 引用了未定义变量

`_build_column_mapping()` 方法的参数签名是 `(self, dwd_table, pk_cols, ods_cols)`，
但方法体内引用了 `ods_table`（第 711 行）和 `cur`（第 713 行），这两个变量不在参数中。

```python
# 第 711 行 — ods_table 未定义
self.logger.error("跳过 %s：ODS 表 %s 缺少 fetched_at 列", dwd_table, ods_table)
# 第 713 行 — cur 未定义
self._log_missing_fetched_at(cur, ods_table)
```

**为什么没出错？** 因为所有 ODS 表都有 `fetched_at` 列，`if "fetched_at" not in ods_set` 永远为 False，
这段代码从未被执行。但如果未来某张 ODS 表缺少 `fetched_at`，会抛出 `NameError`。

**建议**：将 `ods_table` 和 `cur` 加入参数签名，或将这段防御逻辑移到调用方。

### 6.3 `_pick_order_column` 和 `FACT_ORDER_CANDIDATES` 未使用

代码定义了 `FACT_ORDER_CANDIDATES`（7 个候选增量列）和 `_pick_order_column()` 方法，
用于智能选择事实表的增量排序列。

但 `_merge_fact_increment()` 中硬编码了 `order_col = "fetched_at"`，
完全没有调用 `_pick_order_column()`。

**影响**：所有事实表都用 `fetched_at` 做增量过滤，即使表中有更合适的业务时间列（如 `pay_time`）。
这在大多数场景下是可以的（因为 `fetched_at` 是单调递增的），但可能导致：
- 回溯窗口时，`fetched_at` 不能精确反映业务时间
- 水位线可能比实际业务时间偏晚

**建议**：评估是否需要启用 `_pick_order_column`，或明确删除这段未使用代码。

### 6.4 `FACT_MISSING_FILL_TABLES` 定义与行为不一致

`FACT_MISSING_FILL_TABLES` 只包含 `dwd.dwd_assistant_service_log`，
注释说"对于会出现回补旧记录的事实表"。

但 `_insert_missing_by_pk()` 在 `_merge_fact_increment()` 末尾对**所有事实表**无条件调用，
没有检查 `dwd_table in self.FACT_MISSING_FILL_TABLES`。

**影响**：
- 正面：所有事实表都能自动回补缺失记录，更安全
- 负面：对不需要回补的表执行了额外的 LEFT JOIN 查询，有性能开销

**建议**：要么删除 `FACT_MISSING_FILL_TABLES`（承认所有表都需要回补），
要么加上条件判断（只对指定表执行回补）。

### 6.5 SCD2 逐行方法仍保留

`_upsert_scd2_row()`、`_close_current_dim()`、`_insert_dim_row()` 三个方法
是早期逐行处理的实现，现在已被批量方法（`_close_current_dim_bulk` / `_insert_dim_rows_bulk`）替代。

这些方法仍保留在代码中但不被主流程调用。

**建议**：标记为 `@deprecated` 或删除，减少维护负担。

### 6.6 Type1 Upsert 中 SCD2 列的处理

`_merge_dim_type1_upsert()` 在构建 INSERT 值时，会为 SCD2 列填充默认值
（`scd2_start_time=now`、`scd2_version=1`），但在 ON CONFLICT DO UPDATE 时，
`scd2_start_time` 和 `scd2_version` 使用 `COALESCE(现有值, EXCLUDED值)` 保留已有值。

这意味着：如果一张表的 DDL 中有 SCD2 列但 `_merge_dim()` 判定走 Type1 路径
（理论上不会发生，因为有 SCD2 列就走 SCD2 路径），SCD2 列不会被错误覆盖。

这是一个防御性设计，不是 bug。

---

## 七、数据流总结图

```
┌──────────────────────────────────────────────────────────────────┐
│                        ODS 层（原始数据）                         │
│                                                                  │
│  ods.member_profiles          ods.payment_transactions           │
│  ods.settlement_records       ods.table_fee_transactions         │
│  ... (共 23 张 ODS 表)                                           │
└──────────────────────┬───────────────────────────────────────────┘
                       │
                       │  DwdLoadTask.load()
                       │  遍历 TABLE_MAP (40 对)
                       │
          ┌────────────┴────────────┐
          ▼                         ▼
    dim_* 前缀？              其他前缀？
          │                         │
    ┌─────┴─────┐             ┌─────┴─────┐
    ▼           ▼             ▼           ▼
  有SCD2列   无SCD2列      自动水位    窗口覆盖
    │           │             │           │
  SCD2合并   Type1         增量写入    范围写入
    │        Upsert           │           │
    │           │             └─────┬─────┘
    │           │                   │
    │           │             缺失回补
    │           │          (_insert_missing_by_pk)
    │           │                   │
    ▼           ▼                   ▼
┌──────────────────────────────────────────────────────────────────┐
│                        DWD 层（明细数据）                         │
│                                                                  │
│  dwd.dim_member (SCD2)        dwd.dwd_payment (增量)             │
│  dwd.dim_site (SCD2)          dwd.dwd_settlement_head (增量)     │
│  dwd.dim_table (SCD2)         dwd.dwd_refund (增量)              │
│  ... (共 17 维度 + 23 事实)                                      │
└──────────────────────────────────────────────────────────────────┘
```

---

## 八、与 ODS 层的对比

| 方面 | ODS 层 | DWD 层 |
|------|--------|--------|
| 数据来源 | 上游 SaaS API | ODS 表 |
| 写入方式 | 追加写入（INSERT） | 维度：SCD2/Upsert；事实：增量 INSERT |
| 去重机制 | content_hash + skip_unchanged | DISTINCT ON + ON CONFLICT |
| 历史保留 | 所有版本都保留 | 维度 SCD2 保留历史；事实只保留最新 |
| 列名风格 | 与上游 API 一致（驼峰/混合） | 标准化下划线命名 |
| 类型处理 | 原样存储（多为 text/jsonb） | 类型转换（numeric/boolean/date 等） |
| 软删除 | INSERT 删除版本行（is_delete=1） | 不处理（依赖 ODS 的 is_delete 标记） |
| 事务粒度 | 每个任务一次事务 | 每张表一次事务 |

---

## 九、优化方案分析

### 9.1 方案：DWD 增加 content_hash 列，用 hash 对比代替逐列对比

#### 当前痛点

维度表变更检测在 Python 层逐列对比（`_is_row_changed` → `_values_equal`），涉及 4 层类型归一化：

```
当前流程（维度表）：
1. SELECT * FROM dwd表 WHERE scd2_is_current=1  → 全量加载到 Python 内存
2. SELECT DISTINCT ON (pk) ... FROM ods表        → 全量加载到 Python 内存
3. 逐行逐列对比（空值/日期/布尔/数值 4 层归一化）
4. 收集变更主键 → 批量关闭旧版 + 批量插入新版
```

事实表虽然用 SQL 层 `IS DISTINCT FROM` 对比，但需要列举所有非主键列：

```sql
WHERE dwd.col1 IS DISTINCT FROM EXCLUDED.col1
   OR dwd.col2 IS DISTINCT FROM EXCLUDED.col2
   OR dwd.col3 IS DISTINCT FROM EXCLUDED.col3
   -- ... 可能有 20+ 列
```

#### 优化方案

在 DWD 表中增加 `content_hash` 列，直接复用 ODS 的 `content_hash`（基于 payload + is_delete 的 SHA-256）。

```
优化后流程（维度表）：
1. 一条 SQL 搞定：
   SELECT ods.pk, ods.content_hash
   FROM ods表 ods
   JOIN dwd表 dwd ON ods.pk = dwd.pk AND dwd.scd2_is_current = 1
   WHERE ods.content_hash != dwd.content_hash
2. 只对变更的主键执行关闭旧版 + 插入新版

优化后流程（事实表）：
INSERT INTO dwd表 (..., content_hash)
SELECT ..., content_hash FROM ods表
ON CONFLICT (pk) DO UPDATE SET ...
WHERE dwd表.content_hash IS DISTINCT FROM EXCLUDED.content_hash  -- 一列对比代替 20+ 列
```

#### 等价性风险与参数化设计

ODS 的 `content_hash` 基于原始 payload 计算，DWD 列是 payload 的确定性派生。
日常运行时"payload 不变 = DWD 列不变"成立，可以直接复用 hash。

但当 DWD 表结构变更（新增列、修改 FACT_MAPPINGS 映射规则）时，
同一个 payload 会产生不同的 DWD 列值，hash 等价关系被打破。
此时需要回退到逐列对比来刷新所有记录。

建议增加配置参数 `dwd.change_detection_mode`：

| 模式 | 值 | 维度表对比方式 | 事实表对比方式 | 适用场景 |
|------|-----|---------------|---------------|----------|
| hash 模式 | `"hash"`（默认） | `ods.content_hash != dwd.content_hash` | `WHERE content_hash IS DISTINCT FROM` | 日常增量运行 |
| column 模式 | `"column"` | 逐列对比（当前逻辑） | 逐列 `IS DISTINCT FROM`（当前逻辑） | DWD 表结构变更后的首次全量刷新 |

典型工作流：

```
日常运行：change_detection_mode = "hash"（快速）
    ↓
修改了 DWD 表结构或 FACT_MAPPINGS
    ↓
临时切换：change_detection_mode = "column" + window_override 覆盖全量时间范围
    ↓
全量刷新完成，所有 DWD 行的 content_hash 已与 ODS 对齐
    ↓
切回：change_detection_mode = "hash"
```

#### 可行性分析

| 方面 | 评估 |
|------|------|
| DDL 变更 | 所有 40 张 DWD 表需要 `ALTER TABLE ADD COLUMN content_hash VARCHAR(64)` |
| 索引 | 维度表建议加 `(pk, content_hash)` 复合索引；事实表主键索引已够用 |
| 代码改动 | `_merge_dim_scd2` 改为 hash 对比；`_merge_fact_increment` 简化 WHERE 条件；写入时同步存储 hash；新增 `change_detection_mode` 配置读取 |
| 回滚 | `ALTER TABLE DROP COLUMN content_hash`，代码回退即可 |

#### 预期收益

| 场景 | 当前 | 优化后 |
|------|------|--------|
| 维度表变更检测 | Python 全量加载 + 逐列对比 | 数据库层 hash 对比，不需要加载到内存 |
| 事实表冲突判断 | 20+ 列 IS DISTINCT FROM | 1 列 hash IS DISTINCT FROM |
| 网络传输 | 维度表需要传输全部列值 | 只需传输 pk + hash |

### 9.2 事实表只保留最新信息

当前设计已经是这个思路 — 事实表用 `ON CONFLICT DO UPDATE` 就是"有变更就覆盖"。
ODS 保留了所有历史快照，DWD 事实表只需要最新值。

加了 content_hash 后，覆盖逻辑可以更精准：只在 hash 不同时才触发 UPDATE，
避免了当前逐列 `IS DISTINCT FROM` 的开销。

### 9.3 ODS 索引利用情况

ODS 层已为所有 23 张表建立了 `(业务主键, fetched_at DESC)` 复合索引：

```sql
-- 示例
CREATE INDEX idx_ods_member_profiles_latest ON ods.member_profiles (id, fetched_at DESC);
CREATE INDEX idx_ods_settlement_records_latest ON ods.settlement_records (id, fetched_at DESC);
-- ... 共 23 个
```

DWD 装载时的查询模式与索引的匹配情况：

| 查询模式 | 是否命中索引 | 说明 |
|----------|-------------|------|
| `DISTINCT ON (id) ORDER BY id, fetched_at DESC` | ✅ 完美命中 | 复合索引的排序方向完全匹配 |
| `WHERE fetched_at > watermark` | ⚠️ 部分命中 | `fetched_at` 不是索引前缀列，PostgreSQL 可能选择 Seq Scan |
| `LEFT JOIN ON d.pk = o.pk`（回补） | ✅ 命中 | 主键索引 |

#### 可优化点

事实表的水位线过滤 `WHERE fetched_at > watermark` 在数据量大时，
当前的 `(id, fetched_at DESC)` 复合索引不是最优的。
可以考虑为高频/大表额外加一个 `(fetched_at)` 单列索引：

```sql
CREATE INDEX idx_ods_settlement_records_fetched ON ods.settlement_records (fetched_at);
CREATE INDEX idx_ods_payment_transactions_fetched ON ods.payment_transactions (fetched_at);
```

但需要权衡：额外索引会增加 ODS 写入时的开销（每次 INSERT 都要维护索引）。
考虑到 ODS 写入频率远高于 DWD 读取频率，这个优化需要通过 `EXPLAIN ANALYZE` 实测后再决定。

---

## 十、重构路线图

以下为经讨论确认的重构方向，按优先级排列。每项标注了改动范围和风险等级。

### 10.1 统一窗口模式，去掉水位线机制

#### 现状问题

事实表增量有两条路径：

| 模式 | 触发条件 | SQL |
|------|----------|-----|
| 窗口覆盖 | 配置了 `window_override` | `WHERE fetched_at >= start AND fetched_at < end` |
| 自动水位 | 未配置 `window_override` | `WHERE fetched_at > watermark`（只有下界，无上界） |

自动水位模式的问题：
- 需要额外查一次 DB 算水位线（`_get_fact_watermark` → `SELECT MAX(fetched_at)`）
- 没有上界，语义不如窗口模式清晰
- 与 ODS 层的窗口模式不一致（ODS 的 `_resolve_window` 始终算出 start/end）

#### 方案

统一为 `WHERE fetched_at >= start AND fetched_at < end`。`BaseTask._get_time_window()` 已经在自动模式下计算了 `window_start`（基于 cursor + overlap）和 `window_end`（= now），但 `DwdLoadTask.load()` 只在 `use_window=True`（配置了 window_override）时才传给 `_merge_fact_increment`。

改动点：
1. `load()` 中去掉 `use_window` 判断，始终传 `context.window_start / window_end`
2. 删除 `_get_fact_watermark()` 方法
3. `_merge_fact_increment` 中删除 watermark 分支
4. `_insert_missing_by_pk` 中删除 watermark 参数

| 方面 | 评估 |
|------|------|
| 改动量 | 小（约 30 行） |
| 风险 | 低 — BaseTask 已经算好了窗口 |
| 回滚 | 恢复 `use_window` 判断即可 |

### 10.2 删除回补机制（_insert_missing_by_pk）

#### 现状问题

回补的设计初衷是兜底"主增量写入漏掉的记录"。但分析后发现：

1. 主增量写入是一条 SQL（`INSERT ... SELECT ... FROM ods表`），如果某条记录类型转换失败，**整条 SQL 报错、整张表事务 rollback**，不存在"部分行静默丢失"的情况
2. 统一窗口模式后，ODS 和 DWD 处理的数据范围完全一致，回补的 LEFT JOIN 结果集几乎一定为空
3. 并发写入冲突在当前架构下不会发生（单进程顺序执行）

如果真出了类型转换错误，应该直接报错让运维介入，而不是靠回补静默兜底。

#### 方案

直接删除：
1. 删除 `_insert_missing_by_pk()` 方法
2. 删除 `_merge_fact_increment()` 末尾的回补调用
3. 删除 `FACT_MISSING_FILL_TABLES` 常量

| 方面 | 评估 |
|------|------|
| 改动量 | 小（删除约 100 行） |
| 风险 | 低 — 统一窗口后回补本身就是空操作 |
| 前置依赖 | 10.1（统一窗口模式） |

### 10.3 清理死代码和未使用常量

统一清理第六章列出的已知问题：

| 清理项 | 操作 |
|--------|------|
| `BaseDwdTask`（`base_dwd_task.py`） | 删除文件 |
| `_build_column_mapping` 中未定义变量 | 将 `ods_table` / `cur` 加入参数签名 |
| `_pick_order_column` + `FACT_ORDER_CANDIDATES` | 删除（已确认统一用 `fetched_at`） |
| `FACT_MISSING_FILL_TABLES` | 随 10.2 一起删除 |
| SCD2 逐行方法（`_upsert_scd2_row` 等 3 个） | 删除 |

### 10.4 跨层参数提取与层级控制

#### 现状问题

各层任务对配置的使用方式不统一：

| 层 | 基类 | 时间窗口 | 配置读取 |
|----|------|----------|----------|
| ODS | `BaseOdsTask`（继承 BaseTask） | 自己重写 `_resolve_window`（加了 MAX(fetched_at) 兜底） | `self.config.get("run.*")` |
| DWD | `DwdLoadTask`（继承 BaseTask） | 自己搞水位线 `_get_fact_watermark` | `self.config.get("dwd.*")` |
| DWS | `BaseDwsTask`（不继承 BaseTask） | 完全独立的时间范围计算 | `self.config.get("dws.*")` |

共用参数散落在 `run.*` 下，层专属参数各自为政。如果未来 CORE 层也需要 `overlap_seconds` 但值不同，没法区分。

#### 方案

配置结构改为层级覆盖：

```python
# defaults.py 新增
"layer": {
    "ods": {
        "overlap_seconds": 600,
        "window_minutes": {"default_busy": 30, "default_idle": 180},
        # ODS 层专属...
    },
    "dwd": {
        "overlap_seconds": 300,
        "change_detection_mode": "hash",
        "fact_upsert": True,
        # DWD 层专属...
    },
    "dws": {
        # DWS 层专属...
    },
    "core": {
        # CORE 层专属（预留）...
    },
}
```

`AppConfig.get()` 增加层级感知：

```python
def get_for_layer(self, layer: str, key: str, default=None):
    """先查 layer.<层>.<key>，没有再 fallback 到 run.<key>"""
    val = self.get(f"layer.{layer}.{key}")
    if val is not None:
        return val
    return self.get(f"run.{key}", default)
```

代码层级统一：

```
BaseTask（E/T/L 模板 + 统一时间窗口）
  ├── BaseOdsTask（ODS：API 抓取 + 快照去重 + 软删除）
  ├── DwdLoadTask（DWD：维度 SCD2 + 事实增量）
  ├── BaseDwsTask → 重构后继承 BaseTask（DWS：聚合计算）
  └── BaseCoreTask（CORE：预留）
```

| 方面 | 评估 |
|------|------|
| 改动量 | 中（`defaults.py` + `AppConfig` + 各层基类的 config 读取） |
| 风险 | 中 — 需要确保向后兼容（旧的 `run.*` 键仍可用） |
| 分步实施 | 先加 `layer.*` 结构和 `get_for_layer`，再逐层迁移 |

### 10.5 连接器（Connector）与多门店配置

#### 现状

当前配置是"一个 .env 对应一个门店"：

```
STORE_ID=2790685415443269
API_TOKEN=eyJhbG...（该门店的 Token）
```

如果飞球平台接入多个门店，需要为每个门店维护一份 .env 或手动切换 `STORE_ID` + `API_TOKEN`。

#### 需求

- 飞球平台（Connector）下有多个门店（Store）
- 每个门店有自己的 `store_id` 和 `api.token`，其他配置（DB、路径、窗口等）共用
- 日常运行以连接器为单位，自动遍历所有门店

#### 方案：连接器注册表 + 门店列表

在数据库 `meta` schema 中维护连接器和门店的注册信息：

```sql
-- 连接器注册表
CREATE TABLE meta.connector (
    connector_id    SERIAL PRIMARY KEY,
    connector_code  VARCHAR(50) NOT NULL UNIQUE,  -- 如 'feiqiu'
    display_name    VARCHAR(100),
    api_base_url    VARCHAR(255),                  -- 平台级 API 地址
    is_active       BOOLEAN DEFAULT TRUE,
    created_at      TIMESTAMPTZ DEFAULT now()
);

-- 门店注册表
CREATE TABLE meta.connector_store (
    store_id        BIGINT PRIMARY KEY,            -- 上游门店 ID
    connector_id    INT NOT NULL REFERENCES meta.connector(connector_id),
    store_name      VARCHAR(100),
    api_token       TEXT NOT NULL,                  -- 该门店的 Token（加密存储）
    is_active       BOOLEAN DEFAULT TRUE,
    created_at      TIMESTAMPTZ DEFAULT now()
);
```

配置体系变更：

```
当前：
  .env → STORE_ID + API_TOKEN（单门店）

改为：
  .env → CONNECTOR_CODE=feiqiu（连接器级别）
       → DB 连接信息、路径、窗口等（共用配置）
  meta.connector_store → 每个门店的 store_id + api_token（门店级别）
```

运行时流程：

```
1. 读取 CONNECTOR_CODE → 查 meta.connector 获取连接器信息
2. 查 meta.connector_store WHERE connector_id=? AND is_active=TRUE → 获取门店列表
3. 遍历门店列表：
   for store in stores:
       config.override("app.store_id", store.store_id)
       config.override("api.token", store.api_token)
       orchestrator.run_all_tasks()  # ODS → DWD → DWS
```

.env 中不再需要 `STORE_ID` 和 `API_TOKEN`，改为：

```env
# 连接器标识（替代原来的 STORE_ID + API_TOKEN）
CONNECTOR_CODE=feiqiu

# 以下保持不变
PG_DSN=postgresql://...
TIMEZONE=Asia/Shanghai
# ...
```

Token 安全性：
- Token 存在数据库中而非 .env 文件，便于集中管理和轮换
- 建议对 `api_token` 列做应用层加密（如 AES），密钥通过环境变量传入
- 日志中已有脱敏机制（`security.redact_keys` 包含 `token`）

向后兼容：
- 如果 .env 中仍有 `STORE_ID` + `API_TOKEN`，走旧的单门店模式
- 如果 .env 中有 `CONNECTOR_CODE`，走新的多门店模式
- 两种模式互斥，启动时检测并提示

| 方面 | 评估 |
|------|------|
| DDL 变更 | 新增 2 张 meta 表 |
| 代码改动 | 新增连接器加载逻辑 + orchestrator 遍历门店 + config 动态覆盖 |
| 改动量 | 中 |
| 风险 | 低 — 向后兼容，旧模式不受影响 |
| 前置依赖 | 无（可独立实施） |

### 10.6 实施优先级

```
第一步（低风险，立即可做）：
  10.1 统一窗口模式 → 10.2 删除回补 → 10.3 清理死代码
  三项连续做，改动集中在 dwd_load_task.py

第二步（中风险，需设计）：
  10.4 跨层参数提取 — 改 defaults.py + AppConfig + 各层基类

第三步（独立，可并行）：
  10.5 连接器与多门店 — 新增 meta 表 + 加载逻辑
  9.1 DWD content_hash 优化 — DDL + 对比逻辑改造
```
