3. 构建可观测性数据管道

Snowflake 内部已经积累了丰富的可观测性原始数据：ACCOUNT_USAGE 视图中的查询历史、登录记录、仓库负载，INFORMATION_SCHEMA 中的对象元数据，以及 QUERY_HISTORY、WAREHOUSE_EVENTS_HISTORY 等表函数返回的实时事件。但这些数据停留在 Snowflake 内部时，只能被少数具备 SQL 能力的人按需查询，无法驱动告警、仪表盘和跨系统关联分析。本节要解决的核心问题是：如何把这些原生数据源，以近实时、低延迟、可维护的方式，持续搬运到外部可观测性平台（如 Grafana、Datadog、Elastic、自建时序库或 Kafka 下游）。

三种采集模式及其适用边界

从 Snowflake 到外部平台，本质上是一条“采集—转换—加载”的链路。Snowflake 提供了三种主流的增量采集机制，它们的语义和成本差异很大，选错会直接导致延迟或费用失控。

Streams + Tasks 是最经典、控制力最强的模式。Stream 在表上记录变更数据捕获（CDC）偏移量，Task 按调度周期读取 Stream 中的增量行，转换后写入暂存表或直接推送。它的优势是精确的增量语义、可编排依赖、可用 Snowflake 原生 SQL 完成全部转换；代价是需要自己管理调度频率、Stream 过期（默认 14 天未消费会失效）和 Task 树依赖。

Dynamic Tables 是声明式方案。你只定义目标结果集的 SQL 和期望的 TARGET_LAG，Snowflake 自动决定刷新时机和增量计算方式。它极大简化了“物化视图式”的转换层，适合把多张 ACCOUNT_USAGE 表 join、聚合后形成宽表，再交给外部采集器拉取。但它对下游推送不负责，仍需一个外部拉取或导出环节。

外部拉取（Pull） 是最简单的模式：外部采集器（如 Telegraf、Vector、自研 Python 服务）按固定间隔执行 SELECT ... WHERE ingest_time > :last_watermark，把结果推送到目标平台。它不依赖 Stream/Task，但需要自己维护水位线，且高频轮询会消耗仓库资源。

近实时同步链路的关键设计

下面这张图展示了一条典型的混合链路：底层用 Streams 捕获 ACCOUNT_USAGE 与事件表的增量，中间用 Task 做清洗和聚合，再用 Dynamic Table 形成对外可读的宽表，最后由外部采集器通过 Snowflake SQL API 或连接器拉取并写入可观测性平台。图中同时标出了水位线、TARGET_LAG 和推送出口三个关键控制点。

用 Streams + Tasks 实现增量清洗

假设我们要把 ACCOUNT_USAGE.QUERY_HISTORY 中执行时间超过 5 秒的查询，按仓库和用户聚合后输出。第一步是在源表上建 Stream。注意 ACCOUNT_USAGE 视图有约 45 分钟到 3 小时的延迟，因此“近实时”在这里的物理下限由源延迟决定，Stream 只能保证不遗漏、不重复消费。

CREATE OR REPLACE STREAM OBS_DB.CDC.STRM_QUERY_HIST
  ON VIEW SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
  SHOW_INITIAL_ROWS = FALSE;

CREATE OR REPLACE TASK OBS_DB.CDC.TASK_CLEAN_QUERY
  WAREHOUSE = OBS_WH
  SCHEDULE = '1 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('OBS_DB.CDC.STRM_QUERY_HIST')
AS
INSERT INTO OBS_DB.STG.QUERY_SLOW_CLEAN
SELECT
  QUERY_ID,
  WAREHOUSE_NAME,
  USER_NAME,
  TOTAL_ELAPSED_TIME / 1000.0 AS elapsed_sec,
  BYTES_SCANNED,
  START_TIME,
  CURRENT_TIMESTAMP() AS ingest_ts
FROM OBS_DB.CDC.STRM_QUERY_HIST
WHERE TOTAL_ELAPSED_TIME > 5000
  AND METADATA$ACTION = 'INSERT';

这里有两个容易踩坑的细节。其一，WHEN SYSTEM$STREAM_HAS_DATA 让 Task 在没有增量时直接跳过，避免空跑消耗仓库；但该函数只判断“是否有数据”，不判断数据是否满足 WHERE 条件，因此仍可能产生空写入。其二，Stream 的消费是“读取即推进偏移”，如果 Task 中的 INSERT 失败，偏移不会推进，下次会重读，这保证了至少一次语义，但也意味着下游必须做幂等去重，通常用 QUERY_ID 作为主键。

