在线 -
搜索 AIGC编程网
← 返回题目列表

Django 的 bulk_create / bulk_update 有哪些坑？大批量写入怎么做？

中等 第 23 / 27 题 更新于 2026/08/03
Djangobulk_create批量写入性能优化

简化版

bulk_create 和 bulk_update 把「N 条 INSERT/UPDATE」压成「少数几条 SQL」，是 Django 批量写入的首选——插 1 万条数据从「1 万次往返」变成「几十次往返」，通常快 10~100 倍。但它们是用「跳过 ORM 的高层逻辑」换来的速度，代价必须清楚：① 不调用 Model.save()（你重写的 save() 逻辑全部失效）；② 不发 pre_save/post_save 信号；③ 不做 full_clean() 校验（只有数据库约束能拦住脏数据）；④ 不更新 auto_now 字段（bulk_update 里 updated_at 不会自动变）。此外还有两个高频陷阱：bulk_create 在某些数据库上不回填主键——PostgreSQL 可以（用 RETURNING），MySQL 只在不设 ignore_conflicts 时才行，SQLite 3.35+ 才支持；以及**batch_size 必须设**——一次插 100 万条会生成一条巨长的 SQL，撑爆内存或超过数据库的包大小限制（MySQL 的 max_allowed_packet），实践上 batch_size=500~2000 是安全区间。冲突处理用 ignore_conflicts=True（忽略重复）或 update_conflicts=True + unique_fields + update_fields（4.1+ 的 upsert）。**更大的量级（百万行以上）**应该跳过 ORM：PostgreSQL 用 COPY（psycopg 的 copy_from）、MySQL 用 LOAD DATA INFILE，能再快一个数量级。核心记忆：bulk 系列快在「少往返」，代价是「不走 save、不发信号、不校验」；永远设 batch_size；主键回填看数据库。

详细版

批量写入方式对比（插入 10 万行的典型耗时）：

方式SQL 条数耗时（量级）走 save/信号适用
循环 obj.save()100000~180s✅需要业务逻辑、量小
循环 + atomic 包住100000~40s✅量小但要提速
bulk_create(batch_size=1000)100~2s❌常规批量（首选）
COPY / LOAD DATA1~0.5s❌百万级导入
# ① ★bulk_create 基本用法★
objs = [Article(title=f"t{i}", author_id=1) for i in range(100_000)]
Article.objects.bulk_create(objs, batch_size=1000)      # ★必须设 batch_size★

# ② ★主键回填（★看数据库★）★
created = Article.objects.bulk_create(objs, batch_size=1000)
created[0].pk        # PostgreSQL ✓ / MySQL ✓（不带 ignore_conflicts）
                     # ★SQLite 需 3.35+★；★带 ignore_conflicts 时 MySQL ✗★

# ③ ★冲突处理★
Article.objects.bulk_create(objs, ignore_conflicts=True)   # ★重复的跳过★
# ★upsert（4.1+）★
Article.objects.bulk_create(
    objs,
    update_conflicts=True,
    unique_fields=["slug"],            # ★冲突判定依据（PG 必填）★
    update_fields=["title", "views"],  # ★冲突时更新哪些字段★
)

# ④ ★bulk_update★
articles = list(Article.objects.filter(status="draft"))
for a in articles:
    a.status = "published"
    a.published_at = timezone.now()    # ★auto_now 字段不会自动更新，要手动赋值★
Article.objects.bulk_update(articles, ["status", "published_at"], batch_size=500)
# ★生成的是 UPDATE ... SET x = CASE WHEN id=1 THEN .. WHEN id=2 THEN .. END★
# ★注意：所有对象必须已有主键；不能更新主键★

# ⑤ ★不需要读出对象时，update() 更快★
Article.objects.filter(status="draft").update(status="published")
# ★一条 SQL，不读数据到内存★ —— ★能用 update() 就别用 bulk_update()★

# ⑥ ★F 表达式做原子增量★
Article.objects.filter(pk__in=ids).update(views=F("views") + 1)   # ★无竞态★

# ⑦ ★多对多批量★
article.tags.add(*tag_objs)                    # ★一条 SQL（内部就是 bulk_create）★
ArticleTag.objects.bulk_create([...])          # 有额外字段的中间表

# ⑧ ★批量删除★
Article.objects.filter(created__lt=cutoff).delete()
# ★注意：会先 SELECT 出所有 pk（为了级联和信号）★
# ★超大量时分批：★
while Article.objects.filter(created__lt=cutoff).exists():
    ids = Article.objects.filter(created__lt=cutoff).values_list("pk", flat=True)[:10000]
    Article.objects.filter(pk__in=list(ids)).delete()

# ⑨ ★读取大表：iterator() 避免内存爆炸★
for a in Article.objects.iterator(chunk_size=2000):    # ★不缓存整个结果集★
    process(a)

