Django旅行保险系统批量上传100条记录耗时30秒求优化

问题背景 #

开发的旅行保险项目支持代理人批量上传已购保单的旅行者信息，当前实现的流程处理100条Excel记录需耗时30秒，代理人每月至少需上传1000条记录，用户体验极差，需定位性能瓶颈并提供优化方案。

现有流程与数据模型 #

业务流程 #

1. 管理员创建Benefits（医疗费用、失窃保障等）
2. 管理员创建Policy，关联多个Package，为每个Package分配Benefits及限额，分配金额存储在PackageBenefit表
3. 代理人通过Excel上传旅行者信息，数据流向：User → Traveller → PolicyHolder → PolicyUsage

核心数据模型 #

• User：用户基础信息
• Traveller：与User一对一关联，存储旅行者详情
• Policy：保单主表
• Package：与Policy外键关联，保单套餐
• Benefits：保障项目表
• PackageBenefit：与Policy、Package外键关联，存储套餐内Benefit的最高限额
• PolicyHolder：与Traveller、Policy、Package外键关联，记录旅行者的保单购买信息
• PolicyUsage：与PolicyHolder、Package、Benefits关联，存储旅行者的套餐保障限额详情

性能瓶颈分析 #

从现有代码中可定位以下核心性能问题：

1. 关联对象依赖内存实例，无法利用数据库批量ID生成
   当前代码先在内存中创建User、Traveller、PolicyHolder实例再批量创建，但Traveller依赖User的ID、PolicyHolder依赖Traveller的ID，内存实例没有数据库生成的ID，后续PolicyUsage关联时存在隐性问题，且无法发挥批量操作的最优效率。

2. 全量加载PackageBenefit数据冗余
   代码中查询所有PackageBenefit再分组，若PackageBenefit数据量较大，会加载大量不必要的数据，增加内存占用和查询时间。

3. PolicyHolder存储冗余字段
   PolicyHolder中存储了policy_name、package_name、policy_code等冗余字段，这些字段可通过关联Policy、Package查询获取，冗余存储增加了数据写入量和维护成本。

4. 循环内重复计算与IO操作

   • 在循环中逐个处理日期转换（pd.to_datetime），效率远低于pandas批量处理
   • 字符串替换操作（用户名、邮箱生成）在循环外可批量完成，无需逐行处理

5. 日期格式验证效率低
   使用apply逐行验证日期格式，pandas提供了更高效的批量日期转换验证方式。

6. 同步阻塞请求
   上传操作在请求线程中同步执行，处理大量数据时会阻塞请求，导致用户长时间等待。

优化方案 #

1. 调整批量创建顺序，利用数据库生成的ID #

先批量创建User，获取数据库生成的ID后，再批量创建Traveller和PolicyHolder，确保关联关系正确且高效。

2. 优化PackageBenefit查询 #

仅查询当前上传记录涉及的package_id对应的PackageBenefit数据，用values_list获取所需字段，减少数据加载量。

3. 移除PolicyHolder冗余字段 #

删除policy_name、package_name等冗余字段，通过模型关联在查询时动态获取，减少写入数据量。

4. 批量处理数据转换 #

• 在pandas层面批量处理日期转换和格式验证
• 批量生成用户名、邮箱、密码，避免循环内重复操作

5. 异步处理上传任务 #

使用Celery将上传任务放到后台异步执行，用户提交文件后即可收到反馈，无需等待处理完成。

6. 提高单次上传限制 #

配合异步处理，将单次上传限制从100条提高到500-1000条，减少代理人操作次数。

7. 数据库层面优化 #

• 为PolicyHolder、PackageBenefit的关联字段（policy_id、package_id、traveller_id）添加索引
• 确保批量操作使用数据库的批量插入语法（Django的bulk_create默认支持）

代码优化示例 #

核心优化部分代码 #

