📚 课程目录第4章构建成本数据管道

4. 构建成本数据管道

Snowflake 的账单数据天然是“原始、明细、按小时或按天滚动”的形态。以 ORGANIZATION_USAGE 共享或 SNOWFLAKE.ACCOUNT_USAGE 视图为例，WAREHOUSE_METERING_HISTORY 每行代表一个仓库在某个小时内的信用点消耗，METERING_DAILY_HISTORY 则按天汇总存储与计算用量。这些表字段多、粒度细、保留期有限（ACCOUNT_USAGE 通常 1 年，ORGANIZATION_USAGE 更长），直接在上面做 FinOps 分析既慢又贵。因此需要一条成本数据管道，把原始层逐步加工为可分析、可复用的聚合层。

分层模型设计

推荐采用四层结构，命名上保持清晰：

RAW 原始层：直接落地 ORGANIZATION_USAGE 或 ACCOUNT_USAGE 的原始表，保留全部字段与原始粒度，不做任何清洗。这一层是“事实来源”，用于审计与回溯。
STG 清洗层：对原始表做字段重命名、类型转换、空值处理、去重，并补充 cost_date、cost_month、service_category 等派生列。粒度与 RAW 一致，但字段更规范。
DIM 维度层：维护仓库、数据库、用户、标签（tag）等维度表，用于后续按业务归属聚合。
AGG 聚合层：按天、按仓库、按服务、按标签等维度预聚合，形成 FCT_COST_DAILY、FCT_COST_BY_TAG 等宽表，供 BI 与告警直接查询。

分层的关键收益是：RAW 只写一次，STG 做标准化，AGG 做预计算。下游查询不再扫描原始大表，成本与延迟都显著下降。

用 Stream + Task 实现增量加工

RAW 到 STG 的加工适合用 Stream + Task 组合。Stream 捕获源表的 CDC（变更数据捕获），Task 按调度周期消费 Stream 中的增量数据，避免全量重算。下面是一个可参考的示例配置：

-- 1. 在 RAW 层创建流，捕获新到达的用量记录
CREATE OR REPLACE STREAM RAW.STRM_METERING
  ON TABLE RAW.WAREHOUSE_METERING_HISTORY;

-- 2. 创建任务，每 15 分钟消费一次增量
CREATE OR REPLACE TASK STG.TSK_LOAD_METERING
  WAREHOUSE = FINOPS_ETL_WH
  SCHEDULE  = '15 MINUTE'
WHEN
  SYSTEM$STREAM_HAS_DATA('RAW.STRM_METERING')
AS
INSERT INTO STG.WAREHOUSE_METERING_CLEAN
SELECT
  START_TIME::DATE            AS cost_date,
  DATE_TRUNC('MONTH', START_TIME) AS cost_month,
  WAREHOUSE_NAME,
  CREDITS_USED,
  CREDITS_USED_COMPUTE,
  CREDITS_USED_CLOUD_SERVICES,
  CURRENT_TIMESTAMP()         AS loaded_at
FROM RAW.STRM_METERING
WHERE CREDITS_USED > 0;

-- 3. 启动任务（默认创建后为挂起状态）
ALTER TASK STG.TSK_LOAD_METERING RESUME;

这里有几个工程要点：WHEN SYSTEM$STREAM_HAS_DATA(...) 让任务在没有增量时直接跳过，不消耗仓库；SCHEDULE = '15 MINUTE' 是调度频率，不是执行时长；任务创建后默认 SUSPENDED，必须显式 RESUME。如果任务链较长，可以用 AFTER 子句构建 DAG，例如让聚合任务依赖清洗任务完成。

用动态表做声明式聚合

对于 AGG 层，动态表（Dynamic Table）比手写 Task 更省心。它用 SQL 定义目标结果，Snowflake 自动追踪上游变化并按目标滞后（target lag）刷新。示例：

CREATE OR REPLACE DYNAMIC TABLE AGG.FCT_COST_DAILY
  TARGET_LAG = '1 hour'
  WAREHOUSE  = FINOPS_ETL_WH
AS
SELECT
  cost_date,
  cost_month,
  warehouse_name,
  SUM(credits_used)                AS total_credits,
  SUM(credits_used_compute)        AS compute_credits,
  SUM(credits_used_cloud_services) AS cloud_credits
FROM STG.WAREHOUSE_METERING_CLEAN
GROUP BY 1, 2, 3;

TARGET_LAG = '1 hour' 表示聚合结果最多滞后上游 1 小时，Snowflake 会据此决定刷新频率。动态表适合“上游变化、下游按维度聚合”的场景；而 Stream + Task 更适合需要复杂控制逻辑、条件分支或跨源合并的场景。两者可以混用：STG 用 Task 做清洗，AGG 用动态表做聚合。

成本数据管道全景

下图展示从原始账单到聚合层的完整数据流，以及各层之间的依赖与刷新机制：

关键参数与判断依据

管道上线后，需要关注几个可观测指标来判断是否健康。第一是数据新鲜度：查询 AGG.FCT_COST_DAILY 的 MAX(cost_date) 是否接近当前日期，若滞后超过 TARGET_LAG 的两倍，说明刷新受阻。第二是任务执行历史：通过 INFORMATION_SCHEMA.TASK_HISTORY 查看 STATE 字段，SUCCEEDED 为正常，FAILED 需排查 SQL 错误或权限问题。第三是Stream 积压：SYSTEM$STREAM_HAS_DATA 长期为真但任务未消费，可能是任务被挂起或仓库不可用。

实践建议：ETL 仓库单独配置，不要与 BI 查询共用，避免任务刷新抢占查询资源。ETL 仓库可设置 AUTO_SUSPEND = 60、AUTO_RESUME = TRUE，并选用 XS 或 S 规格，因为增量加工的数据量通常不大。

注意：动态表的 TARGET_LAG 不是“精确刷新间隔”，而是“允许的最大滞后”。设置过小会导致刷新过于频繁、成本上升；设置过大则数据不够新鲜。建议从 '1 hour' 起步，根据业务对时效的要求逐步调整。

最后，聚合层的表结构应围绕 FinOps 的核心问题设计：谁在花、花在什么上、花了多少、趋势如何。例如 FCT_COST_DAILY 按 cost_date + warehouse_name 聚合，FCT_COST_BY_TAG 按 cost_month + tag_value 聚合。这样下游做成本分摊、预算对比、异常检测时，只需简单查询聚合表，无需再触碰原始账单。整条管道的目标，就是让“原始数据 → 可分析数据”的转化一次完成、持续运行、可观测、可追溯。