⚠️ 三个必须记住的点：① bulk_create/bulk_update 跳过了 ORM 的四层逻辑——不调用你重写的 Model.save()、不发 pre_save/post_save 信号、不执行 full_clean() 校验、bulk_update 不更新 auto_now 字段。这意味着：如果你在 save() 里生成 slug、在 post_save 里同步搜索索引、靠 auto_now 记录更新时间，这些统统不会发生，而且静默无提示。所以要么在批量操作前后显式补上这些逻辑，要么把校验下沉到数据库约束（CheckConstraint/UniqueConstraint）——数据库约束是批量写入下唯一还生效的防线。② batch_size 必须显式设置。不设的话 Django 会尝试把所有对象放进一条 SQL，10 万行会生成一条几十 MB 的语句——轻则占用大量内存，重则超过 MySQL 的 max_allowed_packet（默认 4~64MB）直接报错，或者形成一个持续几十秒的巨型事务把表锁死。经验值 500~2000：太小则往返次数多，太大则单条语句过长。③ 能用 queryset.update() 就不要用 bulk_update()。update() 是一条 SQL 直接在数据库里改，完全不需要把对象读进 Python；而 bulk_update() 需要你先把对象查出来、在内存里改、再写回去，还会生成很长的 CASE WHEN 语句。只有当「新值需要按每行不同地计算」时才用 bulk_update。

完整版教学

一、为什么循环 save() 那么慢

★ 一次 obj.save() 到底做了多少事：
  ┌────────────────────────────────────────────────┐
  │ ① 触发 pre_save 信号                            │
  │ ② 处理 auto_now / auto_now_add                  │
  │ ③ 判断是 INSERT 还是 UPDATE（可能先 SELECT）     │
  │ ④ ★发一条 SQL 到数据库★                          │
  │ ⑤ ★等待数据库响应（网络往返 RTT）★               │
  │ ⑥ 回填主键                                      │
  │ ⑦ 触发 post_save 信号                           │
  └────────────────────────────────────────────────┘

★ 瓶颈在 ④⑤：★网络往返★
  假设应用和数据库之间 RTT = 1ms（同机房算快的）
  插 100000 条 → ★100000 × 1ms = 100 秒★（★纯等待★）
  + 每次都是独立事务（autocommit）→ ★每次都要 fsync 刷盘★

★ 三级提速（★理解这个阶梯很重要★）：
  ┌─────────────────────────────────────────────────────────┐
  │ 级别 0：裸循环 save()                                     │
  │   100000 次往返 + ★100000 次事务提交★  →  ~180s           │
  ├─────────────────────────────────────────────────────────┤
  │ 级别 1：用 atomic 包起来                                  │
  │   with transaction.atomic():                             │
  │       for o in objs: o.save()                            │
  │   100000 次往返 + ★1 次提交★  →  ~40s                     │
  │   ★提升来自"少了 10 万次 fsync"★                          │
  ├─────────────────────────────────────────────────────────┤
  │ 级别 2：bulk_create(batch_size=1000)                      │
  │   ★100 次往返★ + 1 次提交  →  ~2s                         │
  │   INSERT INTO t (a,b) VALUES (..),(..),(..) ... ★×1000★  │
  ├─────────────────────────────────────────────────────────┤
  │ 级别 3：COPY / LOAD DATA INFILE                           │
  │   ★1 次流式传输★，绕过 SQL 解析  →  ~0.5s                  │
  └─────────────────────────────────────────────────────────┘

★ 关键洞察：
  ★瓶颈从来不是"数据库写得慢"，而是"往返次数 × RTT"★
  → 所有批量优化的本质都是 ★减少往返★
  → 这也解释了为什么"数据库在本机时"提升没那么夸张（RTT ≈ 0）

★ 一个常被忽略的中间选择：
  ★如果你必须走 save()（有业务逻辑），至少包一层 atomic★
  → 零代码改动、零风险、★4 倍以上提速★

理解瓶颈才知道该优化什么：一次 obj.save() 的耗时大头不是数据库写入，而是网络往返 + 事务提交的 fsync。假设 RTT = 1ms，插 10 万条就是 10 万次往返 = 100 秒纯等待，再加上 autocommit 模式下每次都要刷盘。所以提速有个清晰的三级阶梯：级别 1 是用 atomic 包住循环——不改任何逻辑、零风险，只是把 10 万次事务提交变成 1 次，通常能快 4 倍以上，这是很多人忽略的中间选择；级别 2 是 bulk_create，把往返次数从 10 万降到 100；级别 3 是 COPY/LOAD DATA，绕过 SQL 解析直接流式传输。关键洞察是「瓶颈从来不是数据库写得慢，而是往返次数 × RTT」——所有批量优化的本质都是减少往返。

二、bulk_create 的细节与陷阱

★ 生成的 SQL：
  Article.objects.bulk_create(objs, batch_size=3)
  → INSERT INTO article (title, author_id) VALUES
      ('a', 1), ('b', 1), ('c', 1);          # ★一条 SQL 插 3 行★
    INSERT INTO article ... VALUES (...), (...), (...);
    ...

