Snowflake/Snowpark中动态透视或横向展平JSON为列的实现

问题分析 #

你当前的代码核心问题是pivot参数使用逻辑错误：应该以KEY作为透视列，聚合VALUE字段，而非反过来。此外，当JSON键动态变化时，硬编码键列表的方式无法适配，需要先获取所有唯一键再动态构建透视逻辑。

方案1：已知所有可能的JSON键（静态场景） #

修正你的代码逻辑，调整pivot参数即可实现需求：

import snowflake.snowpark as snowpark

def main(session: snowpark.Session): 
    df = session.sql("select * from vnt")
    # 展平JSON并保留原始行的SEQ标识（用于关联同一行的键值对）
    df_flattened = df.join_table_function("flatten", df["SRC"]).select("SEQ", "KEY", "VALUE")
    # 以KEY为透视列，聚合VALUE，指定所有目标键
    df_pivoted = df_flattened.pivot("KEY", ['a','b','c','d']).max("VALUE")
    # 移除SEQ列，得到最终结果
    df_result = df_pivoted.drop("SEQ")
    return df_result

这里选择max("VALUE")作为聚合函数，是因为同一原始JSON行的同一个KEY只会有一个值，聚合操作不会影响结果；保留SEQ是为了确保同一行的键值对能被正确分组透视。

方案2：JSON键完全动态（未知键集合） #

如果JSON的键是动态变化的，需要先查询出所有唯一键，再动态构建透视逻辑：

import snowflake.snowpark as snowpark

def main(session: snowpark.Session): 
    # 1. 查询所有唯一的JSON键
    all_keys = session.sql("SELECT DISTINCT KEY FROM vnt, LATERAL FLATTEN(input => SRC)") \
                     .collect()
    # 提取键的字符串列表
    key_list = [row["KEY"] for row in all_keys]
    
    # 2. 展平JSON并保留原始行标识
    df_flattened = session.table("vnt").join_table_function("flatten", "SRC") \
                     .select("SEQ", "KEY", "VALUE")
    
    # 3. 动态执行透视操作
    df_pivoted = df_flattened.pivot("KEY", key_list).max("VALUE")
    
    # 4. 清理SEQ列并返回结果
    df_result = df_pivoted.drop("SEQ")
    return df_result

关键说明 #

• LATERAL FLATTEN用于从Variant类型中提取所有键值对，自动生成的SEQ字段用来标识同一原始JSON行的所有键值对，避免透视时跨行合并数据。
• 动态场景中先获取所有唯一键，无需硬编码，可自动适配JSON结构的变化。
• 聚合函数选择max/min/first均可，因为同一行同一KEY只会有一个值，不会影响最终输出。

验证结果 #

执行上述代码后，会得到你期望的输出：

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