Django中CSV文件批量上传最佳实践及性能优化方法

• 循环内重复查询数据库：每处理1行CSV就执行2次数据库查询（FoodItem.objects.get + RecipeItem.objects.filter），数据量越大查询开销越高
• 单条插入数据：每创建1条POSData、1条POSIngredientData就执行1次INSERT请求，按单款食品10种食材计算，每1000行CSV就会产生11000次数据库请求，网络IO开销占了99%以上的耗时
• 关于Model.create()和Model.save()的误区：两者性能没有本质差异，create()底层确实是调用save()实现，该说法没有依据，核心问题是单条操作的重复IO开销，和调用哪个方法无关。

核心思路是减少数据库交互次数，把多次查询、多次插入合并为批量操作，优化后单条CSV行处理耗时可降低到毫秒级。

优化步骤 #

1. 预加载所有关联数据到内存
   先提取CSV中所有item_key，一次性查询所有关联的FoodItem和对应的RecipeItem，存储为字典方便快速索引，避免循环查询。
2. 批量攒对象再统一插入
   循环处理CSV行时只生成模型实例存入列表，全部处理完成后调用bulk_create批量插入，大幅降低INSERT请求次数。

优化后代码示例 #

import csv
from django.db import transaction

# 视图内的处理逻辑
@transaction.atomic # 加事务保证所有数据要么全部成功要么全部回滚，还能进一步提升插入性能
def upload_pos_csv(request):
    file = request.FILES['order_file']
    file_content = file.read().decode('utf-8')
    decoded_file = file_content.splitlines()
    csvDictReader = csv.DictReader(decoded_file, delimiter=',')
    
    # 第一步：收集所有item_key，预加载关联数据
    all_item_keys = [int(row['item_key']) for row in csvDictReader]
    # 重置csv读取器
    csvDictReader = csv.DictReader(file_content.splitlines(), delimiter=',')
    
    # 预加载FoodItem存为字典，key是item_key
    food_item_map = {
        item.item_key: item 
        for item in FoodItem.objects.filter(item_key__in=all_item_keys)
    }
    
    # 预加载所有相关RecipeItem，按food_item_id分组，预关联Ingredient避免后续查询
    recipe_item_map = {}
    for recipe in RecipeItem.objects.filter(food_item__item_key__in=all_item_keys).select_related('ingredient'):
        if recipe.food_item_id not in recipe_item_map:
            recipe_item_map[recipe.food_item_id] = []
        recipe_item_map[recipe.food_item_id].append(recipe)
    
    # 第二步：攒批量插入的对象
    pos_data_list = []
    pos_ingredient_list = []
    
    for obj in csvDictReader:
        item_key = int(obj['item_key'])
        if item_key not in food_item_map:
            # 此处补全不存在的FoodItem异常处理逻辑
            continue
        food_item = food_item_map[item_key]
        # 生成POSData实例，暂不插入数据库
        pos_data = POSData(
            food_item=food_item,
            outlet_name=obj['outlet_name'],
            order_date=obj['order_date'],
            # 补全其余需要赋值的字段
        )
        pos_data_list.append(pos_data)
        
        # 生成对应的POSIngredientData实例
        recipes = recipe_item_map.get(food_item.id, [])
        for recipeItem in recipes:
            pos_ingredient = POSIngredientData(
                ingredient=recipeItem.ingredient,
                outlet_name=obj['outlet_name'],
                order_date=obj['order_date'],
                quantity=recipeItem.quantity,
                # 补全其余需要赋值的字段
            )
            pos_ingredient_list.append(pos_ingredient)
    
    # 第三步：批量插入，batch_size根据数据量调整，一般1000-5000为宜
    POSData.objects.bulk_create(pos_data_list, batch_size=1000)
    POSIngredientData.objects.bulk_create(pos_ingredient_list, batch_size=1000)
    
    # 后续业务逻辑

额外优化建议 #

• 如果CSV行数超过10万，可以不用一次性读取整个文件到内存，改用流式读取分批处理，避免服务端内存占用过高
• 如果性能要求极高，可以直接使用数据库原生的批量导入方法（比如PostgreSQL的COPY命令，MySQL的LOAD DATA命令），性能比bulk_create还要高3-10倍。

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