★ ★陷阱 1：主键回填因数据库而异★
  ┌──────────────┬────────────────────────────────────────┐
  │ PostgreSQL   │ ✅ ★总是回填★（用 RETURNING id）        │
  │ MySQL        │ ✅ 回填，★但 ignore_conflicts=True 时 ✗★ │
  │ SQLite       │ ✅ ★3.35+★（Django 4.0+ 支持）          │
  │ Oracle       │ ❌ ★不回填★                             │
  └──────────────┴────────────────────────────────────────┘
  ★ 依赖回填的代码要注意可移植性
  ✓ 稳妥做法：插入后按业务唯一键重新查出来
    Article.objects.filter(slug__in=slugs).in_bulk(field_name="slug")

★ ★陷阱 2：不设 batch_size 的后果★
  bulk_create(100000 个对象)   # ★不设★
  → Django 会尝试放进★尽可能少的 SQL★
  → 生成一条★几十 MB 的 INSERT★
  ✗ MySQL: "Got a packet bigger than 'max_allowed_packet' bytes"
  ✗ 内存暴涨（SQL 字符串 + 参数列表）
  ✗ ★巨型事务长时间持锁★，阻塞其他写入
  ✓ batch_size=500~2000
  ★ 补充：Django 对某些数据库有内部上限（如 SQLite 的变量数限制），
    但★别依赖它★，显式设置最保险

★ ★陷阱 3：不走 save() 意味着什么★
  class Article(models.Model):
      def save(self, *a, **kw):
          if not self.slug:
              self.slug = slugify(self.title)      # ★bulk_create 时不执行！★
          super().save(*a, **kw)
  → 插进去的数据 slug 全是空
  ✓ 显式处理：
    for o in objs:
        o.slug = slugify(o.title)                  # ★手动补★
    Article.objects.bulk_create(objs, batch_size=1000)

★ ★陷阱 4：不发信号★
  @receiver(post_save, sender=Article)
  def sync_index(sender, instance, **kw): ...      # ★bulk_create 不触发★
  ✓ 批量结束后统一处理：
    Article.objects.bulk_create(objs, batch_size=1000)
    reindex_bulk([o.pk for o in objs])             # ★批量重建索引（还更高效）★

★ ★陷阱 5：不做校验★
  bulk_create 不调用 full_clean() → ★max_length 超长、choices 非法值都能进库★
  （★注意：Model.save() 本来也不调用 full_clean()★，只是循环 save 时
    你可能已经在别处校验过了）
  ✓ ★把关键规则做成数据库约束★：
    CheckConstraint(check=Q(amount__gte=0), name="amount_positive")
    → ★这是批量写入下唯一还生效的防线★

★ ★陷阱 6：多表继承（MTI）不支持★
  bulk_create ★不能用于有多表继承的模型★（父子两张表要分别插入）
  → 抛 ValueError
  ✓ 改用抽象基类，或退回循环 save()

★ 冲突处理（★4.1+ 的 upsert★）：
  Article.objects.bulk_create(
      objs,
      update_conflicts=True,
      unique_fields=["slug"],              # ★PG 必填；对应 ON CONFLICT (slug)★
      update_fields=["title", "views"],    # ★DO UPDATE SET title=.., views=..★
  )
  → PG: INSERT ... ON CONFLICT (slug) DO UPDATE SET ...
  → MySQL: INSERT ... ON DUPLICATE KEY UPDATE ...
  ★ 注意：★update_fields 里不能包含 unique_fields★
  ★ 注意：★update_conflicts 时不回填主键★（部分后端）

bulk_create 有六个必须知道的陷阱。主键回填因数据库而异——PostgreSQL 总是回填（用 RETURNING）、MySQL 在带 ignore_conflicts 时不回填、SQLite 需要 3.35+、Oracle 不支持；稳妥做法是插入后按业务唯一键用 in_bulk 重新查。不设 batch_size 会生成几十 MB 的单条 SQL，撑爆 max_allowed_packet 或形成长时间持锁的巨型事务。不走 save()、不发信号、不做校验这三条是同一个本质——它绕过了 ORM 的高层逻辑，所以你在 save() 里生成 slug、在 post_save 里同步搜索索引的代码统统静默失效；应对方式是批量前手动补上字段、批量后统一处理副作用（其实还更高效），并把关键规则下沉成数据库约束——那是批量写入下唯一还生效的防线。另外 bulk_create 不支持多表继承（MTI）的模型，会直接抛 ValueError。冲突处理方面，4.1+ 的 update_conflicts + unique_fields + update_fields 实现了真正的 upsert。

三、bulk_update 与 update() 的取舍

★ bulk_update 生成的 SQL（★看清楚就明白它的代价★）：
  Article.objects.bulk_update(articles, ["status", "views"])
  →
  UPDATE article SET
    status = CASE
        WHEN id = 1 THEN 'published'
        WHEN id = 2 THEN 'draft'
        WHEN id = 3 THEN 'published'
        ...
    END,
    views = CASE
        WHEN id = 1 THEN 100
        ...
    END
  WHERE id IN (1, 2, 3, ...);

  ★ 特点：
    - ★一条 SQL 更新多行，每行值可以不同★
    - ★SQL 长度 ≈ 行数 × 字段数★ → ★batch_size 更重要★（建议 ≤500）
    - 数据库要逐行求值 CASE → ★行数很大时 CPU 开销明显★

