如何将Snowflake Python存储过程结果保存为表并关联查询

一、当前存储过程的限制 #

Snowflake的存储过程（PROCEDURE）无法直接在SELECT语句的FROM子句中像表或表函数那样被引用，这就是你无法用CREATE TABLE ... FROM pivot_table(...)语法的核心原因。要实现你的需求，需将其改为Python表函数（TABLE FUNCTION）。

二、Python表函数实现动态透视表 #

将原存储过程改写为表函数后，可直接在FROM子句中使用，支持关联查询和创建表。以下是修改后的代码：

CREATE OR REPLACE FUNCTION pivot_table(tableName VARCHAR, index VARCHAR, columns VARCHAR, vals VARCHAR)
RETURNS TABLE()
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python', 'pandas')
HANDLER = 'pivot_the_table'
AS
$$
import pandas as pd

def pivot_the_table(table_name, index, columns, vals):
    from snowflake.snowpark import Session
    session = Session.builder.getOrCreate()
    
    df = session.table(table_name).to_pandas()
    df.columns = df.columns.str.lower()
    SEP = ","
    
    # 处理多字段输入
    index_cols = index.split(SEP) if SEP in index else [index]
    pivot_cols = columns.split(SEP) if SEP in columns else [columns]
    value_cols = vals.split(SEP) if SEP in vals else [vals]
    
    # 生成透视表
    pvt = df.pivot_table(index=index_cols, values=value_cols, columns=pivot_cols, aggfunc="sum").reset_index()
    
    # 修复透视表的层级列名
    pvt.columns = ['_'.join(col).strip('_') if isinstance(col, tuple) else col for col in pvt.columns]
    
    return session.create_dataframe(pvt)
$$;

关键修改说明： #

• 将PROCEDURE改为FUNCTION，符合表函数的定义规范
• 调整handler函数参数：表函数无需显式传入session，通过Session.builder.getOrCreate()自动获取
• 修复透视表列名：pandas生成的透视表会产生层级列名，这里将其合并为单一命名（如sales_2023）
• 显式声明pandas依赖包，避免运行时依赖问题

三、使用表函数创建表并关联查询 #

现在你可以用如下语句实现需求：

CREATE OR REPLACE TABLE some_table AS
SELECT
   *
FROM (
    SELECT * FROM TABLE(pivot_table('your_source_table', 'month', 'year', 'sales'))
) AS pvt
LEFT JOIN some_other_table AS oth ON pvt.month = oth.month;

注意：调用表函数时必须用TABLE()包裹。

四、Snowflake原生动态透视表替代方案（可选） #

若不想使用Python，Snowflake原生支持动态透视，但需借助SQL存储过程生成动态SQL：

1. 查询获取所有需要透视的列值
2. 拼接生成包含PIVOT子句的动态SQL
3. 执行动态SQL创建表

不过这种方式灵活性不如Python实现，且需要编写复杂的SQL逻辑。

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