snowflake-platform

Snowflake Platform 技能

使用 snow CLI、Cortex AI 函数、Native Apps 和 Snowpark，在 Snowflake 的 AI Data Cloud 上构建和部署应用。

快速入门

安装 Snowflake CLI

pip install snowflake-cli
snow --version  # Should show 3.14.0+

配置连接

# Interactive setup
snow connection add

# Or create ~/.snowflake/config.toml manually

[connections.default]
account = "orgname-accountname"
user = "USERNAME"
authenticator = "SNOWFLAKE_JWT"
private_key_path = "~/.snowflake/rsa_key.p8"

测试连接

snow connection test -c default
snow sql -q "SELECT CURRENT_USER(), CURRENT_ACCOUNT()"

何时使用此技能

适用场景：

• 在 Snowflake 平台上构建应用
• 在 SQL 查询中使用 Cortex AI 函数
• 为 Marketplace 开发 Native Apps
• 设置 JWT 密钥对身份验证
• 使用 Snowpark Python

不适用场景：

• 构建 Streamlit 应用（请使用 streamlit-snowflake 技能）
• 需要数据工程／ETL 模式
• 使用 BI 工具（Tableau、Looker）

Cortex AI 函数

Snowflake Cortex 可直接在 SQL 中提供大模型功能。函数位于 SNOWFLAKE.CORTEX 架构中。

核心函数

函数	用途	GA 状态
COMPLETE / AI_COMPLETE	根据提示词生成文本	2025 年 11 月 GA
SUMMARIZE / AI_SUMMARIZE	总结文本	GA
TRANSLATE / AI_TRANSLATE	在不同语言之间翻译	2025 年 9 月 GA
SENTIMENT / AI_SENTIMENT	情感分析	2025 年 7 月 GA
AI_FILTER	自然语言筛选	2025 年 11 月 GA
AI_CLASSIFY	对文本／图像分类	2025 年 11 月 GA
AI_AGG	汇总各行的洞察	2025 年 11 月 GA

COMPLETE 函数

-- Simple prompt
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'llama3.1-70b',
    'Explain quantum computing in one sentence'
) AS response;

-- With conversation history
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'llama3.1-70b',
    [
        {'role': 'system', 'content': 'You are a helpful assistant'},
        {'role': 'user', 'content': 'What is Snowflake?'}
    ]
) AS response;

-- With options
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'mistral-large2',
    'Summarize this document',
    {'temperature': 0.3, 'max_tokens': 500}
) AS response;

可用模型：

• llama3.1-70b、llama3.1-8b、llama3.2-3b
• mistral-large2、mistral-7b
• snowflake-arctic
• gemma-7b
• claude-3-5-sonnet（200K 上下文）

模型上下文窗口（2025 年更新）：

模型	上下文窗口	适用场景
Claude 3.5 Sonnet	200,000 个 token	大型文档、长对话
Llama3.1-70b	128,000 个 token	复杂推理、中型文档
Llama3.1-8b	8,000 个 token	简单任务、短文本
Llama3.2-3b	8,000 个 token	快速推理、少量文本
Mistral-large2	不定	查看当前文档
Snowflake Arctic	不定	查看当前文档

Token 换算：约 4 个字符 = 1 个 token。32,000 个字符的文档 ≈ 8,000 个 token。

错误：Input exceeds context window limit → 使用较小的模型或将输入拆分成多个片段。

SUMMARIZE 函数

-- Single text
SELECT SNOWFLAKE.CORTEX.SUMMARIZE(article_text) AS summary
FROM articles
LIMIT 10;

-- Aggregate across rows (no context window limit)
SELECT AI_SUMMARIZE_AGG(review_text) AS all_reviews_summary
FROM product_reviews
WHERE product_id = 123;

TRANSLATE 函数

-- Translate to English (auto-detect source)
SELECT SNOWFLAKE.CORTEX.TRANSLATE(
    review_text,
    '',      -- Empty = auto-detect source language
    'en'     -- Target language
) AS translated
FROM international_reviews;

-- Explicit source language
SELECT AI_TRANSLATE(
    description,
    'es',    -- Source: Spanish
    'en'     -- Target: English
) AS translated
FROM spanish_products;

AI_FILTER（自然语言筛选）

性能：截至 2025 年 9 月，AI_FILTER 已包含自动优化功能，对于适用的查询，速度可提升 2–10 倍，token 用量最多可减少 60%。

-- Filter with plain English
SELECT * FROM customer_feedback
WHERE AI_FILTER(
    feedback_text,
    'mentions shipping problems or delivery delays'
);

