如何在Snowflake中创建适配已透视CSV的表以实现月度自动化加载

1. 先明确透视CSV的结构（核心前提） #

先梳理你的CSV固定列与动态列：

• 固定列：维度列（如产品ID、地区）、月度标识列（如report_month，格式建议用YYYY-MM-DD或YYYYMM）
• 动态列：月度指标列（要么每个CSV的指标列名固定，仅数据对应当月；要么列名带月份后缀，如202401_sales）

以下分两种常见场景给出建表与加载方案：

场景1：月度CSV列结构固定（仅数据对应当月） #

比如每个CSV的列都是product_id, region, report_month, monthly_sales, monthly_cost，仅report_month值对应不同月份。

建表语句 #

创建兼容所有月度数据的固定结构表，日期列用标准类型保证一致性：

CREATE OR REPLACE TABLE PIVOTED_MONTHLY_DATA (
    PRODUCT_ID VARCHAR(50),
    REGION VARCHAR(30),
    REPORT_MONTH DATE, -- 存储月度标识，如'2024-01-01'代表1月
    MONTHLY_SALES NUMBER(18,2),
    MONTHLY_COST NUMBER(18,2)
);

自动化加载要点 #

• 用COPY INTO的默认追加模式加载，不会覆盖已有日期数据：

COPY INTO PIVOTED_MONTHLY_DATA
FROM '@YOUR_STAGE/path/to/monthly_data_202401.csv'
FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1)
ON_ERROR = CONTINUE;

• 用Snowflake任务实现每月自动触发：设置每月固定时间执行加载脚本，比如每月1号加载上月数据。

场景2：透视CSV为宽表（列名带月份后缀） #

如果CSV是累计宽表（如product_id, region, 202312_sales, 202401_sales...），或每月新增列的宽表，推荐转成窄表结构（长期可扩展）。

方案：转成标准化窄表（推荐） #

将宽表的月度列转成report_month+metric_value的键值对，表结构无需随月份新增修改：

步骤1：创建窄表

CREATE OR REPLACE TABLE STANDARD_MONTHLY_DATA (
    PRODUCT_ID VARCHAR(50),
    REGION VARCHAR(30),
    REPORT_MONTH DATE,
    METRIC_NAME VARCHAR(30), -- 如'SALES'、'COST'
    METRIC_VALUE NUMBER(18,2)
);

步骤2：加载宽表并转窄表

先加载到临时表，再用FLATTEN函数转结构插入目标表：

-- 加载CSV到临时表
CREATE OR REPLACE TEMP TABLE TEMP_WIDE_DATA
AS SELECT * FROM '@YOUR_STAGE/path/to/pivoted_wide.csv'
FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1);

-- 转窄表插入目标表
INSERT INTO STANDARD_MONTHLY_DATA
SELECT 
    PRODUCT_ID,
    REGION,
    TO_DATE(SPLIT_PART(KEY, '_', 1), 'YYYYMM') AS REPORT_MONTH, -- 从列名提取月份
    SPLIT_PART(KEY, '_', 2) AS METRIC_NAME,
    VALUE::NUMBER(18,2) AS METRIC_VALUE
FROM TEMP_WIDE_DATA,
LATERAL FLATTEN(INPUT => OBJECT_CONSTRUCT(*), EXCLUDE => ('PRODUCT_ID', 'REGION'));

2. 保证日期列不受影响的关键措施 #

• 用标准日期类型：日期列设为DATE/TIMESTAMP，加载时用TO_DATE做类型转换，避免字符串格式混乱
• 采用追加模式加载：用COPY INTO默认追加或INSERT语句，禁止用REPLACE覆盖全表
• 添加唯一约束：防止重复插入同一维度+日期的记录：

ALTER TABLE STANDARD_MONTHLY_DATA ADD CONSTRAINT UNIQUE_DATA UNIQUE (PRODUCT_ID, REGION, REPORT_MONTH, METRIC_NAME);

3. 自动化加载实现（Snowflake任务） #

创建每月自动触发的任务，执行加载流程：

-- 创建任务（每月1号UTC1点执行）
CREATE OR REPLACE TASK LOAD_MONTHLY_PIVOT_DATA
WAREHOUSE = YOUR_WH
SCHEDULE = 'USING CRON 0 1 1 * * UTC'
AS
BEGIN
    -- 加载上月CSV到临时表
    CREATE OR REPLACE TEMP TABLE TEMP_MONTHLY_DATA
    AS SELECT * FROM '@YOUR_STAGE/path/to/monthly_pivot_' || TO_CHAR(CURRENT_DATE() - INTERVAL '1 MONTH', 'YYYYMM') || '.csv'
    FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1);

    -- 插入窄表（假设指标列名为MONTHLY_SALES）
    INSERT INTO STANDARD_MONTHLY_DATA
    SELECT 
        PRODUCT_ID,
        REGION,
        TO_DATE(TO_CHAR(CURRENT_DATE() - INTERVAL '1 MONTH', 'YYYYMM'), 'YYYYMM') AS REPORT_MONTH,
        'SALES' AS METRIC_NAME,
        MONTHLY_SALES AS METRIC_VALUE
    FROM TEMP_MONTHLY_DATA;
END;

-- 启用任务
ALTER TASK LOAD_MONTHLY_PIVOT_DATA RESUME;

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