★ ★update() vs bulk_update()（★核心决策★）★：
  ┌──────────────────┬────────────────────────────────────────┐
  │ ★update()★        │ ★所有行改成"同一个值"或"同一个表达式"★  │
  │                   │ UPDATE t SET status='x' WHERE ...      │
  │                   │ ★不读数据到内存、SQL 短、最快★          │
  ├──────────────────┼────────────────────────────────────────┤
  │ ★bulk_update()★   │ ★每行的新值不同、且要在 Python 里算★    │
  │                   │ 需要先 SELECT 出对象                    │
  └──────────────────┴────────────────────────────────────────┘
  ★ 决策口诀：★"新值能用 SQL 表达就用 update()"★

★ update() 能表达的比你想的多：
  .update(views=F("views") + 1)                       # ★原子增量★
  .update(total=F("price") * F("qty"))                # 字段间运算
  .update(status=Case(When(views__gt=1000, then=Value("hot")),
                      default=Value("normal")))        # ★条件更新★
  .update(updated=Now())                               # 数据库时间
  → ★这些都不需要读数据到 Python★

★ bulk_update 的限制：
  ✗ ★对象必须已有主键★（未保存的会抛错）
  ✗ ★不能更新主键★
  ✗ ★不返回受影响行数★（4.0 前返回 None，4.0+ 返回行数）
  ✗ ★不更新 auto_now 字段★（★最常见的坑★）
      updated_at = models.DateTimeField(auto_now=True)
      → bulk_update 后 ★updated_at 还是旧值★
      ✓ 手动赋值：o.updated_at = timezone.now()，并加进 fields
  ✗ 不发信号、不走 save()（同 bulk_create）

★ ★并发安全：F 表达式 vs 读改写★
  ✗ 危险（★竞态★）：
    a = Article.objects.get(pk=1)
    a.views += 1                     # ★两个请求同时读到 100★
    a.save()                          # ★结果是 101 而不是 102★
  ✓ 安全：
    Article.objects.filter(pk=1).update(views=F("views") + 1)
    # → ★UPDATE ... SET views = views + 1★（数据库里原子完成）
  ★ F 表达式的本质：★把计算推到数据库，避免"读-改-写"竞态★

看清 bulk_update 生成的 SQL 就明白它的代价——它是一条巨大的 UPDATE ... SET x = CASE WHEN id=1 THEN .. WHEN id=2 THEN .. END，SQL 长度约等于「行数 × 字段数」，所以它的 batch_size 比 bulk_create 更需要压小（建议 ≤500），而且数据库要逐行求值 CASE。核心决策口诀是「新值能用 SQL 表达就用 update()」——update() 一条 SQL 直接在数据库里改、完全不读数据到内存，而且它能表达的比很多人想的多：F("views") + 1（原子增量）、F("price") * F("qty")（字段间运算）、Case/When（条件更新）、Now()。只有当每行的新值不同且必须在 Python 里算时才用 bulk_update。它的限制里最常见的坑是不更新 auto_now 字段——updated_at 会保持旧值，必须手动赋值并加进 fields 列表。最后，F 表达式的本质是把计算推到数据库，避免「读-改-写」的竞态——这是并发安全的关键。

四、更大量级：绕过 ORM

★ 什么时候该跳过 ORM：
  - ★百万行以上的一次性导入★
  - 定时的数据同步/ETL
  - 数据迁移脚本

★ PostgreSQL：★COPY（最快）★
  import io
  from django.db import connection

  buf = io.StringIO()
  for row in rows:
      buf.write(f"{row['title']}\t{row['author_id']}\n")   # ★制表符分隔★
  buf.seek(0)
  with connection.cursor() as cur:
      cur.copy_from(buf, "myapp_article",
                    columns=("title", "author_id"))         # psycopg2
      # psycopg3: cur.copy("COPY myapp_article (title, author_id) FROM STDIN")
  ★ 比 bulk_create 再快 ★3~5 倍★（绕过 SQL 解析和逐行校验）
  ★ 注意：★需要处理转义★（制表符、换行、NULL 用 \N）

★ MySQL：★LOAD DATA INFILE★
  LOAD DATA LOCAL INFILE '/tmp/data.csv'
  INTO TABLE myapp_article
  FIELDS TERMINATED BY ',' ENCLOSED BY '"'
  LINES TERMINATED BY '\n'
  (title, author_id);
  ★ 需要 local_infile=1（客户端和服务端都要开）

★ 通用：★executemany★
  with connection.cursor() as cur:
      cur.executemany(
          "INSERT INTO myapp_article (title, author_id) VALUES (%s, %s)",
          [(r["title"], r["author_id"]) for r in rows],
      )
  ★ 比 bulk_create 略快（少了 ORM 对象构造的开销）
  ★ psycopg2 的 execute_values / execute_batch 更快