def bulk_upload_traveller(request):
    form = AddTravellerProfileForm()

    if request.method == "POST" and request.FILES["file"]:
        file = request.FILES["file"]

        try:
            df = pd.read_excel(file)
        except Exception as e:
            messages.error(request, "Failed to read Excel file, please re-check the file and try again")
            return render(request, "administrator/user_management/travellers/traveller_bulk_upload/bulk_upload_failure.html")

        # 提高单次上传限制
        if len(df) > 1000:
            messages.error(request, "Exceeded maximum limit of 1000 records per upload.")
            return render(request, "administrator/user_management/travellers/traveller_bulk_upload/bulk_upload_failure.html")

        # 批量验证日期格式
        required_dates = ["departure_date", "dob", "return_date"]
        for col in required_dates:
            df[col] = pd.to_datetime(df[col], errors='coerce')
        invalid_date_records = df[df[required_dates].isnull().any(axis=1)]
        if not invalid_date_records.empty:
            messages.error(request, f"Invalid date format in records: {invalid_date_records.index.tolist()}")
            return render(request, "administrator/user_management/travellers/traveller_bulk_upload/bulk_upload_failure.html")

        # 预查询所需的Policy和Package数据
        uploaded_policy_ids = df["policy_id"].unique()
        uploaded_package_ids = df["package_id"].unique()
        _policies = {p.id: (p.policy_name, p.policy_code) for p in Policy.objects.filter(id__in=uploaded_policy_ids)}
        _packages = {p.id: (p.package_name, p.package_code, p.validity) for p in Package.objects.filter(id__in=uploaded_package_ids)}

        # 验证policy_id和package_id有效性
        invalid_policy_records = df[~df["policy_id"].isin(_policies.keys())]
        invalid_package_records = df[~df["package_id"].isin(_packages.keys())]
        # ... 错误处理逻辑略

        # 批量生成用户名、邮箱、密码
        df["username"] = df["username"].fillna(df["name"].str.lower().str.replace(" ", "_"))
        df["email"] = df["email"].fillna(df["username"] + "@ksa.alkhairicare.com")
        df["password"] = df["password"].fillna(df["passport_no"].str.lower().str.replace(" ", "_"))

        # 批量创建User
        users = []
        for _, row in df.iterrows():
            users.append(
                User(
                    username=row["username"],
                    email=row["email"],
                    entity_name=row["name"],
                    password=make_password(row["password"]),
                    role=CUSTOMERS
                )
            )
        User.objects.bulk_create(users, batch_size=200)

        # 获取创建后的User ID映射
        user_username_map = {user.username: user.id for user in User.objects.filter(username__in=df["username"].tolist())}
        df["user_id"] = df["username"].map(user_username_map)

        # 批量创建Traveller
        travellers = []
        for _, row in df.iterrows():
            travellers.append(
                Traveller(
                    user_id=row["user_id"],
                    name=row["name"],
                    passport_no=row["passport_no"],
                    id_card_no=row["id_card_no"],
                    dob=row["dob"],
                    primary_email=row["email"],
                    is_approved=True,
                    gender=row.get("gender", ""),
                    nationality=row.get("nationality", ""),
                    country=row.get("country", ""),
                    hp_no=row.get("hp_no", "")
                )
            )
        Traveller.objects.bulk_create(travellers, batch_size=200)

        # 获取Traveller ID映射
        traveller_user_map = {t.user_id: t.id for t in Traveller.objects.filter(user_id__in=df["user_id"].tolist())}
        df["traveller_id"] = df["user_id"].map(traveller_user_map)

        # 批量创建PolicyHolder（移除冗余字段）
        policy_holders = []
        for _, row in df.iterrows():
            policy_name, policy_code = _policies[row["policy_id"]]
            package_name, package_code, package_validity = _packages[row["package_id"]]
            departure_date = row["departure_date"].replace(hour=0, minute=0, second=0)
            validity_end = departure_date + pd.Timedelta(days=int(package_validity))

            policy_holders.append(
                PolicyHolder(
                    traveller_id=row["traveller_id"],
                    policy_id=row["policy_id"],
                    package_id=row["package_id"],
                    holder_name=row["name"],
                    total_validity=package_validity,
                    validity_start=departure_date,
                    validity_end=validity_end,
                    payment_status="manual",
                    departure_date=row["departure_date"],
                    return_date=row["return_date"],
                    ecert_no=row["ecert_no"]
                )
            )
        PolicyHolder.objects.bulk_create(policy_holders, batch_size=200)

        # 预查询当前上传涉及的PackageBenefit
        package_benefits = PackageBenefit.objects.filter(package_id__in=uploaded_package_ids).values_list(
            "package_id", "benefit_id", "benefit__name", "actual_limit", "currency"
        )
        package_benefit_map = {}
        for pb in package_benefits:
            package_id = pb[0]
            if package_id not in package_benefit_map:
                package_benefit_map[package_id] = []
            package_benefit_map[package_id].append(pb[1:])

        # 批量创建PolicyUsage
        policy_usages = []
        policy_holder_map = {ph.traveller_id: ph.id for ph in PolicyHolder.objects.filter(traveller_id__in=df["traveller_id"].tolist())}
        for _, row in df.iterrows():
            policy_holder_id = policy_holder_map[row["traveller_id"]]
            package_id = row["package_id"]
            for benefit_data in package_benefit_map.get(package_id, []):
                benefit_id, benefit_name, actual_limit, currency = benefit_data
                policy_usages.append(
                    PolicyUsage(
                        policy_holder_id=policy_holder_id,
                        package_id=package_id,
                        benefit_id=benefit_id,
                        benfit_name=benefit_name,
                        actual_limit=actual_limit,
                        remaining_limit=actual_limit,
                        currency=currency
                    )
                )
        PolicyUsage.objects.bulk_create(policy_usages, batch_size=500)

        messages.success(request, "Bulk upload process was successfully completed.", extra_tags="success")
        # ... 后续渲染逻辑略

异步任务改造示例 #

# tasks.py
from celery import shared_task
from django.db import transaction
import pandas as pd
from .models import *

@shared_task
def process_bulk_upload(file_path):
    try:
        df = pd.read_excel(file_path)
        # ... 上述数据验证、批量创建逻辑
        return {"status": "success", "count": len(df)}
    except Exception as e:
        return {"status": "failed", "error": str(e)}

# 视图中调用异步任务
def bulk_upload_traveller(request):
    if request.method == "POST" and request.FILES["file"]:
        file = request.FILES["file"]
        # 保存文件到临时路径
        temp_file_path = f"/tmp/{file.name}"
        with open(temp_file_path, "wb+") as f:
            for chunk in file.chunks():
                f.write(chunk)
        # 触发异步任务
        process_bulk_upload.delay(temp_file_path)
        messages.success(request, "Upload task started, you will be notified when completed.")
        return render(request, "upload_pending.html")
    # ... 其他逻辑

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