如何将CSV文件导入Django模型？探讨方案及性能问题

嘿，我来帮你搞定Django里CSV导入模型的事儿，尤其是你担心的大数据量性能问题～

你一开始的思路其实没问题——用Pandas读取CSV再通过bulk_create批量插入，这本身就是Django里批量导入的常用方案，比循环调用save()高效太多。不过要注意几个细节：

• 提前处理数据类型：比如日期字段要parse_dates，数字字段指定dtype，避免和模型字段类型不匹配。
• 不要在循环里做额外操作：直接把DataFrame的行转换成模型实例列表，减少内存开销。
• 用事务包裹：确保批量操作的原子性，要么全成功要么全回滚。

当数据量上去后，直接全量加载到内存肯定会有问题，这时候可以从这几个方向优化：

• 分块读取与插入：用Pandas的chunksize参数分块加载CSV，每块处理完就批量插入，避免一次性占用过多内存。
• 关闭自动提交 + 事务批量处理：Django默认是自动提交事务，用transaction.atomic()把每块的插入包裹起来，减少数据库的事务提交次数。
• 临时禁用索引与约束：导入前暂时关闭模型的索引、外键约束和触发器（比如PostgreSQL的DISABLE TRIGGER ALL），导入完成后再恢复并重建索引——这能大幅加快插入速度，因为数据库不用每次插入都更新索引。
• 数据库原生导入（极致性能）：如果用的是PostgreSQL，可以用copy_from方法（Django的connection.cursor()执行原生SQL），这比bulk_create快好几倍，适合超大规模数据导入，但需要确保CSV格式和数据库表结构完全匹配。

• 数据验证：导入前一定要校验数据！比如必填字段是否为空、格式是否正确，可以用Django的ModelForm或者Serializer来验证每个实例，避免批量插入失败后回滚。
• 错误处理：分块导入时，要捕获异常并记录错误行，比如把导入失败的行写入日志或另一个CSV，方便后续排查。
• 信号触发问题：bulk_create不会触发模型的pre_save和post_save信号，如果你的业务逻辑依赖这些信号，要么手动触发，要么改用其他方式（比如分块后循环save()，但性能会下降，需要权衡取舍）。
• batch_size设置：bulk_create的batch_size参数要合理，太大可能导致数据库超时，太小会增加请求次数，一般1000-5000比较合适，根据数据库配置调整。

基础批量导入 #

import pandas as pd
from django.db import transaction
from myapp.models import MyModel

def import_csv(file_path):
    # 读取CSV并处理数据类型
    df = pd.read_csv(
        file_path,
        parse_dates=['created_at'],  # 解析日期字段
        dtype={'user_id': int, 'amount': float}  # 指定数字类型
    )
    
    # 转换为模型实例列表
    instances = [
        MyModel(
            username=row['username'],
            email=row['email'],
            amount=row['amount'],
            created_at=row['created_at']
        )
        for _, row in df.iterrows()
    ]
    
    # 事务内批量插入
    with transaction.atomic():
        MyModel.objects.bulk_create(instances, batch_size=1000)

大数据量分块导入 #

def import_large_csv(file_path):
    chunk_size = 2000  # 每块处理2000行
    # 分块读取CSV
    for chunk in pd.read_csv(
        file_path,
        chunksize=chunk_size,
        parse_dates=['created_at'],
        dtype={'user_id': int, 'amount': float}
    ):
        instances = [
            MyModel(
                username=row['username'],
                email=row['email'],
                amount=row['amount'],
                created_at=row['created_at']
            )
            for _, row in chunk.iterrows()
        ]
        # 每块单独用事务包裹
        with transaction.atomic():
            MyModel.objects.bulk_create(instances, batch_size=chunk_size)

临时禁用索引优化（PostgreSQL） #

from django.db import connection

def import_with_index_optimization(file_path):
    table_name = MyModel._meta.db_table  # 获取模型对应的数据库表名
    
    # 禁用触发器和约束
    with connection.cursor() as cursor:
        cursor.execute(f'ALTER TABLE {table_name} DISABLE TRIGGER ALL;')
    
    try:
        # 执行分块导入
        import_large_csv(file_path)
    finally:
        # 恢复触发器和约束，重建索引
        with connection.cursor() as cursor:
            cursor.execute(f'ALTER TABLE {table_name} ENABLE TRIGGER ALL;')
            cursor.execute(f'REINDEX TABLE {table_name};')

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