★ ★导入前后的数据库层优化（★常被忽略但效果巨大★）★：
  ① ★先删索引，导完再建★
     大表上每插一行都要维护 N 个索引 → ★索引越多越慢★
     ALTER TABLE t DROP INDEX idx_x;   -- 导入 --   CREATE INDEX idx_x ON t(x);
     ★ 建索引是一次性排序，比逐行维护快得多
  ② ★暂时关闭外键检查★（MySQL）
     SET FOREIGN_KEY_CHECKS = 0;  ... ;  SET FOREIGN_KEY_CHECKS = 1;
  ③ ★调大事务但别无限大★（几万行一提交）
  ④ PG: ★导入后 ANALYZE★ 更新统计信息（否则查询计划会很差）

★ 内存控制：★读的一侧也要分批★
  ✗ objs = [Article(...) for r in read_all_rows()]     # ★1000 万对象撑爆内存★
  ✓ 分批处理：
    def chunks(iterable, size):
        it = iter(iterable)
        while batch := list(islice(it, size)):
            yield batch
    for batch in chunks(read_rows(), 5000):
        Article.objects.bulk_create([Article(**r) for r in batch])
  ★ ★读大表用 iterator(chunk_size=N)★：
    for a in Article.objects.iterator(chunk_size=2000):
        ...
    → ★不把整个结果集缓存进 QuerySet★
    → PG 上还会用★服务端游标★（server-side cursor）

★ 数量级选择表：
  ┌────────────┬──────────────────────────────────┐
  │ < 100 行    │ ★循环 save() 就行★（可读性优先）  │
  │ 100~10 万   │ ★bulk_create(batch_size=1000)★   │
  │ 10 万~百万  │ bulk_create + ★分批读 + 关索引★   │
  │ > 百万      │ ★COPY / LOAD DATA INFILE★        │
  └────────────┴──────────────────────────────────┘

百万行以上应该跳过 ORM：PostgreSQL 用 COPY（psycopg 的 copy_from/copy），它绕过 SQL 解析和逐行校验，比 bulk_create 再快 3~5 倍；MySQL 用 LOAD DATA INFILE；通用方案是 cursor.executemany（psycopg2 的 execute_values 更快）。但更容易被忽略、效果却更大的是数据库层的优化：先删索引导完再建（大表上每插一行都要维护 N 个索引，而建索引是一次性排序，快得多）、临时关闭外键检查、几万行一提交、PG 导入后跑 ANALYZE 更新统计信息。还有一点必须注意：读的一侧也要分批——[Article(**r) for r in read_all_rows()] 会把 1000 万个对象一次性装进内存；读大表要用 iterator(chunk_size=N)（不缓存整个结果集，PG 上还会用服务端游标）。

五、事务、并发与可观测性

★ 批量操作的事务边界：
  ✗ 一个巨型事务：
    with transaction.atomic():
        Article.objects.bulk_create(all_million_objs, batch_size=1000)
    → ★长时间持锁★、undo/redo 日志暴涨、
      ★失败要全部回滚（前面的功都白费）★、
      ★复制延迟★（从库要重放这个大事务）
  ✓ 分批各自成事务：
    for batch in chunks(rows, 5000):
        with transaction.atomic():
            Article.objects.bulk_create([...], batch_size=1000)
        # ★每批独立提交 → 失败只丢一批 + 可断点续传★
  ★ 取舍：★牺牲"全有或全无"，换取可恢复性和低锁竞争★
  ★ 如果业务必须原子：那就接受大事务，但要控制总量

★ ★断点续传的实现★：
  已导入的记录打标记或记录 offset
  for batch in chunks(rows[last_offset:], 5000):
      ...
      Progress.objects.update_or_create(job="import",
                                        defaults={"offset": offset})
  ★ 大批量导入★必须能中断后继续★，否则跑 3 小时挂了就要重来

★ 并发写入的注意点：
  ① ★bulk_create 与唯一约束★：
     并发两个进程插相同 slug → ★IntegrityError★
     ✓ ignore_conflicts=True 或 update_conflicts
  ② ★死锁★：
     两个批次以★不同顺序★更新相同的行 → 死锁
     ✓ ★批内按主键排序★后再操作 → 保证加锁顺序一致
  ③ ★对线上的影响★：
     大批量写会挤占 IO、污染 buffer pool、拖慢正常查询
     ✓ ★限速★：每批之间 sleep 一小会儿
     ✓ ★放在低峰期★，或走从库/独立实例

★ 可观测性（★大批量任务必备★）：
  import time, logging
  t0 = time.perf_counter()
  for i, batch in enumerate(chunks(rows, 5000)):
      Article.objects.bulk_create([...], batch_size=1000)
      if i % 20 == 0:
          logging.info("imported=%d elapsed=%.1fs rate=%.0f/s",
                       (i+1)*5000, time.perf_counter()-t0,
                       (i+1)*5000 / (time.perf_counter()-t0))
  ★ 至少要能回答：★进度多少、速率多少、预计还要多久★

★ 测试批量代码：
  with self.assertNumQueries(2):          # ★验证真的只发了 2 条 SQL★
      Article.objects.bulk_create(objs_of_1000, batch_size=500)
  ★ 这能挡住"不小心退回逐条 save"的回归

