snowflake-data-engineering

Snowflake 数据工程

此技能涵盖 Snowflake 上的 SQL 约定、管道架构（Dynamic Tables、Streams、Tasks、Snowpipe）、性能调优以及成本和访问管理。

构建 Snowflake 管道的工作流

1. 接入原始数据 — 使用 Snowpipe（AUTO_INGEST = TRUE）从外部暂存区持续加载文件，或使用 Snowpipe Streaming 通过 SDK 低延迟地逐行摄取数据。
2. 选择转换方式 — 大多数管道优先使用声明式的 Dynamic Tables；只有在需要过程式逻辑或调用存储过程时，才改用 Streams + Tasks。
3. 建模半结构化数据 — 将原始 JSON/Avro/Parquet 以 VARIANT 类型接入，然后尽早将其展平为类型化的关系列。
4. 串联管道阶段 — 在 Dynamic Tables 之间逐层构建（或使用 Streams 为 Tasks 提供数据），让每个阶段的范围从原始数据逐步收窄到清洗后数据，再到聚合数据。
5. 调优性能 — 仅在查询模式确有需要时添加聚类键或 Search Optimization；为查询添加标签以便进行成本归因。
6. 设置访问控制 — 采用最小权限原则实施 RBAC，使用功能角色（loader、transformer、analyst），并为敏感数据设置掩码策略和行访问策略。
7. 监控成本与数据新鲜度 — 跟踪 WAREHOUSE_METERING_HISTORY 和 QUERY_HISTORY，设置 Resource Monitors，并验证 TARGET_LAG 是否符合实际的数据新鲜度要求。

SQL 和半结构化数据

• 对 JSON、Avro、Parquet 和 ORC 数据使用 VARIANT、OBJECT 和 ARRAY 类型。
• 使用冒号表示法访问嵌套字段，并显式转换类型：src:customer.name::STRING、src:price::NUMBER(10,2)、src:created_at::TIMESTAMP_NTZ。
• 使用 LATERAL FLATTEN 展平数组：

SELECT f.value:name::STRING AS name
FROM my_table, LATERAL FLATTEN(input => src:items) f;

• 只要半结构化数据中包含日期、以字符串存储的数字或数组，就应将其展平为关系列 — 将数据保留在 VARIANT 中会使 Snowflake 的自动子列化无法发挥作用。
• 出于同样的原因，避免在同一个 VARIANT 字段中混用不同类型。
• 请注意，JSON null 会存储为字符串 "null"，与 SQL NULL 不同。若要将它们视为相同值，请在加载时使用 STRIP_NULL_VALUES => TRUE。

SQL 编码标准

• 所有标识符均使用 snake_case；避免使用带引号的标识符。
• 为提高可读性，优先使用 CTE，而不是多层嵌套子查询。
• 使用 CREATE OR REPLACE 编写幂等 DDL。
• 批量加载使用 COPY INTO，绝不逐行执行 INSERT。
• 使用 MERGE 执行插入或更新：

MERGE INTO target t USING source s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.name = s.name
WHEN NOT MATCHED THEN INSERT (id, name) VALUES (s.id, s.name);

• 在存储过程中，在 SQL 语句中引用会话变量或局部变量时，为其添加前缀 :：

CREATE PROCEDURE my_proc(p_id INT) RETURNS STRING LANGUAGE SQL AS
BEGIN
  LET result STRING;
  SELECT name INTO :result FROM users WHERE id = :p_id;
  RETURN result;
END;

性能优化

• 仅为超大型表（多 TB）添加聚类键，并选择经常用于 WHERE/JOIN/GROUP BY 的列：

ALTER TABLE large_events CLUSTER BY (event_date, region);

• 对高基数列上的点查找或子串/正则表达式搜索，使用 Search Optimization Service：

ALTER TABLE logs ADD SEARCH OPTIMIZATION ON EQUALITY(sender_ip), SUBSTRING(error_message);

• 使用 Materialized Views 预先计算开销较大的单表聚合。
• 使用 RESULT_SCAN(LAST_QUERY_ID()) 重用先前结果，避免重复运行相同查询。
• 为查询添加标签以便进行成本归因：ALTER SESSION SET QUERY_TAG = 'etl_daily_load';
• 绝不要对宽表运行 SELECT * — 这会抵消列式裁剪的优势。

数据管道

选择合适的原语：

方式	适用场景
Dynamic Tables	声明式：定义查询，由 Snowflake 管理刷新。适用于大多数管道的默认选择。
Streams + Tasks	命令式 CDC + 调度；需要过程式逻辑或调用存储过程时使用。
Snowpipe	从 S3/GCS/Azure 持续加载文件。
Snowpipe Streaming	通过 SDK（Java、Python）低延迟地逐行摄取数据。

Dynamic Tables

CREATE OR REPLACE DYNAMIC TABLE cleaned_events
  TARGET_LAG = '5 minutes'
  WAREHOUSE = transform_wh
  AS
  SELECT event_id, event_type, user_id, event_data:page::STRING AS page, event_timestamp
  FROM raw_events
  WHERE event_type IS NOT NULL;

