📚 课程目录第4章 Snowpipe持续加载

4. Snowpipe持续加载

Snowpipe 是 Snowflake 提供的持续数据摄取服务，它把“文件到达云存储”这一事件转化为“数据进入目标表”的自动化动作。与手动执行 COPY INTO 不同，Snowpipe 通过事件驱动或轮询方式感知新文件，并在秒级到分钟级内完成加载。理解它的关键在于：Snowpipe 本身不存储数据，它是一套由事件触发的 COPY 执行机制，真正的加载仍由虚拟仓库完成，只是这个仓库由 Snowflake 按需启动和计费。

要使用 Snowpipe，通常需要完成四件事：准备目标表与外部 Stage、创建 Pipe、配置事件通知（或轮询）、监控与容错。下面按实际落地顺序展开。

创建 Pipe 与自动摄取配置

Pipe 是 Snowpipe 的核心对象，它绑定一个 COPY 语句，并记录已加载文件的元数据。创建 Pipe 前，先要有指向云存储的外部 Stage。以 S3 为例：

CREATE OR REPLACE STAGE my_ext_stage
  URL='s3://my-bucket/incoming/'
  STORAGE_INTEGRATION = my_s3_int;

CREATE OR REPLACE PIPE sales_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO sales_raw
  FROM @my_ext_stage
  FILE_FORMAT = (TYPE = 'JSON' STRIP_OUTER_ARRAY = TRUE)
  ON_ERROR = 'CONTINUE';

AUTO_INGEST = TRUE 表示由云存储事件通知触发。创建后执行 SHOW PIPES，会看到 notification_channel 字段，它是一个 SQS 队列 ARN（S3）或 Event Grid/Notification 端点（Azure/GCP）。把这个 ARN 配置到 S3 桶的事件通知上，文件一落地就会向该队列发消息，Snowpipe 随即触发加载。若无法使用事件通知，可退化为 AUTO_INGEST = FALSE 配合 ALTER PIPE ... REFRESH 或调用 REST API insertFiles 主动投递文件列表。

通知集成（Notification Integration）是跨云账户授权的桥梁。以 S3 为例，需要先创建存储集成和通知集成，把 Snowflake 生成的 IAM 用户 ARN 与外部 ID 填入 AWS 信任策略，再把 SQS 队列策略授权给该 IAM 用户。常见错误是只配了存储集成却忘了队列策略，导致文件到达后 Pipe 无反应。

监控加载状态

Pipe 创建后，最直接的观察方式是查询 SYSTEM$PIPE_STATUS('sales_pipe')，返回 JSON 中包含 executionState、pendingFileCount、lastIngestedTimestamp 等字段。pendingFileCount 持续大于 0，说明有文件排队未被消费，可能是仓库太小、文件过多或通知积压。更细粒度的历史记录来自两个视图：

SELECT * FROM TABLE(INFORMATION_SCHEMA.PIPE_USAGE_HISTORY(
  DATE_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP()),
  PIPE_NAME => 'SALES_PIPE'));

SELECT FILE_NAME, STATUS, ROW_COUNT, ERROR_MESSAGE
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'SALES_RAW',
  START_TIME => DATEADD('hour', -1, CURRENT_TIMESTAMP())))
ORDER BY LAST_LOAD_TIME DESC;

PIPE_USAGE_HISTORY 给出文件数、字节数和消耗的 credits，是成本核算的主要依据；COPY_HISTORY 则逐文件显示 LOADED、LOAD_FAILED、PARTIALLY_LOADED 等状态，配合 ERROR_MESSAGE 定位格式错误、权限问题或 schema 不匹配。若怀疑某批文件未被加载，可用 ALTER PIPE sales_pipe REFRESH 触发一次元数据扫描，但要注意 REFRESH 只对 AUTO_INGEST=FALSE 或需要补漏时使用，频繁调用会带来额外开销。

处理重复与失败文件

Snowpipe 的元数据会记录每个已加载文件的路径和 ETag，因此同一文件被重复投递时不会重复加载，这是它天然的去重能力。但要注意：如果文件内容变了而路径没变，Snowpipe 仍会视为同一文件而跳过，此时需要改名或使用 FORCE = TRUE 的 COPY 手动补载。失败文件通常分两类：一类是格式错误，ON_ERROR='CONTINUE' 会跳过坏行并记录到 COPY_HISTORY，适合容忍少量脏数据；另一类是权限或 Stage 路径错误，会导致整个文件 LOAD_FAILED，需要修正后重新投递。对于 PARTIALLY_LOADED 的文件，建议先查 ERROR_MESSAGE，确认是行级问题还是文件级问题，再决定是否重跑。

成本与延迟权衡

Snowpipe 的计费方式是“按文件加载消耗的仓库积分”，而不是按仓库运行时长。这意味着小文件高频到达会显著推高成本，因为每个文件都会触发一次仓库启动和一次 COPY 执行。实践中常见的优化是：在云存储侧做小文件合并，或让上游按 100MB～250MB 量级批量落盘；同时把 PIPE 的仓库设为较小规格（如 XSMALL），因为 Snowpipe 的加载任务本身并不重。延迟方面，事件驱动通常能在文件到达后数十秒内完成加载；若使用轮询，则受 PIPE_EXECUTION 间隔影响，默认约 1 分钟，可通过调整参数缩短，但会增加扫描开销。权衡的核心是：延迟要求越高、文件越碎，成本越高；反之，批量越大、延迟容忍度越高，单位数据成本越低。

建议为每个 Pipe 单独设置一个监控查询，定期检查 pendingFileCount 和 COPY_HISTORY 中的 LOAD_FAILED 数量，把“文件积压”和“加载失败”作为两个独立告警项，而不是只看总文件数。

最后给出一个可直接参考的 Pipe 配置检查清单：确认外部 Stage 可访问、确认通知集成已授权到队列、确认 Pipe 的 AUTO_INGEST 与云侧事件通知一致、确认 ON_ERROR 策略符合数据质量要求、确认目标表 schema 与文件格式匹配、确认已建立 PIPE_USAGE_HISTORY 与 COPY_HISTORY 的例行查询。把这六项固定为上线前检查，可以避免绝大多数“文件到了但没进表”的问题。