批量操作的事务边界是个真实的取舍：一个巨型事务会长时间持锁、undo 日志暴涨、失败要全部回滚（前面的功都白费）、造成主从复制延迟；分批各自成事务则牺牲「全有或全无」，换来可恢复性和低锁竞争——大批量导入通常选后者，并配合断点续传（记录 offset 或给已处理记录打标记），否则跑三小时挂了就要重来。并发方面有三点：唯一约束冲突用 ignore_conflicts/update_conflicts 处理；死锁源于两个批次以不同顺序更新相同的行，批内按主键排序就能保证加锁顺序一致；对线上的影响（挤占 IO、污染 buffer pool）要靠限速和低峰期执行来缓解。最后，大批量任务必须有可观测性——至少能回答「进度多少、速率多少、还要多久」；测试上可以用 assertNumQueries 挡住「不小心退回逐条 save」的回归。

六、实践清单

★ 写批量代码前的自检：
  □ ★设了 batch_size 吗★（bulk_create 500~2000，bulk_update ≤500）
  □ ★save() 里的逻辑手动补了吗★（slug 生成、字段派生）
  □ ★post_save 的副作用批量补了吗★（索引、缓存、通知）
  □ ★bulk_update 的 auto_now 字段手动赋值了吗★
  □ ★关键校验有数据库约束兜底吗★
  □ ★能用 update() 代替 bulk_update() 吗★
  □ ★读的一侧分批了吗★（iterator / chunks）
  □ ★事务边界合理吗★（不是一个巨型事务）
  □ ★能断点续传吗★（超过 10 分钟的任务必须能）
  □ ★有进度日志吗★
  □ ★冲突怎么处理★（ignore / update / 报错）

★ 方法速查：
  ┌────────────────────┬──────────────────────────────────┐
  │ bulk_create        │ ★批量插入★，不走 save/信号         │
  │ bulk_update        │ 批量改，每行值不同；★CASE WHEN★    │
  │ ★update()★         │ ★所有行同一表达式，最快★           │
  │ in_bulk()          │ ★按 pk/唯一键批量取 → dict★       │
  │ iterator()         │ ★流式读，不缓存结果集★             │
  │ values_list(flat)  │ 只取需要的列，★省内存★             │
  │ ★only/defer★       │ 限制加载的字段                    │
  │ count()/exists()   │ ★别用 len(qs) / if qs★           │
  └────────────────────┴──────────────────────────────────┘

★ 一个完整的导入模板：
  def import_articles(rows, batch=5000):
      total, t0 = 0, time.perf_counter()
      for chunk in chunks(rows, batch):
          objs = []
          for r in chunk:
              o = Article(**r)
              o.slug = slugify(o.title)          # ★补 save() 的逻辑★
              objs.append(o)
          with transaction.atomic():             # ★每批独立事务★
              created = Article.objects.bulk_create(
                  objs, batch_size=1000,
                  update_conflicts=True,          # ★幂等：可重复执行★
                  unique_fields=["slug"],
                  update_fields=["title"],
              )
          total += len(created)
          logging.info("progress=%d rate=%.0f/s", total,
                       total / (time.perf_counter() - t0))
      reindex_bulk(...)                          # ★补 post_save 的副作用★

★ 一句话总结：
  ★"bulk 系列的速度来自『减少往返』，代价是『跳过 ORM 高层逻辑』；
    永远设 batch_size、手动补 save/信号/auto_now 的活、
    能用 update() 就别用 bulk_update、百万级直接上 COPY。"★

自检清单里最容易漏掉的四条是：手动补 save() 里的派生字段逻辑、批量后补 post_save 的副作用、bulk_update 的 auto_now 字段手动赋值、以及超过 10 分钟的任务必须能断点续传。方法速查表里还有几个配套的读取优化：in_bulk()（按 pk 或唯一键批量取成 dict）、iterator()（流式读）、values_list(flat=True)（只取需要的列）、以及用 count()/exists() 而不是 len(qs)/if qs。那个导入模板体现了三个关键设计：每批独立事务（可断点续传）、用 update_conflicts 实现幂等（脚本可重复执行而不报错）、进度日志。

记忆钩子：「批量写入的瓶颈★从来不是数据库写得慢，而是『往返次数 × RTT』★——所以优化的本质就是★减少往返★。三级阶梯：①★只加 transaction.atomic() 包住循环★（10 万次提交变 1 次，零风险快 4 倍，最容易被忽略）②★bulk_create(batch_size=1000)★（往返从 10 万降到 100，快 10100 倍）③★COPY / LOAD DATA INFILE★（绕过 SQL 解析，再快 35 倍）。bulk 系列的速度是★用『跳过 ORM 高层逻辑』换来的★，四个代价：★不调用你重写的 save()★（slug 生成失效）、★不发 pre_save/post_save 信号★（索引/缓存不同步）、★不做 full_clean 校验★、★bulk_update 不更新 auto_now★（updated_at 保持旧值）——而且★全部静默无提示★；应对是批量前手动补字段、批量后统一处理副作用（反而更高效）、★把关键规则下沉成数据库约束（批量下唯一还生效的防线）★。★batch_size 必须显式设★（不设会生成几十 MB 单条 SQL，撑爆 MySQL 的 max_allowed_packet 或形成巨型持锁事务）：bulk_create 500~2000，★bulk_update ≤500★（因为它生成 CASE WHEN，长度≈行数×字段数）。★主键回填看数据库★：PG 总是回填（RETURNING）、MySQL 带 ignore_conflicts 时不回填、SQLite 需 3.35+、Oracle 不支持。★核心决策口诀：新值能用 SQL 表达就用 update() 而不是 bulk_update()★——update() 一条 SQL 不读数据到内存，且能表达 F()+1、F() 乘 F()、Case/When、Now()；★F 表达式还能避免『读-改-写』的竞态★。大量导入还要：★分批各自成事务★（换可恢复性和低锁竞争）+ ★断点续传★ + ★先删索引导完再建★ + 读的一侧用 ★iterator(chunk_size)★ + 进度日志。bulk_create ★不支持多表继承 MTI★。」