-- Combine with SQL predicates for maximum optimization
-- Query planner applies standard filters FIRST, then AI on smaller dataset
SELECT * FROM support_tickets
WHERE created_date > '2025-01-01'  -- Standard filter applied first
  AND AI_FILTER(description, 'customer is angry or frustrated');

最佳实践：始终将 AI_FILTER 与传统 SQL 谓词（日期范围、类别等）结合使用，在 AI 处理前缩小数据集范围。这样可以最大限度地发挥自动优化的效果。

限流：在使用高峰期，AI 函数请求可能会受到限流，并返回可重试错误。生产应用应实现指数退避（参见已知问题 #10）。

AI_CLASSIFY

-- Categorize support tickets
SELECT
    ticket_id,
    AI_CLASSIFY(
        description,
        ['billing', 'technical', 'shipping', 'other']
    ) AS category
FROM support_tickets;

计费

Cortex AI 函数按 token 计费：

• 约 4 个字符 = 1 个 token
• 输入和输出 token 均计费
• 费率因模型而异（较大的模型成本更高）

大规模场景下的成本管理（来自社区）：

真实生产案例研究显示，单个 AI_COMPLETE 查询处理 11.8 亿条记录，耗费的额度接近 5,000 美元。需要关注的成本因素：

1. 跨区域推理：在你所在区域不可用的模型会产生额外的数据传输成本
2. 仓库空闲时间：未使用的计算资源仍会计费，但过于激进的自动挂起会增加恢复开销
3. 大型表连接：包含 AI 函数的复杂查询会成倍增加成本

-- This seemingly simple query can be expensive at scale
SELECT
    product_id,
    AI_COMPLETE('mistral-large2', 'Summarize: ' || review_text) as summary
FROM product_reviews  -- 1 billion rows
WHERE created_date > '2024-01-01';

-- Cost = (input tokens + output tokens) × row count × model rate
-- At scale, this adds up fast

最佳实践：

• 在应用 AI 函数之前筛选数据集
• 合理配置仓库大小（不要过度配置）
• 使用 QUERY_HISTORY 视图监控额度消耗
• 考虑批量处理，而不是逐行执行 AI 操作

来源：Snowflake Cortex AI 的隐性成本（包含计费证据的社区博客）

身份验证

JWT 密钥对身份验证

重要：Snowflake 使用两种账户标识符格式：

格式	示例	用途
组织-账户	irjoewf-wq46213	REST API URL、连接配置
账户定位符	NZ90655	JWT 声明（iss、sub）

两者不可互换！

查找账户定位符

SELECT CURRENT_ACCOUNT();  -- Returns: NZ90655

生成 RSA 密钥对

# Generate private key (PKCS#8 format required)
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out ~/.snowflake/rsa_key.p8 -nocrypt

# Generate public key
openssl rsa -in ~/.snowflake/rsa_key.p8 -pubout -out ~/.snowflake/rsa_key.pub

# Get fingerprint for JWT claims
openssl rsa -in ~/.snowflake/rsa_key.p8 -pubout -outform DER | \
  openssl dgst -sha256 -binary | openssl enc -base64

向用户注册公钥

-- In Snowflake worksheet (requires ACCOUNTADMIN or SECURITYADMIN)
ALTER USER my_user SET RSA_PUBLIC_KEY='MIIBIjANBgkq...';

JWT 声明格式

iss: ACCOUNT_LOCATOR.USERNAME.SHA256:fingerprint
sub: ACCOUNT_LOCATOR.USERNAME

示例：

iss: NZ90655.JEZWEB.SHA256:jpZO6LvU2SpKd8tE61OGfas5ZXpfHloiJd7XHLPDEEA=
sub: NZ90655.JEZWEB

SPCS 容器身份验证（v4.2.0+）

2026 年 1 月新增：在 Snowpark Container Services 中运行时，连接器会自动检测并使用 SPCS 服务标识符令牌。

# No special configuration needed inside SPCS containers
import snowflake.connector

# Auto-detects SPCS_TOKEN environment variable
conn = snowflake.connector.connect()

这样一来，容器化 Snowpark 服务无需显式凭据即可顺畅完成身份验证。

来源：版本 v4.2.0

Snow CLI 命令

项目管理

# Initialize project
snow init

# Execute SQL
snow sql -q "SELECT 1"
snow sql -f query.sql

# View logs
snow logs

Native App 命令

# Development
snow app run              # Deploy and run locally
snow app deploy           # Upload to stage only
snow app teardown         # Remove app

# Versioning
snow app version create V1_0
snow app version list
snow app version drop V1_0

