Snowflake如何将时序行转换为连续列？

我有一张存储物品流转步骤的表，用SQL创建语句如下：

CREATE OR REPLACE TABLE ITEM_PROCESS (PK_ID, ID, DATE, STATUS, NAME)
AS
SELECT * FROM VALUES
(1, 1, '2023-01-01 00:00:00'::TIMESTAMP_NTZ, 'Arrive in Warehouse', 'PS5'),
(2, 1, '2023-02-01 00:00:00'::TIMESTAMP_NTZ, 'Packaging', 'PS5'),
(3, 1, '2023-03-01 00:00:00'::TIMESTAMP_NTZ, 'Shipping', 'PS5'),
(4, 1, '2023-04-01 00:00:00'::TIMESTAMP_NTZ, 'Received', 'PS5'),
(5, 2, '2023-01-01 00:00:00'::TIMESTAMP_NTZ, 'Arrive in Warehouse', 'Fan'),
(6, 2, '2023-01-02 00:00:00'::TIMESTAMP_NTZ, 'Checking failures', 'Fan'),
(7, 2, '2023-01-03 00:00:00'::TIMESTAMP_NTZ, 'Shipping', 'Fan'),
(8, 2, '2023-01-04 00:00:00'::TIMESTAMP_NTZ, 'Received', 'Fan')

需要对STATUS列执行类似PIVOT的操作，将其拆分为多列并填入对应的DATE值，预期输出如下：

约束条件 #

• 流程状态的顺序预先定义在一张单独的表中，格式示例：(1, 'Arrive in Warehouse'), (2, 'Checking failures'), (3, 'Packaging'),...
• 透视后必须保留原表的列顺序逻辑：比如原表中NAME在最后一列，透视后也需放在最后；实际业务表中DATE列前有3列，STATUS列后还有大量列
• 每个ID-DATE-STATUS组合唯一，无需去重
• 表中每行有主键PK_ID，不确定是否有用

可接受SQL或存储过程实现，优先选择Python存储过程。

方法一：Python存储过程（推荐） #

该方法可动态读取状态顺序表，自动生成符合列顺序要求的透视结果，同时严格保留原表中STATUS列前后的其他列顺序。

步骤1：创建状态顺序表（确保表已存在） #

CREATE OR REPLACE TABLE PROCESS_ORDER (ORDER_NUM, STATUS_NAME)
AS
SELECT * FROM VALUES
(1, 'Arrive in Warehouse'),
(2, 'Checking failures'),
(3, 'Packaging'),
(4, 'Shipping'),
(5, 'Received');

步骤2：编写Python存储过程 #

CREATE OR REPLACE PROCEDURE PIVOT_ITEM_PROCESS()
RETURNS TABLE()
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'pivot_handler'
AS
$$
import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col, max as sf_max, to_date

def pivot_handler(session: snowpark.Session):
    # 读取状态顺序表，获取按定义顺序排列的状态名称
    process_order_df = session.table("PROCESS_ORDER").order_by("ORDER_NUM")
    status_list = process_order_df.select("STATUS_NAME").collect()
    status_names = [row.STATUS_NAME for row in status_list]
    
    # 读取原表，拆分出STATUS列前后的所有列
    item_process_df = session.table("ITEM_PROCESS")
    all_columns = item_process_df.columns
    status_index = all_columns.index("STATUS")
    # 筛选STATUS之前的列（排除DATE，因为DATE是透视值）和STATUS之后的列
    pre_status_cols = [col_name for col_name in all_columns[:status_index] if col_name != "DATE"]
    post_status_cols = all_columns[status_index+1:]
    
    # 执行透视操作：按前置列+后置列分组，将STATUS转为列，DATE取对应唯一值
    pivoted_df = item_process_df.group_by(pre_status_cols + post_status_cols)\
                                .pivot("STATUS", status_names)\
                                .agg(sf_max(to_date(col("DATE"))))
    
    # 调整最终列顺序：前置列+状态列（按定义顺序）+后置列
    final_column_order = pre_status_cols + status_names + post_status_cols
    final_df = pivoted_df.select(final_column_order)
    
    return final_df
$$;

步骤3：执行存储过程 #

CALL PIVOT_ITEM_PROCESS();

方法二：静态SQL PIVOT（适合状态固定的场景） #

如果状态列表长期不变，可以直接编写静态PIVOT语句，手动指定列顺序以符合要求：

SELECT 
    ID,
    "Arrive in Warehouse",
    "Checking failures",
    "Packaging",
    "Shipping",
    "Received",
    NAME
FROM (
    SELECT 
        ID,
        STATUS,
        TO_DATE(DATE) AS DATE_VAL,
        NAME
    FROM ITEM_PROCESS
)
PIVOT (
    MAX(DATE_VAL) FOR STATUS IN (
        'Arrive in Warehouse' AS "Arrive in Warehouse",
        'Checking failures' AS "Checking failures",
        'Packaging' AS "Packaging",
        'Shipping' AS "Shipping",
        'Received' AS "Received"
    )
)
ORDER BY ID;

若原表中STATUS前后有更多列，需在子查询中包含这些列，并在最终SELECT语句中按原表顺序排列。

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