七、常见误区与追问

  • 误区：bulk_create 只是「快一点的循环 save」，行为完全一样。 完全不一样——它绕过了 ORM 的整个高层逻辑：不调用 Model.save()（所以你在 save() 里写的 slug 生成、字段派生、状态推导全部失效）、不发 pre_save/post_save 信号（搜索索引不更新、缓存不失效、通知不发送）、不执行 full_clean()（max_length 超长、choices 非法值都能进库）、bulk_update 还不更新 auto_now 字段。最麻烦的是这一切都是静默的——没有警告、没有报错，数据就那样进去了，往往几周后才发现「怎么有一批数据的 slug 是空的」。所以每次用 bulk 系列前都要问自己：「这个模型的 save() 和信号里有什么逻辑？我需要手动补吗？」
  • 误区：batch_size 不设也没关系，Django 会自动处理。 Django 确实会根据数据库后端做一些内部限制（比如 SQLite 的变量数上限），但不要依赖它。不设 batch_size 时 Django 会尝试把尽可能多的行放进一条 SQL，10 万行就是一条几十 MB 的 INSERT——后果有三：① MySQL 直接报 Got a packet bigger than 'max_allowed_packet' bytes（默认只有 4~64MB）；② 内存暴涨（SQL 字符串加参数列表）；③ 形成一个持续几十秒的巨型事务，长时间持锁阻塞其他写入、还会造成主从复制延迟。经验值是 bulk_create 用 500~2000、bulk_update 用 ≤500（因为它生成的 CASE WHEN 语句长度约等于「行数 × 字段数」，同样行数下 SQL 长得多）。
  • 误区：要批量修改数据，就该用 bulk_update。 大多数场景应该用 queryset.update()。区别在于：update() 是一条 SQL 直接在数据库里改（UPDATE t SET status='x' WHERE ...），完全不需要把对象读进 Python；而 bulk_update() 要求你先 SELECT 出所有对象、在内存里逐个改、再生成一条巨大的 CASE WHEN 写回去——多了一次全量读、一次内存占用、一条超长 SQL。而且 update() 能表达的比很多人以为的多：F("views") + 1（原子增量，还顺便避免了竞态）、F("price") * F("qty")（字段间运算）、Case/When（按条件给不同的值）、Now()（数据库时间）。判断口诀：「新值能用 SQL 表达出来，就用 update()」；只有当每行的新值必须在 Python 里算（比如调用了外部服务、跑了复杂算法）时，才轮到 bulk_update。
  • 误区：bulk_create 返回的对象一定带主键，可以直接拿去建关联。 取决于数据库后端。PostgreSQL 总是回填（它用 INSERT ... RETURNING id）；MySQL 会回填，但一旦加了 ignore_conflicts=True 就不回填（因为无法确定哪些行真的插入了）；SQLite 需要 3.35+（Django 4.0 起支持）；Oracle 不回填。所以「插入后立刻用 created[0].pk 建关联对象」这种写法在 PostgreSQL 上跑得好好的，换到 MySQL + ignore_conflicts 就全是 None。稳妥的做法是插入后按业务唯一键重新查一次：Article.objects.in_bulk(slugs, field_name="slug") 拿到 {slug: obj} 的映射，再去建关联——多一次查询，换来跨数据库的确定性。
  • 误区：把整个百万行导入包在一个 transaction.atomic() 里最安全，失败能全部回滚。 「全有或全无」听起来很美，但代价往往不可接受：① 长时间持锁（几分钟到几十分钟），阻塞线上正常写入；② undo/redo 日志暴涨，可能撑爆磁盘；③ 主从复制延迟（从库要重放这一个大事务，期间读从库的业务全部看到旧数据）；④ 失败就前功尽弃——跑了两小时在 90% 处报错，一切从头再来。实践中更常见的选择是分批各自成事务（每 5000 行提交一次）加上断点续传（记录已处理的 offset 或给记录打标记）——牺牲了原子性，换来可恢复、低锁竞争、可观测进度。如果业务上确实必须原子（比如财务对账），那就接受大事务，但要控制单次总量并安排在低峰期。
  • 追问：bulk_create 的 ignore_conflicts 和 update_conflicts 分别对应什么 SQL？怎么选？ ignore_conflicts=True 在 PostgreSQL 上生成 INSERT ... ON CONFLICT DO NOTHING、MySQL 上是 INSERT IGNORE——遇到唯一约束冲突就跳过那一行，不报错也不更新。适合「只要保证存在，重复的不管」的场景（比如导入标签、去重写入日志）。update_conflicts=True（Django 4.1+）需要同时给 unique_fields（冲突判定依据，PostgreSQL 必填）和 update_fields（冲突时更新哪些字段），生成 PostgreSQL 的 INSERT ... ON CONFLICT (slug) DO UPDATE SET ... 或 MySQL 的 INSERT ... ON DUPLICATE KEY UPDATE ...——这就是真正的 upsert，适合「同步外部数据源，有则更新无则插入」。两个细节：update_fields 里不能包含 unique_fields；用了 update_conflicts 后部分后端不回填主键。选择上：幂等的导入脚本用 update_conflicts（可以反复执行）、只做去重用 ignore_conflicts。
  • 追问：读取大量数据时怎么避免内存爆炸？ 三个层次。① iterator(chunk_size=N)——普通 QuerySet 会把所有结果缓存进 _result_cache（10 万行模型对象轻松几百 MB），iterator() 则不缓存，配合 chunk_size 分批从数据库取；在 PostgreSQL 上它还会使用服务端游标（数据留在数据库侧，按需传输）。代价是不能重复迭代、不能配合 prefetch_related（Django 4.1 起支持了，但需要设 chunk_size）。② 只取需要的列——values_list("id", flat=True) 或 values("id", "title") 返回元组/字典而不是模型对象，内存能省一个数量级；only()/defer() 则是返回模型对象但限制加载的字段（注意访问未加载字段会触发额外查询）。③ 分批处理源数据——如果数据来自文件或 API，用 islice 之类的工具切成块，别一次性构造出全部模型对象。另外提醒两个常见浪费：用 count() 而不是 len(qs)（后者会把全部结果读进来）、用 exists() 而不是 if qs。
  • 追问：为什么「先删索引、导入完再重建」能大幅提速？ 因为索引维护是逐行的、而建索引是一次性批量的。有索引时每插入一行，数据库都要在 B+ 树里定位插入位置、可能触发页分裂、写 WAL/redo 日志——如果表上有 5 个索引，插一行就是 1 次数据写 + 5 次索引写，而且索引页的写入是随机 IO（按索引键顺序分布，与插入顺序无关），非常慢。而先删索引、导完再 CREATE INDEX，数据库可以把整列数据读出来一次性排序、顺序批量构建 B+ 树，是顺序 IO且能充分利用内存排序缓冲区——通常快好几倍。配套的做法还有：临时关闭外键检查（MySQL 的 SET FOREIGN_KEY_CHECKS=0，省去每行的引用校验查询）、导入后跑 ANALYZE（PostgreSQL，更新统计信息，否则新数据上的查询计划会很差）。但要注意风险：删索引期间线上查询会变慢甚至不可用、唯一索引删掉后重复数据可能进库（重建时会失败），所以这套做法主要适用于初始化导入或离线表，不适合有在线读写的表。

