如何用Python结合Merge Into批量更新Snowflake的temp表

一、可以用MERGE INTO实现（无需物理源表） #

Snowflake的MERGE INTO并不强制依赖物理源表，你可以通过VALUES子句直接构造虚拟源数据集，把内存中的列表/元组数据嵌入进去，通过ID匹配目标表后完成批量更新。

核心SQL示例 #

MERGE INTO temp AS target
USING (
    SELECT ID, W1, W2, W3
    FROM VALUES
        (1, 10, 20, 30),  -- ID=1对应的新W1/W2/W3值
        (2, 15, 25, 35),  -- ID=2对应的新值
        (3, 12, 22, 32)   -- ID=3对应的新值
    AS source(ID, W1, W2, W3)
) AS source
ON target.ID = source.ID
WHEN MATCHED THEN
    UPDATE SET
        target.W1 = source.W1,
        target.W2 = source.W2,
        target.W3 = source.W3;

结合Tkinter代码的实现 #

修改你的Update_Fn函数，动态生成参数化的MERGE语句，避免SQL注入风险：

def Update_Fn():
    updates = self.records.selection()
    update_data = []
    
    # 收集选中行的ID和对应输入的W1/W2/W3值
    for item in updates:
        item_vals = self.records.item(item, 'values')
        id_val = item_vals[0]
        # 注意：需根据实际逻辑调整为获取当前行对应输入框的值（比如用字典映射行ID到输入变量）
        w1_val = W1.get()
        w2_val = W2.get()
        w3_val = W3.get()
        update_data.append( (id_val, w1_val, w2_val, w3_val) )
    
    if not update_data:
        return  # 无更新数据直接返回
    
    # 构造参数化的MERGE SQL
    placeholders = ', '.join(['(%s, %s, %s, %s)'] * len(update_data))
    flatten_params = [val for row in update_data for val in row]
    
    sql_merge = f"""
    MERGE INTO temp AS target
    USING (
        SELECT ID, W1, W2, W3
        FROM VALUES {placeholders}
        AS source(ID, W1, W2, W3)
    ) AS source
    ON target.ID = source.ID
    WHEN MATCHED THEN
        UPDATE SET
            target.W1 = source.W1,
            target.W2 = source.W2,
            target.W3 = source.W3;
    """
    
    # 执行更新
    ctx = snowflake.connector.connect(
        user="你的用户名",
        password="你的密码",
        account="你的账号",
        database="你的数据库"
    )
    cs = ctx.cursor()
    try:
        cs.execute(sql_merge, flatten_params)
        ctx.commit()
    finally:
        cs.close()
        ctx.close()

二、替代方案 #

如果觉得MERGE INTO逻辑复杂，也可以选择以下两种更简单的方案：

方案1：批量执行UPDATE语句 #

通过executemany批量执行单条UPDATE语句，适合数据量较小的场景：

def Update_Fn():
    updates = self.records.selection()
    update_data = []
    
    for item in updates:
        item_vals = self.records.item(item, 'values')
        id_val = item_vals[0]
        w1_val = W1.get()
        w2_val = W2.get()
        w3_val = W3.get()
        # 参数顺序对应SQL中的占位符：W1, W2, W3, ID
        update_data.append( (w1_val, w2_val, w3_val, id_val) )
    
    sql_update = """
    UPDATE temp
    SET W1 = %s, W2 = %s, W3 = %s
    WHERE ID = %s;
    """
    
    ctx = snowflake.connector.connect(
        user="你的用户名",
        password="你的密码",
        account="你的账号",
        database="你的数据库"
    )
    cs = ctx.cursor()
    try:
        cs.executemany(sql_update, update_data)
        ctx.commit()
    finally:
        cs.close()
        ctx.close()

方案2：Pandas DataFrame写入临时表再MERGE #

如果数据量较大，可将内存数据转为DataFrame，写入Snowflake临时表后执行MERGE，效率更高：

import pandas as pd

def Update_Fn():
    updates = self.records.selection()
    update_rows = []
    
    for item in updates:
        item_vals = self.records.item(item, 'values')
        id_val = item_vals[0]
        w1_val = W1.get()
        w2_val = W2.get()
        w3_val = W3.get()
        update_rows.append({
            'ID': id_val,
            'W1': w1_val,
            'W2': w2_val,
            'W3': w3_val
        })
    
    if not update_rows:
        return
    
    df = pd.DataFrame(update_rows)
    
    # 连接Snowflake并写入临时表
    ctx = snowflake.connector.connect(
        user="你的用户名",
        password="你的密码",
        account="你的账号",
        database="你的数据库"
    )
    
    # 写入临时表（Snowflake临时表前缀用#）
    df.to_sql(
        name='#temp_update',
        con=ctx,
        index=False,
        if_exists='replace'
    )
    
    # 执行MERGE
    sql_merge = """
    MERGE INTO temp AS target
    USING #temp_update AS source
    ON target.ID = source.ID
    WHEN MATCHED THEN
        UPDATE SET
            target.W1 = source.W1,
            target.W2 = source.W2,
            target.W3 = source.W3;
    """
    
    cs = ctx.cursor()
    try:
        cs.execute(sql_merge)
        ctx.commit()
    finally:
        cs.close()
        ctx.close()

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