📚 课程目录第3章 Snowflake数据加载与准备

3. Snowflake数据加载与准备

在 Snowflake 上做机器学习，模型效果的上限往往在数据进入特征表之前就已经被决定了。本章聚焦“数据如何进入 Snowflake、如何被清洗成可训练的数据集”这条链路。核心对象只有四个：外部阶段（External Stage）、COPY INTO、Snowpipe 和半结构化数据（VARIANT）。理解它们之间的分工，比记住语法更重要。

一、外部阶段：把云存储挂载成 Snowflake 的“数据入口”

外部阶段本质是一个指向云对象存储（如 Amazon S3、Azure Blob、GCS）的命名引用，它保存了存储位置、存储集成（Storage Integration）和文件格式信息。创建后，Snowflake 就能像访问本地目录一样列出和读取文件。一个典型的 S3 外部阶段定义如下：

CREATE OR REPLACE STAGE ml_raw_stage
  URL = 's3://ml-datalake/raw/events/'
  STORAGE_INTEGRATION = s3_ml_int
  FILE_FORMAT = (TYPE = JSON);

LIST @ml_raw_stage;   -- 查看阶段内文件

这里的关键是 STORAGE_INTEGRATION：它通过 IAM Role 让 Snowflake 以受控身份访问桶，避免在阶段里硬编码密钥。对机器学习场景，建议按“数据源 + 分区”划分阶段，例如 ml_raw_stage/events/dt=2024-06-01/，这样后续 COPY 可以按前缀增量加载，而不是每次全量扫描。

二、COPY INTO：批量加载与清洗的第一道关口

COPY INTO 是把阶段文件写入目标表的主力命令。它支持列映射、转换表达式、错误容忍和去重。下面示例把 JSON 事件加载到一张带 VARIANT 列的原始表，并在加载时做字段提取：

CREATE OR REPLACE TABLE raw_events (
  event_id   STRING,
  user_id    STRING,
  event_time TIMESTAMP_NTZ,
  payload    VARIANT,
  src_file   STRING
);

COPY INTO raw_events (event_id, user_id, event_time, payload, src_file)
FROM (
  SELECT
    $1:event_id::STRING,
    $1:user_id::STRING,
    TO_TIMESTAMP_NTZ($1:ts::NUMBER, 3),
    $1,
    METADATA$FILENAME
  FROM @ml_raw_stage
)
FILE_FORMAT = (TYPE = JSON)
ON_ERROR = 'CONTINUE'
PATTERN = '.*events.*[.]json';

几个实践要点：ON_ERROR='CONTINUE' 让坏行跳过并记录到 COPY_HISTORY，便于事后排查；METADATA$FILENAME 保留来源文件，是数据血缘和幂等重跑的关键；PATTERN 用正则限定文件，避免误加载。加载完成后，用 SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(...)) 查看 ROW_COUNT、ERROR_COUNT 和 STATUS，判断本次加载是否干净。

COPY INTO 具有幂等性：默认会记录已加载文件，重复执行同一命令不会重复导入同一文件（除非使用 FORCE = TRUE）。这对定时任务非常重要。

三、Snowpipe：让数据“到达即加载”

当数据以分钟级持续到达时，手动跑 COPY 就不合适了。Snowpipe 通过事件通知（S3 Event → SQS → Snowflake）或 REST API 触发，自动执行预定义的 COPY 语句。它使用独立的计算资源，按实际消耗计费，不占用虚拟仓库。

CREATE OR REPLACE PIPE ml_events_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO raw_events (event_id, user_id, event_time, payload, src_file)
  FROM (
    SELECT $1:event_id::STRING, $1:user_id::STRING,
           TO_TIMESTAMP_NTZ($1:ts::NUMBER, 3), $1, METADATA$FILENAME
    FROM @ml_raw_stage
  )
  FILE_FORMAT = (TYPE = JSON);

SHOW PIPES LIKE 'ml_events_pipe';
-- 查看 notification_channel，配置到 S3 事件通知

创建后需执行 SHOW PIPES 获取 notification_channel（一个 SQS ARN），在 S3 桶上配置事件通知指向它。之后新文件落地即触发加载。用 SELECT SYSTEM$PIPE_STATUS('ml_events_pipe') 检查管道状态，pendingFileCount 持续增长通常意味着通知未正确配置或权限不足。

四、半结构化数据处理：从 VARIANT 到特征表

JSON 进入 VARIANT 列后，需要用 : 路径访问和 :: 转型。对于嵌套数组，用 LATERAL FLATTEN 展开成行。下面把事件中的商品列表展开为“用户-商品”交互记录，这是推荐模型常见的训练样本：

CREATE OR REPLACE TABLE user_item_interactions AS
SELECT
  e.user_id,
  f.value:item_id::STRING   AS item_id,
  f.value:qty::NUMBER       AS qty,
  e.event_time
FROM raw_events e,
LATERAL FLATTEN(input => e.payload:items) f
WHERE e.event_time >= '2024-06-01';

展开后还要处理缺失值、类型不一致和重复。例如用 COALESCE(qty, 1) 补默认值，用 ROW_NUMBER() OVER (PARTITION BY user_id, item_id ORDER BY event_time DESC) 去重保留最新一条。这些清洗动作应固化成视图或动态表，保证每次训练拿到的样本口径一致。

五、为机器学习准备高质量数据集

加载只是起点。真正决定模型质量的是特征表的口径。建议在原始表之上建立分层：raw_events（原样落地）→ stg_events（类型规范、去重）→ user_item_interactions（训练样本）。每层用视图或动态表定义，避免手工改数。训练前用一组检查确认数据可用：

检查项	示例查询	判断标准
主键唯一	`COUNT(*) vs COUNT(DISTINCT event_id)`	两者相等
时间范围	`MIN/MAX(event_time)`	覆盖目标训练窗口
缺失率	`SUM(CASE WHEN item_id IS NULL THEN 1 ELSE 0 END)/COUNT(*)`	低于约定阈值（如 1%）
类别分布	`GROUP BY item_id ORDER BY COUNT(*) DESC LIMIT 20`	无异常长尾或单一主导

半结构化数据展开后行数可能膨胀数十倍，务必在 FLATTEN 之后再做聚合或采样，否则计算成本会显著上升。对超大 JSON，可先用 OBJECT_KEYS 或 ARRAY_SIZE 探查结构，再决定展开策略。

把这条链路串起来：外部阶段负责“看得见文件”，COPY INTO 负责“批量、可控地写入”，Snowpipe 负责“持续、自动地写入”，VARIANT + FLATTEN 负责“把嵌套结构变成表格行”。四者配合，才能稳定产出可供训练的高质量数据集。