Python导入CSV到MySQL时实现仅追加新行或更新已有记录避免重复

原有代码核心问题 #

• 每次执行都会执行DROP TABLE IF EXISTS AuditLogs删除并重建表，属于全量覆盖逻辑，完全不符合增量同步需求
• 插入数据使用普通INSERT语句，没有重复校验逻辑，重复执行必然产生重复记录

改造核心思路 #

1. 确定业务唯一标识：从AuditLogs的字段中选择不会重复的字段作为主键（比如Id或ActivityId，根据实际业务规则选即可），建表时添加主键约束
2. 使用MySQL原生的INSERT ... ON DUPLICATE KEY UPDATE语法：当插入数据主键冲突时，自动更新其余字段；无冲突时直接插入新数据，完美满足「新增行追加、更新行同步、不变行不重复加载」的需求
3. 移除删表重建逻辑，改为表不存在时才创建，已存在则直接复用
4. 优化提交逻辑，改为批量提交提升导入性能

修改后完整代码 #

import mysql.connector as msql
from mysql.connector import Error
import pandas as pd

# 此处保留你原有读取CSV到AuditLogData的逻辑即可
# AuditLogData = pd.read_csv("你的CSV文件路径")

try:
    conn = msql.connect(host='127.0.0.1', database='AuditLogDB',user='root', 
    password='Pharma123')
    if conn.is_connected():
        cursor = conn.cursor()
        cursor.execute("select database();")
        record = cursor.fetchone()
        print("已连接到数据库: ", record)

        # 移除删表逻辑，改为表不存在时才创建
        # 此处默认Id为业务唯一主键，如果你实际用其他字段做唯一键，把PRIMARY KEY加到对应字段上即可
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS AuditLogs (
            Id nvarchar (50) PRIMARY KEY, 
            CreationTime nvarchar (50),
            UserId nvarchar(50),
            Operation nvarchar(50),
            Workload nvarchar(50),
            DatasetName nvarchar(50),
            ReportName nvarchar(50),
            ReportId nvarchar(50),
            Activity nvarchar(50),
            ActivityId nvarchar(50),
            DatasetId nvarchar(50),
            LastRefreshTime nvarchar(50),
            ObjectId nvarchar(255),
            SiteUrl nvarchar(50),
            SourceFileName nvarchar(50),
            ClientIP nvarchar(50),
            UserAgent nvarchar(50),
            Email nvarchar(50)
        )
        """
        cursor.execute(create_table_sql)
        print("AuditLogs表初始化完成")

        # 改写INSERT语句为带冲突更新的逻辑
        insert_sql = """
        INSERT INTO AuditLogDB.AuditLogs VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
        ON DUPLICATE KEY UPDATE
            CreationTime = VALUES(CreationTime),
            UserId = VALUES(UserId),
            Operation = VALUES(Operation),
            Workload = VALUES(Workload),
            DatasetName = VALUES(DatasetName),
            ReportName = VALUES(ReportName),
            ReportId = VALUES(ReportId),
            Activity = VALUES(Activity),
            ActivityId = VALUES(ActivityId),
            DatasetId = VALUES(DatasetId),
            LastRefreshTime = VALUES(LastRefreshTime),
            ObjectId = VALUES(ObjectId),
            SiteUrl = VALUES(SiteUrl),
            SourceFileName = VALUES(SourceFileName),
            ClientIP = VALUES(ClientIP),
            UserAgent = VALUES(UserAgent),
            Email = VALUES(Email)
        """
        # 批量执行插入，大幅提升性能
        cursor.executemany(insert_sql, AuditLogData.values.tolist())
        conn.commit()
        print(f"同步完成，共处理{cursor.rowcount}条记录")

except Error as e:
    print("数据库操作出错: ", e)
finally:
    if conn.is_connected():
        cursor.close()
        conn.close()
        print("数据库连接已关闭")

注意事项 #

• 主键选择要准确：如果Id字段不是全局唯一的，你可以替换为ActivityId等其他能唯一标识单条日志的字段，建表时把主键约束加在对应字段上即可
• 如果CSV文件体量非常大，可以额外记录CSV的最后读取行号/最后修改时间，每次只读取新增行再做校验，进一步提升同步效率

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