Snowflake Development

何时使用

• 在 Snowflake 上编写 SQL、设计数据管道（Dynamic Tables / Streams + Tasks / Snowpipe）、调用 Cortex AI 函数或 Agent、用 Snowpark Python 处理数据、做 dbt 物化、性能调优或安全加固时使用。
• 不该用：目标不是 Snowflake（如 MySQL、PostgreSQL、BigQuery、Spark）；只是写与厂商无关的通用 SQL；或缺少必要的库/Schema/权限/成功标准——此时应先停下来澄清。
• 输出不能替代针对具体环境的验证、测试与专家评审。

步骤

1. 判定任务类别：SQL、管道、Cortex、Snowpark、dbt、性能或安全，选择对应小节规则。
2. 写 SQL 时统一用 snake_case、CTE（WITH）替代嵌套子查询、CREATE OR REPLACE 做幂等 DDL、生产环境禁止 SELECT *（列存只扫描引用列）。避免双引号标识符（会产生大小写敏感名）。
3. 搭管道时按下表选型，默认优先 Dynamic Tables。
4. 用 Cortex 时只用新版 AI_* 函数，避开已废弃函数。
5. Snowpark/​dbt/性能/安全按各自约束落地，最后自查「常见错误」表。

指令

SQL 关键约束

• 存储过程（BEGIN...END）内，SQL 语句中引用变量/参数必须加冒号 : 前缀，否则报 "invalid identifier"。
  • 错误：SELECT name INTO result FROM users WHERE id = p_id;
  • 正确：SELECT name INTO :result FROM users WHERE id = :p_id;
• 半结构化（VARIANT/OBJECT/ARRAY）：访问嵌套字段并始终强制转型，如 src:customer.name::STRING、src:price::NUMBER(10,2)。JSON null 会存为字符串 "null"，加载时用 STRIP_NULL_VALUE = TRUE。展开数组用 LATERAL FLATTEN(input => src:items)。
• Upsert 用 MERGE INTO ... WHEN MATCHED THEN UPDATE ... WHEN NOT MATCHED THEN INSERT ...。

管道选型

Dynamic Tables 规则：TARGET_LAG 上紧下松逐级设置；增量 DT 不能依赖全量刷新 DT；用显式列名（SELECT * 遇 schema 变更会断）；基表须保持 change tracking 开启；两个 DT 之间不能夹视图。

Streams + Tasks：Task 创建后是 SUSPENDED，必须 ALTER TASK ... RESUME;；常配 WHEN SYSTEM$STREAM_HAS_DATA('stream_name') 仅在有数据时触发。

Cortex AI

• 用新版函数：AI_COMPLETE/AI_CLASSIFY/AI_FILTER/AI_EXTRACT/AI_SENTIMENT/AI_PARSE_DOCUMENT/AI_REDACT。
• 已废弃，禁用：COMPLETE、CLASSIFY_TEXT、EXTRACT_ANSWER、PARSE_DOCUMENT、SUMMARIZE、TRANSLATE、SENTIMENT、EMBED_TEXT_768。
• 分类用 AI_CLASSIFY 而非 AI_COMPLETE。
• TO_FILE 的 stage 路径与文件名是两个独立参数：TO_FILE('@db.schema.mystage', 'invoice.pdf')，不要写成 TO_FILE('@stage/file.pdf')。
• Cortex Agent：用 $spec$ 分隔符（不是 $$）；models 必须是对象不是数组；tool_resources 是独立顶层对象不嵌在 tools 里；编辑规格里不要放空/null（会清空原值）；工具描述是质量第一要素；生产 Agent 先克隆再改，禁止直改。

Snowpark Python

• 禁止硬编码凭据，从环境变量读取（os.environ[...]）。
• DataFrame 惰性执行，触发于 collect()/show()；大 DataFrame 不要 collect()，尽量服务端处理。
• 批量/ML 用向量化 UDF（比标量 UDF 快 10-100 倍）。

dbt on Snowflake

• 近实时/流式集市用动态表物化：{{ config(materialized='dynamic_table', snowflake_warehouse='transforming', target_lag='1 hour') }}。
• 大事实表用增量：{{ config(materialized='incremental', unique_key='event_id') }}。
• Snowflake 专有配置可叠加：{{ config(transient=true, copy_grants=true, query_tag='team_daily') }}。
• 用 {{ this }} 必须包在 {% if is_incremental() %} 守卫内。

性能 / 安全

• Cluster key 仅用于多 TB 表，建在 WHERE/JOIN/GROUP BY 列上；点查用 Search Optimization：ALTER TABLE t ADD SEARCH OPTIMIZATION ON EQUALITY(col);。
• 仓库从 X-Small 起步逐步放大，设 AUTO_SUSPEND = 60、AUTO_RESUME = TRUE，按工作负载分仓。
• 调用前先估 AI 成本：SELECT SUM(AI_COUNT_TOKENS('claude-4-sonnet', text)) FROM table;。
• 最小权限 RBAC，对象级授权用 database role；定期审计 SHOW GRANTS OF ROLE ACCOUNTADMIN;；用网络策略做 IP 白名单，PII 列用 masking policy、多租户隔离用 row access policy。

示例

声明式管道：用 Dynamic Table 做近实时清洗。

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

命令式管道：Stream + Task（注意建后须 RESUME）。

CREATE OR REPLACE STREAM raw_stream ON TABLE raw_events;

CREATE OR REPLACE TASK process_events
    WAREHOUSE = transform_wh
    SCHEDULE = 'USING CRON 0 */1 * * * America/Los_Angeles'
    WHEN SYSTEM$STREAM_HAS_DATA('raw_stream')
    AS INSERT INTO cleaned_events SELECT ... FROM raw_stream;

ALTER TASK process_events RESUME;  -- Task 默认 SUSPENDED，必须显式恢复

文本分类：用 AI_CLASSIFY 取标签。

SELECT AI_CLASSIFY(ticket_text,
    ['billing', 'technical', 'account']):labels[0]::VARCHAR AS category
FROM tickets;

注意事项

常见错误排查：

仅在任务明确落在上述范围内时使用本技能；缺输入/权限/安全边界/成功标准时先澄清。

互见

• sql-query-builder：通用 SQL 查询构造，与本条的 Snowflake 专有语法互补。
• rag-pipeline-builder：构建 RAG 管道时可结合 Cortex AI 函数与向量能力。
• csv-data-cleaner：入库前的本地 CSV 数据清洗。

本条采编自 sickn33/antigravity-awesome-skills（MIT）。