使用内部阶段执行Snowflake的COPY INTO命令加载数据失败

1. 未指定文件格式处理CSV表头 #

你的CSV包含表头，但COPY INTO默认会将第一行（表头）当作数据行加载，加上你设置了ON_ERROR='continue'，错误会被直接跳过，导致数据未被正确写入表中。必须明确设置跳过表头的规则。

2. COPY INTO未匹配到上传后的文件 #

PUT命令上传本地CSV后，Snowflake会自动将文件压缩为.gz格式，若仅指定阶段名，COPY INTO可能无法精准匹配到目标文件，需要明确指向压缩后的文件路径。

3. 冗余Cursor创建（非核心问题，但可优化） #

你每次执行SQL都新建Cursor，虽不影响功能，但复用同一Cursor更便于获取执行结果、排查问题。

修复后的代码 #

import snowflake.connector

conn = snowflake.connector.connect(
    user='username',
    password='password',
    account='account',
)     

# 复用同一个cursor
curs = conn.cursor()

try:
    # 创建并切换数据库
    curs.execute("CREATE DATABASE IF NOT EXISTS test_snowflake")
    curs.execute("USE DATABASE test_snowflake")

    # 创建目标表
    curs.execute("""
        CREATE OR REPLACE TABLE shoppers_table(
            first_name string, 
            last_name string
        )
    """)

    # 创建适配CSV的文件格式（跳过表头）
    curs.execute("""
        CREATE OR REPLACE FILE FORMAT csv_format 
        TYPE = CSV 
        FIELD_OPTIONALLY_ENCLOSED_BY = '"'
        SKIP_HEADER = 1
    """)

    # 创建内部阶段并关联文件格式
    curs.execute("""
        CREATE OR REPLACE STAGE shoppers_table 
        FILE_FORMAT = csv_format
        COPY_OPTIONS = (ON_ERROR='continue')
    """)

    # 上传本地文件到阶段
    put_result = curs.execute("PUT file:///Users/navid/Desktop/snowflake_test.csv @shoppers_table")
    # 打印上传结果，确认文件是否成功上传
    print("PUT执行结果:", put_result.fetchall())

    # 复制阶段中的压缩文件到目标表
    copy_result = curs.execute("COPY INTO shoppers_table FROM @shoppers_table/snowflake_test.csv.gz")
    # 打印复制结果，查看实际加载行数
    print("COPY执行结果:", copy_result.fetchall())

finally:
    # 关闭资源
    curs.close()
    conn.close()

额外排查建议 #

• 执行LIST @shoppers_table;查看阶段内的文件列表，确认上传的文件是否存在、文件名是否正确。
• 执行COPY INTO shoppers_table FROM @shoppers_table FILE_FORMAT = csv_format ON_ERROR='continue' VALIDATION_MODE = RETURN_ALL_ERRORS;，查看具体的加载错误信息，精准定位问题。

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