如何让Snowflake/Snowpark脚本块返回COPY INTO语句的执行结果？

要在包含TRUNCATE和COPY INTO的匿名块里拿到COPY INTO的执行结果，核心是显式捕获COPY INTO的结果集并通过RETURN返回，具体实现如下：

匿名块脚本写法 #

DECLARE
    copy_result RESULTSET; -- 声明变量存储COPY INTO结果
BEGIN
    -- 先执行TRUNCATE
    TRUNCATE TABLE your_target_table;
    -- 捕获COPY INTO的结果集
    copy_result := (SELECT * FROM TABLE(COPY INTO your_target_table FROM @your_s3_stage));
    -- 返回结果集
    RETURN TABLE(copy_result);
END;

关键步骤说明 #

• 用DECLARE定义RESULTSET类型的变量，专门用来存放COPY INTO返回的结果数据
• COPY INTO语句本身返回的是结果集，需要通过TABLE()函数转换为可查询的表结构，再用子查询赋值给变量
• 最后通过RETURN TABLE(copy_result)把结果集返回，这样Snowpark执行这个匿名块时就能获取到COPY INTO的执行结果，而不是默认的[Row(anonymous block=None)]

Snowpark Python中调用示例 #

import pandas as pd
from snowflake.snowpark import Session

# 初始化Snowpark会话
session = Session.builder.configs({
    "account": "your_account",
    "user": "your_user",
    "password": "your_password",
    "warehouse": "your_warehouse",
    "database": "your_db",
    "schema": "your_schema"
}).create()

# 执行匿名块并获取结果
copy_result = session.sql("""
DECLARE
    copy_result RESULTSET;
BEGIN
    TRUNCATE TABLE your_target_table;
    copy_result := (SELECT * FROM TABLE(COPY INTO your_target_table FROM @your_s3_stage));
    RETURN TABLE(copy_result);
END;
""").collect()

# 将结果存入日志表
if copy_result:
    log_df = pd.DataFrame(copy_result)
    session.write_pandas(log_df, "your_log_table", mode="append", quote_identifiers=False)

这样既不用拆分两条语句，又能正常捕获COPY INTO的执行结果，适配你的ELT管道需求。

内容的提问来源于stack exchange，提问作者Sebastian