# Publishing
snow app publish --version V1_0 --patch 0

# Release Channels
snow app release-channel list
snow app release-channel add-version --channel ALPHA --version V1_0
snow app release-directive set default --version V1_0 --patch 0 --channel DEFAULT

Streamlit 命令

snow streamlit deploy --replace
snow streamlit deploy --replace --open

暂存区命令

snow stage list
snow stage copy @my_stage/file.txt ./local/

Native App 开发

项目结构

my_native_app/
├── snowflake.yml           # Project config
├── manifest.yml            # App manifest
├── setup_script.sql        # Installation script
├── app/
│   └── streamlit/
│       ├── environment.yml
│       └── streamlit_app.py
└── scripts/
    └── setup.sql

snowflake.yml

definition_version: 2

native_app:
  name: my_app
  package:
    name: my_app_pkg
    distribution: external    # For marketplace
  application:
    name: my_app
  source_stage: stage/dev
  artifacts:
    - src: manifest.yml
      dest: manifest.yml
    - src: setup_script.sql
      dest: setup_script.sql
    - src: app/streamlit/environment.yml
      dest: streamlit/environment.yml
    - src: app/streamlit/streamlit_app.py
      dest: streamlit/streamlit_app.py
  enable_release_channels: true  # For ALPHA/BETA channels

manifest.yml

manifest_version: 1

artifacts:
  setup_script: setup_script.sql
  default_streamlit: streamlit/streamlit_app.py

# Note: Do NOT include privileges section - Native Apps can't declare privileges

外部访问集成

调用外部 API 的 Native App 需要进行以下设置：

-- 1. Create network rule (in a real database, NOT app package)
CREATE DATABASE IF NOT EXISTS MY_APP_UTILS;

CREATE OR REPLACE NETWORK RULE MY_APP_UTILS.PUBLIC.api_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('api.example.com:443');

-- 2. Create integration
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION my_app_integration
  ALLOWED_NETWORK_RULES = (MY_APP_UTILS.PUBLIC.api_rule)
  ENABLED = TRUE;

-- 3. Grant to app
GRANT USAGE ON INTEGRATION my_app_integration
  TO APPLICATION MY_APP;

-- 4. CRITICAL: Attach to Streamlit (must repeat after EVERY deploy!)
ALTER STREAMLIT MY_APP.config_schema.my_streamlit
  SET EXTERNAL_ACCESS_INTEGRATIONS = (my_app_integration);

警告：每次运行 snow app run 时都会重置步骤 4。每次部署后都必须重新执行！

共享数据模式

当 Native App 需要使用外部数据库中的数据时：

-- 1. Create shared_data schema in app package
CREATE SCHEMA IF NOT EXISTS MY_APP_PKG.SHARED_DATA;

-- 2. Create views referencing external database
CREATE OR REPLACE VIEW MY_APP_PKG.SHARED_DATA.MY_VIEW AS
SELECT * FROM EXTERNAL_DB.SCHEMA.TABLE;

-- 3. Grant REFERENCE_USAGE (CRITICAL!)
GRANT REFERENCE_USAGE ON DATABASE EXTERNAL_DB
  TO SHARE IN APPLICATION PACKAGE MY_APP_PKG;

-- 4. Grant access to share
GRANT USAGE ON SCHEMA MY_APP_PKG.SHARED_DATA
  TO SHARE IN APPLICATION PACKAGE MY_APP_PKG;
GRANT SELECT ON ALL VIEWS IN SCHEMA MY_APP_PKG.SHARED_DATA
  TO SHARE IN APPLICATION PACKAGE MY_APP_PKG;

在 setup_script.sql 中引用 shared_data.view_name（而不是原始数据库）。

Marketplace 发布

安全审核工作流

# 1. Deploy app
snow app run

# 2. Create version
snow app version create V1_0

# 3. Check security review status
snow app version list
# Wait for review_status = APPROVED

# 4. Set release directive
snow app release-directive set default --version V1_0 --patch 0 --channel DEFAULT

# 5. Create listing in Snowsight Provider Studio (UI only)

安全审核状态

状态	含义	操作
NOT_REVIEWED	尚未运行扫描	检查 DISTRIBUTION 是否为 EXTERNAL
IN_PROGRESS	正在扫描	等待
APPROVED	已通过	可以发布
REJECTED	未通过	修复问题或申诉
MANUAL_REVIEW	人工审核中	等待（可能需要数天）

会触发人工审核的情况：外部访问集成、Streamlit 组件、网络调用。

Provider Studio 字段

