如何优化无外键多表数据库的数据查询与CSV导出效率？

针对你遇到的N+1查询导致30万条数据导出耗时3小时的问题，核心优化思路是消除重复查询+降低内存占用，以下是具体可落地的方案：

1. 彻底解决N+1查询：批量预取关联数据 #

因为数据库没有外键，无法使用Django的select_related或prefetch_related，所以我们手动批量查询所有需要的关联数据，用字典做ID到名称的映射，避免每条用户数据单独查询关联表：

步骤1：预取关联映射字典 #

在视图中先收集所有用户涉及的关联ID，再批量查询关联表生成映射：

def get(self, request, **kwargs):
    # 1. 收集所有需要的关联ID
    # 获取所有用户的sectionid、deptid（以及其他关联表ID）
    user_rel_ids = Userstable.objects.using(dbname).values_list('sectionid', 'deptid')
    section_ids = {sid for sid, _ in user_rel_ids}
    dept_ids = {did for _, did in user_rel_ids}
    # 若有其他关联表，同理收集对应ID集合

    # 2. 批量查询关联表，生成ID到名称的映射字典
    section_map = {s.id: s.sectionname for s in Section.objects.using(dbname).filter(id__in=section_ids)}
    dept_map = {d.id: d.deptname for d in Section.objects.using(dbname).filter(id__in=dept_ids)}
    # 其他关联表同理生成映射，比如role_map = {...}

步骤2：修改序列化器（或直接跳过序列化器） #

如果继续使用Serializer，将映射字典传入context，在SerializerMethodField中直接取值：

class UserSerializer(serializers.ModelSerializer):
    section = serializers.SerializerMethodField()
    department = serializers.SerializerMethodField()

    class Meta:
        model = Userstable
        fields = '__all__'

    def get_section(self, obj):
        return self.context['section_map'].get(obj.sectionid, '')

    def get_department(self, obj):
        return self.context['dept_map'].get(obj.deptid, '')

在视图中传入context：

serializer = UserSerializer(
    datas, 
    context={
        'sector': sector,
        'section_map': section_map,
        'dept_map': dept_map
    }, 
    many=True
)

2. 流式输出CSV，避免内存爆炸 #

30万条数据一次性加载到内存会导致严重的性能瓶颈，改用StreamingHttpResponse流式写入CSV，无需临时文件，边查边输出：

import csv
from django.http import StreamingHttpResponse

def get(self, request, **kwargs):
    # 预取关联映射字典（同步骤1）
    # ...

    # 流式响应配置
    response = StreamingHttpResponse(content_type='text/csv')
    response['Content-Disposition'] = "attachment; filename=All-users.csv"
    
    writer = csv.writer(response)
    # 写入表头（按需调整字段）
    writer.writerow(['用户ID', '用户名', '部门', '科室', ...])
    
    # 分批加载用户数据，避免一次性加载所有记录到内存
    users = Userstable.objects.using(dbname).all().iterator(chunk_size=1000)
    
    # 逐行写入CSV
    for user in users:
        writer.writerow([
            user.id,
            user.username,
            dept_map.get(user.deptid, ''),
            section_map.get(user.sectionid, ''),
            # 其他字段依次补充
        ])
    
    return response

3. 额外优化建议 #

• 跳过Serializer直接处理：Serializer的序列化过程有一定性能开销，直接访问模型字段+映射字典的方式比用Serializer更快，即上面流式输出的写法
• 数据库索引优化：给Userstable的sectionid、deptid等关联ID字段添加数据库索引，能大幅提升批量查询关联ID的速度
• 调整chunk_size：根据服务器内存情况调整iterator(chunk_size=xxx)的数值，一般1000-2000较为合适
• 避免临时文件：原代码中写入临时文件再读取的方式会增加IO开销，用StreamingHttpResponse直接输出更高效

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