八、加强记忆

批量写入的瓶颈从来不是「数据库写得慢」，而是「往返次数 × RTT」——所以所有优化的本质都是减少往返。三级阶梯：① 只加 transaction.atomic() 包住循环（10 万次事务提交变 1 次，零代码风险快 4 倍以上，是最容易被忽略的中间选择）；② bulk_create(batch_size=1000)（往返从 10 万降到 100，快 10100 倍）；③ COPY / LOAD DATA INFILE（绕过 SQL 解析，再快 35 倍）。bulk 系列的速度是用「跳过 ORM 高层逻辑」换来的，四个代价：不调用你重写的 save()（slug 生成失效）、不发 pre_save/post_save 信号（索引和缓存不同步）、不做 full_clean() 校验、bulk_update 不更新 auto_now（updated_at 保持旧值）——而且全部静默无提示；应对方式是批量前手动补字段、批量后统一处理副作用（反而更高效）、并把关键规则下沉成数据库约束，那是批量写入下唯一还生效的防线。batch_size 必须显式设置（不设会生成几十 MB 的单条 SQL，撑爆 MySQL 的 max_allowed_packet 或形成长时间持锁的巨型事务）：bulk_create 用 500~2000，bulk_update 用 ≤500（它生成 CASE WHEN，SQL 长度约等于行数 × 字段数）。主键回填看数据库：PostgreSQL 总是回填（RETURNING）、MySQL 带 ignore_conflicts 时不回填、SQLite 需 3.35+、Oracle 不支持——依赖回填的代码不可移植，稳妥做法是用 in_bulk(field_name=唯一键) 重新查。核心决策口诀：新值能用 SQL 表达就用 update() 而不是 bulk_update()——update() 一条 SQL、不读数据到内存，且能表达 F()+1、F()*F()、Case/When、Now()，F 表达式还顺便避免了「读-改-写」的竞态。大批量导入还要注意：分批各自成事务（换取可恢复性和低锁竞争）、断点续传、先删索引导完再建、读的一侧用 iterator(chunk_size)、以及进度日志。最后记住 bulk_create 不支持多表继承（MTI）的模型。