字段	最大长度	备注
标题	72 个字符	应用名称
副标题	128 个字符	一句话介绍
描述	10,000 个字符	HTML 编辑器
业务需求	最多 6 项	从下拉列表中选择
快速入门示例	最多 10 项	标题 + 描述 + SQL
数据字典	必填	数据列表的必填项（2025）

付费列表的先决条件

#	要求
1	完整的 Snowflake 账户（非试用账户）
2	ACCOUNTADMIN 角色
3	Provider Profile 已获批准
4	已配置 Stripe 账户
5	已接受 Provider 和 Consumer 条款
6	联系 Marketplace Ops

注意：无法将免费列表转换为付费列表，必须创建新列表。

Snowpark Python

会话设置

from snowflake.snowpark import Session

connection_params = {
    "account": "orgname-accountname",
    "user": "USERNAME",
    "password": "PASSWORD",  # Or use private_key_path
    "warehouse": "COMPUTE_WH",
    "database": "MY_DB",
    "schema": "PUBLIC"
}

session = Session.builder.configs(connection_params).create()

DataFrame 操作

# Read table
df = session.table("MY_TABLE")

# Filter and select
result = df.filter(df["STATUS"] == "ACTIVE") \
           .select("ID", "NAME", "CREATED_AT") \
           .sort("CREATED_AT", ascending=False)

# Execute
result.show()

# Collect to Python
rows = result.collect()

行访问（常见陷阱）

# WRONG - dict() doesn't work on Snowpark Row
config = dict(result[0])

# CORRECT - Access columns explicitly
row = result[0]
config = {
    'COLUMN_A': row['COLUMN_A'],
    'COLUMN_B': row['COLUMN_B'],
}

DML 统计信息（v4.2.0+）

2026 年 1 月新增：SnowflakeCursor.stats 属性可提供细粒度的 DML 统计信息，适用于 rowcount 不足以提供所需信息的操作（例如 CTAS 查询）。

# Before v4.2.0 - rowcount returns -1 for CTAS
cursor.execute("CREATE TABLE new_table AS SELECT * FROM source WHERE active = true")
print(cursor.rowcount)  # Returns -1 (not helpful!)

# After v4.2.0 - stats property shows actual row counts
cursor.execute("CREATE TABLE new_table AS SELECT * FROM source WHERE active = true")
print(cursor.stats)  # Returns {'rows_inserted': 1234, 'duplicates': 0, ...}

来源：版本 v4.2.0

UDF 和存储过程

from snowflake.snowpark.functions import udf, sproc

# Register UDF
@udf(name="my_udf", replace=True)
def my_udf(x: int) -> int:
    return x * 2

# Register Stored Procedure
@sproc(name="my_sproc", replace=True)
def my_sproc(session: Session, table_name: str) -> str:
    df = session.table(table_name)
    count = df.count()
    return f"Row count: {count}"

REST API（SQL API v2）

REST API 是通过 Cloudflare Workers 以编程方式访问 Snowflake 的基础。

端点

https://{org-account}.snowflakecomputing.com/api/v2/statements

必需的请求头（重要）

所有请求都必须包含这些请求头——缺少 Accept 会导致静默失败：

const headers = {
  'Authorization': `Bearer ${jwt}`,
  'Content-Type': 'application/json',
  'Accept': 'application/json',  // REQUIRED - "null" error if missing
  'User-Agent': 'MyApp/1.0',
};

异步查询处理

即使是简单查询也会返回异步响应（HTTP 202）。务必实现轮询：

// Submit returns statementHandle, not results
const submit = await fetch(url, { method: 'POST', headers, body });
const { statementHandle } = await submit.json();

// Poll until complete
while (true) {
  const status = await fetch(`${url}/${statementHandle}`, { headers });
  if (status.status === 200) break;  // Complete
  if (status.status === 202) {
    await sleep(2000);  // Still running
    continue;
  }
}

Workers 子请求限制

方案	限制	安全轮询
免费	50	45 次，每次间隔 2 秒 = 最长 90 秒
付费	1,000	100 次，每次间隔 500 毫秒 = 最长 50 秒

获取超时

Workers fetch() 默认没有超时。请始终使用 AbortController：

const response = await fetch(url, {
  signal: AbortSignal.timeout(30000),  // 30 seconds
  headers,
});

超时后取消

发生超时时取消查询，以避免产生仓库费用：

POST /api/v2/statements/{statementHandle}/cancel

完整实现请参阅 templates/snowflake-rest-client.ts。

已知问题

1. 账户标识符混淆

症状：JWT 身份验证静默失败，查询未显示在 Query History 中。

