如何用Python手动实现Snowflake PUT及COPY INTO写入VARIANT列

核心问题说明 #

需要实现两个目标：

• 脱离write_pandas封装，手动在Python中完成「本地文件暂存→PUT上传→COPY INTO入库」的全流程
• 解决XML数据写入VARIANT列时首尾自动加双引号，导致后续类型转换、XML解析失败的问题

注：quote_identifiers=False参数仅控制表名、列名这类SQL标识符是否加引号，和数据内容的转义逻辑完全无关，这也是之前加了该参数仍然出现引号包裹问题的根本原因。

一、手动实现PUT+COPY INTO全流程实操代码 #

以下示例直接将原始XML字符串写入VARIANT列，全程不会额外添加引号，不需要额外创建命名阶段，直接使用目标表自带的内置阶段即可：

import pandas as pd
import snowflake.connector
import os
from tempfile import NamedTemporaryFile

# 替换为实际业务数据，rowData为存储原始XML字符串的列表
rowData = ["<root><bizId>1001</bizId><msg>test content</msg></root>"]
pandasDF = pd.DataFrame(rowData, columns=["DATACOLUMN"])

# 替换为实际Snowflake连接信息
conn = snowflake.connector.connect(
    user="你的账号",
    password="你的密码",
    account="你的Snowflake账户标识",
    warehouse="使用的计算仓库名",
    database="目标数据库名",
    schema="目标schema名"
)
cur = conn.cursor()
temp_file_path = ""

try:
    # 1. 生成本地临时文件，逐行写入原始XML，不做任何引号转义
    with NamedTemporaryFile(mode="w", suffix=".csv", delete=False, encoding="utf-8") as f:
        for xml_content in pandasDF["DATACOLUMN"]:
            f.write(xml_content + "\n")
        temp_file_path = f.name

    # 2. 执行PUT命令，将本地文件上传到目标表的内置阶段
    put_cmd = f"PUT 'file://{temp_file_path}' @%OI_PNET_GET_PERFORMX_UPL2 OVERWRITE=TRUE AUTO_COMPRESS=TRUE"
    cur.execute(put_cmd)

    # 3. 执行COPY INTO命令加载数据，核心是关闭字段分割、引号识别逻辑
    copy_cmd = """
    COPY INTO OI_PNET_GET_PERFORMX_UPL2 (DATACOLUMN)
    FROM @%OI_PNET_GET_PERFORMX_UPL2
    FILE_FORMAT = (
        TYPE = CSV,
        PARSE_HEADER = FALSE,
        FIELD_DELIMITER = NONE,
        FIELD_OPTIONALLY_ENCLOSED_BY = NONE,
        RECORD_DELIMITER = '\n',
        COMPRESSION = AUTO
    )
    PURGE = TRUE
    """
    cur.execute(copy_cmd)
    print(f"数据写入完成，共加载{cur.rowcount}行")

finally:
    # 资源清理
    if os.path.exists(temp_file_path):
        os.remove(temp_file_path)
    cur.close()
    conn.close()

关键参数说明 #

• @%OI_PNET_GET_PERFORMX_UPL2：@%前缀代表对应表的内置阶段，不需要提前手动创建，权限和表绑定
• FIELD_DELIMITER = NONE：不对单行内容做字段分割，整行作为单个字段处理
• FIELD_OPTIONALLY_ENCLOSED_BY = NONE：不识别任何引号包裹符，不会自动解析、添加字段首尾的引号，从根源解决多余引号问题
• PURGE = TRUE：数据加载完成后自动清理阶段内的临时文件，不需要手动删除

二、继续使用write_pandas的快速修复方案 #

如果不想手动维护PUT/COPY INTO逻辑，可以在调用write_pandas时自定义文件格式参数，覆盖默认的引号转义规则即可：

from snowflake.connector.pandas_tools import write_pandas
import pandas as pd

pandasDF = pd.DataFrame(rowData, columns=["DATACOLUMN"])

success, nchunks, nrows, _ = write_pandas(
    conn=engine,
    df=pandasDF,
    table_name="OI_PNET_GET_PERFORMX_UPL2",
    quote_identifiers=False,
    # 传入和手动加载一致的文件格式规则，关闭引号转义
    file_format="(TYPE=CSV, FIELD_DELIMITER=NONE, FIELD_OPTIONALLY_ENCLOSED_BY=NONE, RECORD_DELIMITER='\n')"
)

写入验证 #

写入完成后执行以下SQL校验数据正确性，正常返回的DATACOLUMN类型为OBJECT，没有首尾引号，可直接使用XMLGET函数做解析：

SELECT DATACOLUMN, TYPEOF(DATACOLUMN) AS col_type FROM OI_PNET_GET_PERFORMX_UPL2 LIMIT 10;

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