如何分批更新Django中大量DeficienciesAct关联记录？

Django模型定义 #

class Partner(models.Model):
    sap_code = models.CharField(max_length=64, null=True, blank=True, verbose_name='sap id', default=uuid4)
    # 其他字段


class DeficienciesAct(models.Model):
    partner = models.ForeignKey('partners.Partner', null=True, on_delete=models.CASCADE)
    sap_code = models.CharField(max_length=64, null=True)
    # 其他字段

现有数据 #

Partner表 #

DeficienciesAct表 #

DeficienciesAct表共有50万条记录，其中部分记录的sap_code在Partner表中不存在。需要通过sap_code匹配，将DeficienciesAct的partner_id更新为对应Partner记录的ID。

已编写的更新脚本：

DeficienciesAct.objects.filter(
        partner__isnull=True, sap_code__in=Subquery(Partner.objects.values_list("sap_code"))
    ).annotate(
        new_partner_id=Subquery(
            Partner.objects.filter(
                sap_code=OuterRef('sap_code')
            ).values('id')[:1]
        )
    ).update(partner_id=F("new_partner_id"))

该脚本可正常运行，但担心处理大量记录会影响PostgreSQL性能，希望实现分批更新的方案，且不修改模型/表结构。

方案1：按ID范围分段处理 #

通过锁定ID区间分批更新，避免一次性占用大量数据库资源：

from django.db.models import F, Subquery, OuterRef

# 每批处理的记录数，可根据数据库性能调整（建议1000-5000）
BATCH_SIZE = 2000

# 获取需要更新的记录ID边界
target_query = DeficienciesAct.objects.filter(
    partner__isnull=True, sap_code__in=Subquery(Partner.objects.values_list("sap_code"))
)
id_bound = target_query.aggregate(min_id=models.Min('id'), max_id=models.Max('id'))
min_id, max_id = id_bound['min_id'], id_bound['max_id']

if not min_id or not max_id:
    print("无需要更新的记录")
else:
    current_id = min_id
    while current_id <= max_id:
        end_id = current_id + BATCH_SIZE - 1
        # 处理当前ID区间的记录
        DeficienciesAct.objects.filter(
            id__gte=current_id, id__lte=end_id,
            partner__isnull=True, sap_code__in=Subquery(Partner.objects.values_list("sap_code"))
        ).annotate(
            new_partner_id=Subquery(
                Partner.objects.filter(sap_code=OuterRef('sap_code')).values('id')[:1]
            )
        ).update(partner_id=F("new_partner_id"))
        
        print(f"完成ID范围 {current_id}-{end_id} 的更新")
        current_id = end_id + 1

方案2：迭代器分批获取ID再更新 #

利用Django的iterator分批获取目标记录ID，减少内存占用：

from django.db.models import F, Subquery, OuterRef
import itertools

BATCH_SIZE = 2000

# 分批获取需要更新的记录ID
target_ids = DeficienciesAct.objects.filter(
    partner__isnull=True, sap_code__in=Subquery(Partner.objects.values_list("sap_code"))
).values_list('id', flat=True).iterator(chunk_size=BATCH_SIZE)

# 按批次处理
for batch_ids in iter(lambda: list(itertools.islice(target_ids, BATCH_SIZE)), []):
    DeficienciesAct.objects.filter(id__in=batch_ids).annotate(
        new_partner_id=Subquery(
            Partner.objects.filter(sap_code=OuterRef('sap_code')).values('id')[:1]
        )
    ).update(partner_id=F("new_partner_id"))
    print(f"完成 {len(batch_ids)} 条记录的更新")

方案3：Raw SQL分批更新（PostgreSQL专属） #

直接使用原生SQL优化批量更新逻辑，性能更可控：

from django.db import connection

BATCH_SIZE = 2000
offset = 0

while True:
    with connection.cursor() as cursor:
        cursor.execute("""
            UPDATE deficiencies_act da
            SET partner_id = (
                SELECT p.id FROM partner p WHERE p.sap_code = da.sap_code LIMIT 1
            )
            WHERE da.partner_id IS NULL
              AND da.sap_code IN (SELECT sap_code FROM partner)
              LIMIT %s OFFSET %s
        """, [BATCH_SIZE, offset])
        affected_rows = cursor.rowcount
    
    if affected_rows == 0:
        break
    
    print(f"完成 {affected_rows} 条记录的更新")
    offset += BATCH_SIZE

注意事项 #

• 调整BATCH_SIZE：根据服务器和数据库负载测试合适值，避免过大导致锁表或性能骤降。
• 低峰期执行：尽量选择业务流量低的时段运行脚本，减少对线上服务的影响。
• 原子性可选：如果需要保证每批更新的原子性，可以给每批操作添加事务，但会增加少量性能开销。

内容的提问来源于stack exchange，提问作者Альберт Александров