用 Dynamic Tables 构建对外宽表

清洗后的暂存表还需要与仓库维度、用户维度关联，形成外部平台可直接消费的宽表。这一步用 Dynamic Table 最省心：

CREATE OR REPLACE DYNAMIC TABLE OBS_DB.MART.DT_OBS_WIDE
  TARGET_LAG = '1 minute'
  WAREHOUSE = OBS_WH
AS
SELECT
  q.QUERY_ID,
  q.WAREHOUSE_NAME,
  q.USER_NAME,
  q.elapsed_sec,
  q.BYTES_SCANNED,
  w.WAREHOUSE_SIZE,
  w.CREDIT_RATE_PER_HOUR,
  q.elapsed_sec / 3600.0 * w.CREDIT_RATE_PER_HOUR AS est_credits,
  q.START_TIME,
  q.ingest_ts
FROM OBS_DB.STG.QUERY_SLOW_CLEAN q
LEFT JOIN OBS_DB.DIM.WAREHOUSE_META w
  ON q.WAREHOUSE_NAME = w.WAREHOUSE_NAME;

TARGET_LAG = '1 minute' 表示 Snowflake 会尽力让该表的数据新鲜度不超过 1 分钟。实际刷新频率由 Snowflake 根据数据变化量和仓库负载自动决定，可能快于也可能略慢于该值。宽表中额外计算了 est_credits，把查询耗时折算成估算的 credit 消耗，这样外部平台可以直接对“慢查询造成的成本”做告警，而不只是对耗时告警。

加载到外部平台：水位线与幂等

外部采集器读取 DT_OBS_WIDE 时，必须维护水位线，否则每次全量拉取会迅速放大成本。推荐用 ingest_ts 作为水位线字段，并在目标平台侧以 QUERY_ID 做 upsert。一个可用的采集循环如下：

-- 采集器每次执行
SELECT * FROM OBS_DB.MART.DT_OBS_WIDE
WHERE ingest_ts > :last_watermark
ORDER BY ingest_ts
LIMIT 10000;

-- 成功后更新水位线
-- last_watermark = max(ingest_ts) from this batch

需要注意，Dynamic Table 的刷新可能对同一行产生多次更新（例如维度表变化导致 join 结果变化），因此水位线只保证“不遗漏”，不保证“不重复”。目标平台侧必须用 QUERY_ID 做幂等写入。如果外部平台不支持 upsert，可以改为只拉取 ingest_ts 单调递增的追加型指标，把维度变化通过单独的维表同步通道处理。

端到端延迟与成本权衡

整条链路的端到端延迟由四段构成：源视图延迟（ACCOUNT_USAGE 约 45 分钟至 3 小时，事件表通常更短）、Stream 捕获延迟（近实时）、Task 调度周期（示例为 1 分钟）、Dynamic Table 刷新（TARGET_LAG 1 分钟）、外部采集间隔（示例为 30 秒）。如果业务要求 5 分钟内的告警，就必须避开 ACCOUNT_USAGE 视图，改用 INFORMATION_SCHEMA 的实时表函数或 QUERY_HISTORY 表函数，它们的数据延迟通常在秒级到分钟级。

成本方面，Task 和 Dynamic Table 的刷新都会消耗仓库 credit。控制手段包括：用 SYSTEM$STREAM_HAS_DATA 短路空跑、把 TARGET_LAG 从 1 分钟放宽到 5 分钟、为可观测性负载单独建 XS 仓库并设置 STATEMENT_TIMEOUT_IN_SECONDS 和 AUTO_SUSPEND。一个常见的反模式是让 Dynamic Table 直接 join 大表且 TARGET_LAG 设为 1 分钟，导致仓库几乎常驻运行，成本远超收益。

最后，可观测性管道本身也需要被观测。建议在外部平台中为这条链路建立独立的健康指标：Stream 积压行数、Task 最近成功时间、Dynamic Table 刷新延迟、采集器水位线滞后秒数、目标平台写入失败率。当这些指标异常时，先判断是源延迟、调度失败还是下游写入问题，再决定是调整 TARGET_LAG、扩容仓库还是修复采集器，而不是盲目提高刷新频率。