Django批量创建按条件更新：CSV导入大数据库优化方案咨询

针对大型数据库下的CSV批量更新需求，核心思路是仅查询CSV涉及的external_id关联数据，避免全表加载占用内存，同时结合Django批量API完成操作，具体步骤如下：

1. 提取CSV中的external_id，批量查询关联数据 #

从已校验/清理后的CSV数据中收集所有external_id，仅查询数据库中匹配这些ID的记录，大幅降低内存开销：

# cleaned_data是处理后的CSV数据列表（每个元素为符合MyModel的字典）
csv_external_ids = {item['external_id'] for item in cleaned_data}

# 批量查询数据库中匹配的记录，仅获取必要字段
existing_records = MyModel.objects.filter(
    external_id__in=csv_external_ids
).values('id', 'external_id', 'valid_from')

# 转换为字典，方便快速查找：key=external_id，value=包含主键和日期的字典
existing_dict = {
    rec['external_id']: {
        'id': rec['id'],
        'valid_from': rec['valid_from']
    } for rec in existing_records
}

2. 区分需要创建和更新的数据 #

遍历CSV数据，按条件分类处理：

to_create = []
to_update = []

for data in cleaned_data:
    ext_id = data['external_id']
    new_valid_from = data['valid_from']
    
    if ext_id not in existing_dict:
        # 该external_id不存在，加入创建列表
        to_create.append(MyModel(**data))
    else:
        existing_info = existing_dict[ext_id]
        # 仅当新日期更晚时，加入更新列表
        if new_valid_from > existing_info['valid_from']:
            # 构造带主键的实例，用于后续批量更新
            update_instance = MyModel(id=existing_info['id'], **data)
            to_update.append(update_instance)

3. 执行批量操作 #

用Django的批量API完成创建和更新，避免逐行操作的性能损耗：

# 批量创建新记录
if to_create:
    MyModel.objects.bulk_create(to_create)

# 批量更新符合条件的记录，指定需要更新的字段（按需调整）
if to_update:
    MyModel.objects.bulk_update(to_update, ['valid_from', '其他需要更新的字段'])

额外优化建议 #

• 分批次处理CSV：若CSV数据量极大（如百万级），可拆分为每1000行一批次循环处理，进一步降低内存压力。
• 添加数据库索引：给MyModel的external_id字段添加索引，加速external_id__in的查询速度：

  class MyModel(models.Model):
      external_id = models.CharField(max_length=255, db_index=True)
      valid_from = models.DateField()
      # ...其他字段
  

• 用values()替代实例查询：查询时返回字典而非模型实例，减少内存占用。

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