-- Chain for multi-step pipelines
CREATE OR REPLACE DYNAMIC TABLE user_sessions
  TARGET_LAG = '10 minutes'
  WAREHOUSE = transform_wh
  AS
  SELECT user_id, MIN(event_timestamp) AS session_start, MAX(event_timestamp) AS session_end,
         COUNT(*) AS event_count
  FROM cleaned_events GROUP BY user_id;

TARGET_LAG 设置数据新鲜度目标。REFRESH_MODE 可以是 AUTO、FULL 或 INCREMENTAL。使用 ALTER DYNAMIC TABLE ... SET TARGET_LAG / REFRESH / SUSPEND / RESUME 管理生命周期。

流（CDC）

CREATE OR REPLACE STREAM raw_events_stream ON TABLE raw_events;

流会添加 METADATA$ACTION、METADATA$ISUPDATE 和 METADATA$ROW_ID 列。对于仅插入数据的源，设置 APPEND_ONLY = TRUE 以降低开销。

任务（计划执行/触发执行）

CREATE OR REPLACE TASK process_events
  WAREHOUSE = transform_wh
  SCHEDULE = 'USING CRON 0 */1 * * * America/Los_Angeles'
  WHEN SYSTEM$STREAM_HAS_DATA('raw_events_stream')
  AS
  INSERT INTO cleaned_events
  SELECT event_id, event_type, user_id, event_timestamp
  FROM raw_events_stream WHERE event_type IS NOT NULL;

使用 CREATE TASK child_task ... AFTER parent_task ... 构建任务 DAG。任务默认以 SUSPENDED 状态创建——记得执行 ALTER TASK ... RESUME，否则任务不会运行。

Snowpipe

CREATE OR REPLACE PIPE my_pipe AUTO_INGEST = TRUE AS
  COPY INTO raw_events FROM @my_external_stage FILE_FORMAT = (TYPE = 'JSON');

一种常见的端到端模式是使用 Snowpipe 加载原始数据，再将数据传入一系列动态表。

时间旅行和数据保护

• 使用时间旅行查询历史数据（默认 1 天，Enterprise+ 最高可达 90 天）：

SELECT * FROM my_table AT(TIMESTAMP => '2026-01-15 10:00:00'::TIMESTAMP);
SELECT * FROM my_table BEFORE(STATEMENT => '<query_id>');

• 使用 UNDROP TABLE/SCHEMA/DATABASE 恢复已删除的对象。
• 使用零拷贝克隆创建开发/测试环境或备份，无需重复存储：CREATE TABLE clone CLONE source;、CREATE SCHEMA dev CLONE prod;。

Snowflake Postgres

• Snowflake 提供托管的 PostgreSQL（v16/17/18），并完全兼容线协议：CREATE POSTGRES INSTANCE my_instance COMPUTE_FAMILY='STANDARD_S' STORAGE_SIZE_GB=50;
• 使用 pg_lake 扩展连接 OLTP 与分析工作负载。该扩展会公开 Iceberg 表，可供 Postgres 和 Snowflake 读取。
• 使用 FORK 进行时间点恢复，并为生产实例设置 HIGH_AVAILABILITY = TRUE。

仓库和成本管理

• 根据查询复杂度而非原始数据量选择仓库大小——从 X-Small 开始，仅在需要时扩展。
• 设置 AUTO_SUSPEND = 60 和 AUTO_RESUME = TRUE；按工作负载使用独立仓库，避免繁重任务挤占交互式查询的资源。
• 使用多集群仓库扩展并发能力，而非提升单个查询的速度。
• 使用临时表存储暂存数据，以避免 Fail-safe 存储成本。
• 通过 SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY 和 WAREHOUSE_METERING_HISTORY 监控支出，并设置资源监视器以限制积分消耗。

访问控制

• 使用数据库角色授予对象权限，实施最小权限 RBAC。
• 使用掩码策略保护 PII，并使用行访问策略实现多租户隔离。
• 围绕管道阶段设置功能角色：加载器（写入原始数据）、转换器（读取原始数据、写入分析数据）、分析师（仅读取分析数据）。

数据共享和 Iceberg

• 使用 CREATE SHARE 实现跨账户零拷贝数据共享，或使用 Snowflake Marketplace 进行外部交换。
• 使用 CREATE ICEBERG TABLE ... CATALOG='SNOWFLAKE' EXTERNAL_VOLUME='vol' BASE_LOCATION='path/'; 创建 Iceberg 表，以便与 Spark、Flink 和 Trino 互操作。

反模式

• 对于动态表能够以声明式方式表达的简单转换，不要使用 Streams + Tasks。
• 不要将 TARGET_LAG 设得比实际新鲜度要求更短——它会直接影响计算成本。
• 创建任务后不要忘记执行 RESUME；任务创建时处于 SUSPENDED 状态。
• 不要对宽表运行 SELECT *；在多 TB 表上添加集群键之前，也不要跳过聚类分析。
• 不要在可复用的管道代码中硬编码数据库/架构名称。