Django中CSV导入速度极慢，求更快实现方案

你的代码慢的核心原因是每次update_or_create都会触发两次数据库操作（先查询是否存在，再执行插入/更新），2100行数据就会产生至少4200次数据库交互，加上Django ORM的额外开销，导致耗时过长。以下是几个高效的优化方法：

方法1：批量查询+分组批量操作 #

先一次性读取所有CSV数据，批量查询已存在的sn，再将数据分为「新增」和「更新」两组，分别用批量API处理，大幅减少数据库交互次数：

from django.db import transaction

def import_csv(reader):
    # 收集所有CSV数据和对应的sn
    rows = list(reader)
    sn_list = [row[0] for row in rows]
    
    # 批量查询已存在的sn及其对象，用字典映射方便快速查找
    existing_items = {item.sn: item for item in Item.objects.filter(sn__in=sn_list)}
    
    create_list = []
    update_list = []
    
    for row in rows:
        sn = row[0]
        if sn in existing_items:
            # 更新已有对象的字段
            item = existing_items[sn]
            item.field_a = row[1]
            item.field_b = row[2]
            item.field_c = row[3]
            item.field_d = row[4]
            update_list.append(item)
        else:
            # 构建新对象加入新增列表
            create_list.append(Item(
                sn=sn,
                field_a=row[1],
                field_b=row[2],
                field_c=row[3],
                field_d=row[4]
            ))
    
    # 原子事务中执行批量操作，保证数据一致性
    with transaction.atomic():
        if create_list:
            Item.objects.bulk_create(create_list)
        if update_list:
            Item.objects.bulk_update(update_list, ['field_a', 'field_b', 'field_c', 'field_d'])

方法2：使用SQLite原生INSERT OR REPLACE批量执行 #

直接调用数据库原生语句批量处理，绕开Django ORM的单条操作开销，速度提升最明显：

from django.db import connection

def import_csv(reader):
    # 跳过CSV表头（如果有）
    next(reader)
    
    # 整理批量插入数据
    data_batch = []
    for row in reader:
        data_batch.append((row[0], row[1], row[2], row[3], row[4]))
    
    # 执行原生SQL批量操作
    with connection.cursor() as cursor:
        cursor.executemany("""
            INSERT OR REPLACE INTO yourapp_item (sn, field_a, field_b, field_c, field_d)
            VALUES (?, ?, ?, ?, ?)
        """, data_batch)

注意：替换yourapp_item为你的实际数据库表名（Django默认规则是「应用名_模型名」小写）；INSERT OR REPLACE会删除旧记录再插入新记录，若模型有其他需保留的关联字段，谨慎使用。

方法3：最小改动优化——手动控制事务 #

Django默认每次ORM操作都会自动提交事务，用transaction.atomic()包裹所有操作，减少事务提交的开销，是对原有代码的最小改动：

from django.db import transaction

def import_csv(reader):
    with transaction.atomic():
        for row in reader:
            Item.objects.update_or_create(
                sn=row[0],
                defaults={
                    "field_a": row[1],
                    "field_b": row[2],
                    "field_c": row[3],
                    "field_d": row[4],
                }
            )

该方法能小幅提升速度，但效果不如前两种方法显著。

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