Neo-ZQYY/apps/etl/connectors/feiqiu/docs/architecture/data_flow.md

数据流向详解：ODS → DWD → DWS

整体数据流

上游 SaaS API / 离线 JSON
        │
        ▼
┌───────────────────────────────────────┐
│  ODS 层（ods）                         │
│  操作数据存储 — 原始数据落地            │
│  保留源 payload，便于回溯              │
│  23 张 ODS 表，对应 23 个 API 端点     │
└───────────────┬───────────────────────┘
                │ DWD_LOAD_FROM_ODS
                ▼
┌───────────────────────────────────────┐
│  DWD 层（dwd）                         │
│  明细数据 — 清洗、标准化、关联          │
│  维度表走 SCD2（缓慢变化维度）          │
│  事实表按时间增量写入                   │
└───────┬───────────────┬───────────────┘
        │               │ CORE_DIM_SYNC
        │ DWS 汇总任务   ▼
        │       ┌───────────────────────┐
        │       │  Core 层（core）      │
        │       │  跨平台统一维度层      │
        │       │  dim_assistant / dim_ │
        │       │  member / dim_site /  │
        │       │  dim_table            │
        │       └───────────────────────┘
        ▼
┌───────────────────────────────────────┐
│  DWS 层（dws）                         │
│  数据服务 — 汇总、指标                  │
│  助教业绩 / 财务日报 / 会员分析         │
│  工资计算                               │
└───────────────┬───────────────────────┘
                │ INDEX 指数任务
                ▼
┌───────────────────────────────────────┐
│  INDEX 层（dws）                       │
│  自定义指数算法                         │
│  WBI / NCI / RS / OS / MS / ML         │
│  日快照存储（stat_date）               │
└───────────────────────────────────────┘

ODS 层（操作数据存储）

• Schema：ods
• 职责：从上游 SaaS API 抓取原始数据并落地，保留完整源 payload
• 数据来源：在线 API 抓取（APIClient）或离线 JSON 回放（LocalJsonClient）
• 任务模式：每个业务实体对应一个 ODS 任务（如 ODS_PAYMENT、ODS_MEMBER、ODS_SETTLEMENT_RECORDS 等），由 BaseOdsTask + OdsTaskSpec 声明式配置驱动，通过 _build_task_class() 工厂函数动态生成任务类
• 加载方式：_insert_records_schema_aware schema-aware 写入——运行时从 information_schema 读取目标表结构，按列名自动匹配，批量 upsert + 冲突处理（ON CONFLICT 策略可配置）

核心业务实体（23 个 ODS 任务）

助教账号档案（assistant_accounts_master）、结账记录（settlement_records）、台费计费流水（table_fee_transactions）、助教服务流水（assistant_service_records）、助教废除记录（assistant_cancellation_records）、门店商品销售流水（store_goods_sales_records）、支付流水（payment_transactions）、退款流水（refund_transactions）、平台/团购券核销（platform_coupon_redemption_records）、会员档案（member_profiles）、会员储值卡（member_stored_value_cards）、会员余额变动（member_balance_changes）、充值结算（recharge_settlements）、团购套餐定义（group_buy_packages）、团购套餐核销（group_buy_redemption_records）、库存汇总（goods_stock_summary）、库存变化记录（goods_stock_movements）、台桌维表（site_tables_master）、库存商品分类树（stock_goods_category_tree）、门店商品档案（store_goods_master）、台费折扣/调账（table_fee_discount_records）、租户商品档案（tenant_goods_master）、结账小票详情（settlement_ticket_details）。

DWD 层（明细数据）

• Schema：dwd
• 职责：对 ODS 原始数据进行清洗、标准化、关联，生成可分析的明细数据
• 核心任务：DWD_LOAD_FROM_ODS
• 质量检查：DWD_QUALITY_CHECK

维度处理（SCD2）

维度表采用 SCD2（缓慢变化维度 Type 2）策略，由 scd/ 模块处理：

• 会员维度（dim_member）
• 助教维度（dim_assistant）
• 商品维度（dim_product）
• 台桌维度（dim_table）
• 套餐维度（dim_package）

每条维度记录包含 valid_from、valid_to、is_current 字段，支持历史版本追溯。

事实表处理

事实表按时间增量写入，由 loaders/facts/ 中的加载器处理：

• 订单事实、支付事实、退款事实、小票明细、充值结算、台费流水等

DWS 层（数据服务）

• Schema：dws
• 职责：基于 DWD 明细数据进行汇总计算，输出业务指标和分析结果

汇总任务分类

类别	任务示例	建议频率
助教业绩	DWS_ASSISTANT_DAILY、DWS_ASSISTANT_MONTHLY	每小时 / 每日
财务日报	DWS_FINANCE_DAILY、DWS_FINANCE_INCOME_STRUCTURE	每小时
会员分析	DWS_MEMBER_CONSUMPTION、DWS_MEMBER_VISIT	每日
工资计算	DWS_ASSISTANT_SALARY	每月（月初）
指数算法	DWS_WINBACK_INDEX、DWS_NEWCONV_INDEX、DWS_RELATION_INDEX	每 2-4 小时
Core 维度同步	CORE_DIM_SYNC	每日（DWD_LOAD_FROM_ODS 之后）