原因：在 JWT 声明中使用了组织账户格式，而不是账户定位符。

修复方法：使用 SELECT CURRENT_ACCOUNT() 获取实际的账户定位符。

2. 外部访问重置

症状：执行 snow app run 后 API 调用失败。

原因：每次部署时都会重置外部访问集成的关联。

修复方法：每次部署后重新运行 ALTER STREAMLIT ... SET EXTERNAL_ACCESS_INTEGRATIONS。

3. 发布渠道语法

症状：ALTER APPLICATION PACKAGE ... SET DEFAULT RELEASE DIRECTIVE 失败。

原因：启用发布渠道后，旧版 SQL 语法无法使用。

修复方法：使用 snow CLI：snow app release-directive set default --version V1_0 --patch 0 --channel DEFAULT

4. 工件嵌套

症状：文件出现在 streamlit/streamlit/ 中，而不是 streamlit/ 中。

原因：snowflake.yml 中的目录映射会嵌套文件夹名称。

修复方法：在 artifacts 中逐个列出文件，不要列出目录。

5. 缺少 REFERENCE_USAGE

症状："A view that is added to the shared content cannot reference objects from other databases"

原因：共享数据缺少 GRANT REFERENCE_USAGE ON DATABASE。

修复方法：使用外部数据库时，在运行 snow app run 前始终授予 REFERENCE_USAGE。

6. REST API 缺少 Accept 标头

症状：轮询请求出现 "Unsupported Accept header null is specified"。

原因：初始请求包含 Accept: application/json，但轮询请求没有。

修复方法：对所有请求（提交、轮询、取消）使用统一的标头辅助函数。

7. Workers Fetch 永久挂起

症状：Worker 持续挂起，等待 Snowflake 响应。

原因：Cloudflare Workers 的 fetch() 默认没有超时。

修复方法：对所有 Snowflake 请求始终使用 AbortSignal.timeout(30000)。

8. 子请求过多

症状：轮询期间出现 "Too many subrequests" 错误。

原因：每秒轮询一次 × 600 次尝试 = 600 个子请求，超出限制。

修复方法：每隔 2-5 秒轮询一次，最多尝试 45 次（免费版）或 100 次（付费版）。

9. 仓库未自动恢复（误判）

症状：查询返回 statementHandle，但始终无法完成（代码 090001 持续出现）。

原因：090001 表示“正在运行”，不是错误。仓库确实正在恢复，只是需要时间。

修复方法：自动恢复功能正常。请等待更长时间，或先显式恢复：POST /api/v2/warehouses/{wh}:resume

10. Connector 4.x 中的内存泄漏（当前问题）

错误：长时间运行的 Python 应用显示内存随时间增长 来源：GitHub Issue #2727，#2725 影响版本：snowflake-connector-python 4.0.0 - 4.2.0

发生原因：

• SessionManager 使用 defaultdict，阻止垃圾回收
• SnowflakeRestful.fetch() 持有的引用会在查询执行期间造成泄漏

预防措施： 复用连接，而不是反复创建新连接。修复正在通过 PR #2741 和 PR #2726 推进。

# AVOID - creates new connection each iteration
for i in range(1000):
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()
    cursor.execute("SELECT 1")
    cursor.close()
    conn.close()

# BETTER - reuse connection
conn = snowflake.connector.connect(...)
cursor = conn.cursor()
for i in range(1000):
    cursor.execute("SELECT 1")
cursor.close()
conn.close()

状态：预计在 connector v4.3.0 或更高版本中修复

11. 高峰使用期间的 AI 函数限流

错误：“由于使用量过高，请求已被限流。请重试。” 来源：Snowflake Cortex 文档 影响范围：所有 Cortex AI 函数（COMPLETE、FILTER、CLASSIFY 等）

发生原因： 在使用量高峰期间，AI/LLM 请求可能会被限流，以管理平台容量。被限流的请求会返回错误，需要手动重试。

预防措施： 使用指数退避实现重试逻辑：

import time
import snowflake.connector

def execute_with_retry(cursor, query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return cursor.execute(query).fetchall()
        except snowflake.connector.errors.DatabaseError as e:
            if "throttled" in str(e).lower() and attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
            else:
                raise

状态：已记录的行为，暂无修复计划

参考资料

• Snowflake CLI 文档
• SQL REST API 参考
• SQL API 身份验证
• Cortex AI 函数
• Native Apps 框架
• Snowpark Python
• Marketplace 发布

相关技能

• streamlit-snowflake - Snowflake 中的 Streamlit 应用