如何高效加载Django模型对应DB表的1000万条数据并处理回写？

处理千万级数据时，核心要避免一次性加载全量数据导致内存溢出，同时减少数据库IO开销，以下是落地性强的解决方案：

一、优先选择数据库层面直接更新（最高效） #

如果你的处理逻辑可以用数据库表达式或SQL实现，完全不需要将数据加载到Python内存，这是最优解。

示例：用Django内置表达式更新 #

比如要给score字段统一加10，直接用F()表达式：

from django.db.models import F
from myapp.models import TargetModel

TargetModel.objects.update(score=F('score') + 10)

示例：复杂逻辑用原生SQL #

如果逻辑无法用Django ORM表达，直接执行原生SQL：

from django.db import connection

with connection.cursor() as cursor:
    cursor.execute("""
        UPDATE myapp_targetmodel 
        SET processed_content = UPPER(original_content) 
        WHERE id > 0
    """)

二、必须加载数据到Python处理时的分批方案 #

如果处理逻辑依赖Python函数（如复杂字符串处理、外部API调用），采用按主键分批查询+批量更新的模式，避免OFFSET导致的性能衰减。

1. 基础分批处理流程 #

from myapp.models import TargetModel
from django.db import transaction

def process_single_obj(obj):
    # 自定义处理逻辑，比如修改字段值
    obj.processed_field = obj.original_field.strip().upper()
    return obj

batch_size = 1000  # 根据服务器内存调整，建议500-2000
last_id = 0

while True:
    # 用主键范围查询，利用主键索引快速定位，避免OFFSET性能问题
    with transaction.atomic():
        batch = list(TargetModel.objects.filter(id__gt=last_id).order_by('id')[:batch_size])
        if not batch:
            break
        
        # 批量处理对象
        processed_batch = [process_single_obj(obj) for obj in batch]
        
        # 批量更新，仅指定需要修改的字段
        TargetModel.objects.bulk_update(processed_batch, ['processed_field'])
    
    last_id = batch[-1].id
    print(f"已处理到ID: {last_id}")

2. 优化点说明 #

• 主键分批的优势：替代TargetModel.objects.all()[i:i+batch_size]的切片方式，避免数据库扫描前N行的性能损耗，尤其是数据量越大，效果越明显。
• 事务包裹批次：每一批更新放在一个事务中，减少事务提交的IO开销，同时保证批次内数据的一致性。
• 手动控制内存：每批次处理完后，Python会自动回收该批次对象的内存，也可手动del batch+import gc; gc.collect()加速回收。

三、避免踩坑的关键细节 #

• 禁用Model.objects.all()：千万级数据下，此方法会将所有对象加载到内存直接导致OOM（内存溢出）。
• 慎用save()方法：单个对象调用save()会触发一次SQL更新，千万级数据下会产生百万级SQL请求，性能极差，必须用bulk_update。
• 预加载关联数据：如果处理时需要访问关联模型，用select_related（一对一/外键）或prefetch_related（多对多）提前加载，避免N+1查询问题：

  batch = list(TargetModel.objects.filter(id__gt=last_id)
               .order_by('id')
               .select_related('user')  # 预加载关联的User模型
               [:batch_size])
  

• 跳过信号触发：bulk_update不会触发模型的pre_save/post_save信号，如果业务依赖这些信号，需手动处理或权衡性能。

四、极端场景下的并行处理 #

如果单进程处理速度无法满足需求，可采用多进程并行处理，但需注意数据库连接的隔离性：

from multiprocessing import Pool
from myapp.models import TargetModel

def process_batch(batch_ids):
    # 每个进程独立初始化数据库连接
    batch = list(TargetModel.objects.filter(id__in=batch_ids))
    for obj in batch:
        obj.processed_field = obj.original_field.strip().upper()
    TargetModel.objects.bulk_update(batch, ['processed_field'])

# 先拆分所有主键为多个批次
all_ids = list(TargetModel.objects.values_list('id', flat=True))
id_batches = [all_ids[i:i+1000] for i in range(0, len(all_ids), 1000)]

# 启动4个进程并行处理（根据CPU核心数调整）
with Pool(processes=4) as pool:
    pool.map(process_batch, id_batches)

注意：并行处理会增加数据库负载，需根据数据库的承载能力调整进程数，避免压垮数据库。

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