Django批量查询优化的多种实现方案

目录

• 一、基础批量查询方法
• 1. 批量获取对象列表
• 2. 按条件批量筛选
• 3. 批量值查询（减少内存占用）
• 二、高级批量查询技术
• 1. 批量关联查询优化
• 2. 批量ID查询
• 3. 批量分页查询
• 三、批量操作API
• 1. 批量创建
• 2. 批量更新
• 3. 批量删除
• 四、性能优化技巧
• 1. 使用only()和defer()减少字段加载
• 2. 批量处理关联对象
• 3. 使用原生SQL进行复杂批量操作
• 五、实战场景示例
• 1. 批量更新电影评分
• 2. 批量导入电影数据
• 3. 批量处理电影标签
• 六、性能对比
• 七、最佳实践建议
• 总结

# 获取所有电影对象（避免内存溢出）
films = Film.objects.all()

# 使用迭代器处理大规模数据 
for film in films.iterator(chunk_size=1000):
    # 处理每个电影对象
    print(film.title)

# 获取2023年上映的电影
films_2023 = Film.objects.filter(year=2023)

# 获取评分大于4.0的电影
high_rated_films = Film.objects.filter(rating__gt=4.0)

# 只获取需要的字段
film_titles = Film.objects.values_list('title', flat=True)

# 获取特定字段组合
film_data = Film.objects.values('id', 'title', 'year')

# 使用select_related优化外键查询
films = Film.objects.select_related('studio').all()

# 使用prefetch_related优化多对多关系
films = Film.objects.prefetch_related('actors').all()

# 获取特定ID集合的电影
film_ids = [101, 205, 307, 409]
films = Film.objects.in_bulk(film_ids, field_name='id')

# 使用结果
film_205 = films[205]

from django.core.paginator import Paginator

# 每页100条记录
paginator = Paginator(Film.objects.all(), 100)

for page_num in range(1, paginator.num_pages + 1):
    page = paginator.page(page_num)
    for film in page.object_list:
        # 处理每部电影
        process_film(film)

# 创建多个电影对象
films_to_create = [
    Film(title="电影A", year=2023),
    Film(title="电影B", year=2022),
    Film(title="电影C", year=2021)
]

# 批量创建（返回创建的对象列表）
created_films = Film.objects.bulk_create(films_to_create)

# 批量更新评分
films_to_update = Film.objects.filter(year=2023)
for film in films_to_update:
    film.rating = 4.5  # 更新评分

# 批量更新到数据库
Film.objects.bulk_update(films_to_update, ['rating'])

# 删除所有2020年之前的电影
Film.objects.filter(year__lt=2020).delete()

# 只加载必要字段
films = Film.objects.only('title', 'year')

# 排除大字段
films = Film.objects.defer('plot')

# 批量添加演员
film = Film.objects.get(id=101)
actors_to_add = Actor.objects.filter(id__in=[1, 2, 3])
film.actors.add(*actors_to_add)

# 批量移除演员
film.actors.remove(*actors_to_add)

from django.db import connection

def bulk_update_ratings():
    with connection.cursor() as cursor:
        cursor.execute("""
            UPDATE film_information
            SET rating = rating * 1.1
            WHERE year >= 2020
        """)

def update_film_ratings():
    # 获取需要更新的电影
    films = Film.objects.filter(year__gte=2020)
    
    # 批量更新评分
    for film in films:
        film.rating = calculate_new_rating(film)
    
    # 批量提交到数据库
    Film.objects.bulk_update(films, ['rating'], batch_size=500)

def import_films_from_csv(csv_path):
    import csv
    films_to_create = []
    
    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            films_to_create.append(Film(
                title=row['title'],
                year=int(row['year']),
                # ...其他字段
            ))
    
    # 分批次创建（每批1000条）
    batch_size = 1000
    for i in range(0, len(films_to_create), batch_size):
        batch = films_to_create[i:i+batch_size]
        Film.objects.bulk_create(batch)

def process_film_tags():
    # 获取所有电影及其标签
    films = Film.objects.prefetch_related('tags')
    
    for film in films:
        # 处理每个电影的标签
        tags_to_add = calculate_new_tags(film)
        film.tags.add(*tags_to_add)
        
        # 移除不需要的标签
        tags_to_remove = get_obsolete_tags(film)
        film.tags.remove(*tags_to_remove)

方法	10,000条记录耗时	内存占用	适用场景
普通循环	15.2s	高	小数据集
bulk_create	1.8s	中	批量创建
bulk_update	2.1s	中	批量更新
原生SQL	0.3s	低	超大规模数据
分页处理	8.5s	低	内存敏感场景

选择合适的批量大小：

• 推荐 batch_size=500-1000
• 根据数据库性能调整

事务管理：

from django.db import transaction

with transaction.atomic():
    # 批量操作代码
    Film.objects.bulk_create(films)

错误处理：

try:
    Film.objects.bulk_update(films, ['rating'])
except Exception as e:
    logger.error(f"批量更新失败: {str(e)}")
    # 回退或重试逻辑

性能监控：

from django.db import connection

print(connection.queries)  # 查看执行的SQL查询

Django 提供了多种批量查询和处理方法：

• 查询优化：select_related, prefetch_related, values()
• 批量操作：bulk_create, bulk_update, in_bulk
• 内存管理：iterator(), Paginator
• 高级技巧：原生SQL、事务管理

通过合理使用这些技术，可以将处理10万条记录的时间从分钟级降低到秒级，同时显著减少内存占用。关键是根据具体场景选择最合适的批量处理方法，并注意错误处理和性能监控。

以上就是Django批量查询优化的多种实现方案的详细内容，更多关于Django批量查询优化的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Django中QuerySet查询优化之prefetch_related详解
• Django ORM 事务和查询优化的操作方法
• 详解Django中views数据查询使用locals()函数进行优化
• Django查询优化及ajax编码格式原理解析
• 详解Django的model查询操作与查询性能优化