怎样在Python中用Django实现批量数据高效导入？-Python教程-PHP中文网

bulk_create()比循环save()快得多，因其将多条INSERT合并为单条SQL语句，避免重复的模型验证、信号触发、事务开销和网络往返；实测10万条数据，前者秒级完成，后者可能卡几分钟甚至OOM。

怎样在python中用django实现批量数据高效导入？

为什么 `bulk_create()` 比循环 `save()` 快得多

因为 bulk_create() 把多条 INSERT 合并成一条 SQL 语句发给数据库，避免了每次 save() 都触发模型验证、信号、事务开销和网络往返。实测 10 万条数据，前者通常在秒级完成，后者可能卡住几分钟甚至 OOM。

但要注意：它跳过 Model.full_clean() 和 pre_save/post_save 信号，也不返回带主键的实例（除非指定 update_fields 或用 PostgreSQL 的 returning）。所以导入前得自己校验字段合法性。

必须确保所有对象的字段值已清洗干净（比如 None 不会自动转成空字符串）
外键字段必须传入已存在的 id 或实例，不能传字符串名称
MySQL 不支持 ignore_conflicts=True，PostgreSQL 才行；若需去重，得先用 get_or_create 或手动查重

如何安全处理 CSV 文件并分批调用 `bulk_create()`

直接读完整 CSV 再一次性 bulk_create() 容易内存爆炸，尤其当单文件超 50 万行时。推荐边读边构造模型实例，每 1000–5000 条提交一次。

示例关键逻辑：

with open('data.csv') as f:
    reader = csv.DictReader(f)
    objs = []
    for row in reader:
        # 字段映射 + 类型转换（如 int(row['age'])）
        try:
            obj = MyModel(
                name=row['name'].strip(),
                age=int(row['age']),
                category_id=int(row['category_id'])
            )
            objs.append(obj)
        except (ValueError, KeyError):
            # 记录错误行号和原因，别让整个批次失败
            logger.warning(f"Skip invalid row: {row}")
<pre class="brush:php;toolbar:false;">    if len(objs) >= 2000:
        MyModel.objects.bulk_create(objs, batch_size=2000)
        objs.clear()

# 处理余量
if objs:
    MyModel.objects.bulk_create(objs, batch_size=2000)

python-flask-vue-creator

创建 Flask + Vue.js 全栈项目，采用单端口架构、uv 依赖管理、结构化日志和生产级前端设计。当用户要求以下操作时触发：创建新的 Flask+Vue 项目、搭建 Web 应用、构建全栈 Python+Vue 应用、初始化 Flask Web 项目、设置单端口 Flask 前后端项目。触发词：新建项目、创建项目、搭建项目、Flask项目、Vue项目。

下载

用 batch_size 参数控制单次插入条数，避免 MySQL 的 max_allowed_packet 限制
CSV 中空字符串要显式转为 None 或默认值，否则可能违反 NOT NULL 约束
如果表有唯一索引且数据可能重复，加 ignore_conflicts=True（仅 PostgreSQL）或改用 on_conflict_do_nothing（Django 4.2+）

遇到 `IntegrityError: duplicate key violates unique constraint` 怎么办

这说明你试图插入重复主键或唯一字段（比如 email 已存在），而没做预处理。Django 默认不帮你跳过或更新。

方案一（推荐）：导入前用 MyModel.objects.filter(email__in=email_list).values_list('email', flat=True) 找出已存在值，过滤掉重复项
方案二：用 Django 4.2+ 的 on_conflict（仅 PostgreSQL）：
MyModel.objects.bulk_create(objs, ignore_conflicts=False, update_conflicts=True, update_fields=['updated_at'])
方案三：改用 get_or_create() 循环 —— 别这么做，性能断崖式下跌，只适合几千条以内

注意：MySQL 用户基本只能选方案一，因为它的 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE 在 Django ORM 层不被 bulk_create 支持。