自定义指数算法

系统实现了六个自定义业务指数，参数存储在 dws.cfg_index_parameters：

指数	全称	说明
WBI	Winback Index	召回指数
NCI	New Conversion Index	新客转化指数
RS	Relation Score	关系评分
OS	Overall Score	综合评分
MS	Member Score	会员评分
ML	Manual Ledger	人工台账

公共参数：percentile_lower/upper（分位截断锚点）、ewma_alpha（指数加权移动平均平滑系数）。

指数日快照（2026-04 重构）

三张指数表（relation_index / winback_index / newconv_index）新增 stat_date DATE NOT NULL DEFAULT CURRENT_DATE 字段，唯一约束扩展为 (site_id, member_id[, assistant_id], stat_date)。每次任务运行追加当日快照而非覆盖历史，为指数回测、历史演化分析提供原始数据基础。Winback 任务配套 365 天保留策略，清理超过 365 天的历史快照以控制表体积。

Core 维度层（跨平台统一维度）

• Schema：core
• 定位：跨平台统一维度层，屏蔽 ODS/DWD 多数据源（飞球 / 美团 / 后续新平台）的字段差异，供 app 视图层和业务库 FDW 统一引用
• 核心任务：CORE_DIM_SYNC（归属 DWS 层任务目录 tasks/dws/core_dim_sync_task.py）
• 上游依赖：DWD_LOAD_FROM_ODS（必须先完成 DWD 维度表 SCD2 写入）
• 同步范围：4 张维度表 — dim_assistant / dim_member / dim_site / dim_table
• 数据来源：DWD 层 scd2_is_current = 1 的当前版本快照
• 同步策略：TRUNCATE + 全量 INSERT（维度表行数较小，执行时间极短，避免增量合并复杂度）
• 起因：修复 app 视图在 core 维度为空时返回空结果的问题（关联审计 docs/audit/changes/2026-04-15__meituan-settle-core-sync.md）

Utility 工具任务

工具任务位于 tasks/utility/，不在日常编排链路中，由手动 CLI 触发，服务于 schema 初始化、历史数据回填和回测推演等一次性或周期性离线需求。

DWS_INDEX_BACKFILL（指数历史快照回填）

• 源码：tasks/utility/index_backfill_task.py
• 职责：逐天调用三个指数任务（DWS_RELATION_INDEX / DWS_WINBACK_INDEX / DWS_NEWCONV_INDEX）生成历史日快照，回填缺失的 stat_date 行
• 性能优化：三个指数任务实例在整轮回填中创建 1 次、循环 N 天复用，避免 N×3 次初始化开销（以 255 天为例避免 765 次初始化）
• 前置条件：三张指数表已完成日快照改造（stat_date 字段 + 复合唯一约束）
• 用途：指数算法参数调优验证、历史对照、补齐生产数据缺口
• 关联专题：P19 指数回测框架 Phase 1e

DWS_TASK_SIMULATION（任务引擎历史推演）

• 源码：tasks/utility/task_simulation_task.py
• 职责：基于指数日快照，逐天重放 task_generator + recall_detector 逻辑，还原任务的完整生命周期（生成 / 升级 / 转移 / 完成 / follow_up）
• 前置条件：需先运行 DWS_INDEX_BACKFILL 产出完整的历史指数快照作为推演输入
• 数据库连接：额外使用业务库独立连接（APP_DB_DSN 环境变量），推演结果写入业务库任务相关表
• 任务引擎双模式配合：task_engine.py 同期升级为双模式 — HTTP 实时模式（日常生产）+ 推演模式（回测，由 as_of_date 参数驱动），两种模式共用核心规则逻辑
• 用途：回测验证分级分配、升级、转移规则的参数配置与边界行为
• 关联专题：P19 指数回测框架 Phase 2

ETL 管理层

• Schema：meta
• 职责：调度元数据管理
• 内容：游标（水位）记录、任务运行记录、调度配置
• 关键组件：cursor_manager.py（水位管理）、run_tracker.py（运行记录）

窗口切分与补偿

大时间范围的数据抓取会按窗口切分执行，避免单次请求数据量过大：

配置项	默认值	说明
run.window_split.unit	day	切分单位：day / week / month / none
run.window_split.days	10	切分天数
run.window_split.compensation_hours	2	补偿小时数（处理跨窗口数据）

数据质量保障

• DWD_QUALITY_CHECK：DWD 层质量检查
• quality/integrity_service.py：完整性检查服务（余额一致性等）
• tasks/verification/：ETL 后置校验（ODS/DWD/DWS/指数校验器）