django 不该直接做 etl 主干流程，因其 orm 和请求响应模型非为长时批处理设计，易致超时、内存溢出、任务卡死；应由 celery + pandas 承担清洗，django 仅负责调度、状态管理、权限与展示。
为什么 Django 不该直接做 ETL 主干流程
Django 的 ORM 和请求响应模型不是为长时、批处理、内存敏感型 ETL 设计的。你硬在 views.py 里写 pd.read_csv() + save()，很快会遇到超时、内存溢出、任务卡死、无法重试等问题。Django 适合调度、状态管理、结果展示和权限控制——ETL 的“脏活”得交给更合适的人干。
用 Django 管理 Celery + Pandas 的清洗任务
把清洗逻辑写成独立函数，注册为 Celery 任务；Django 只负责触发、传参、查状态、存日志。这样既复用 Django 的用户/权限/API，又避开其同步阻塞限制。
-
tasks.py中定义清洗任务：接受文件路径或数据源 ID，返回清洗后记录数与错误摘要 - 用
django-celery-beat配置定时任务，比如每天凌晨 2 点拉取 S3 上的raw_orders_*.csv - 清洗失败时，Celery 自动重试（
autoretry_for=(Exception,)），同时写入CleanLog模型，带traceback字段 - 避免在任务中直接操作
request或HttpResponse—— Celery worker 没有请求上下文
如何让 Pandas 清洗逻辑可配置、不硬编码
把字段映射、缺失值策略、类型转换规则从代码里抽出来，存在数据库或 YAML 文件里。否则每次加个新数据源就得改 Python 文件、发版、重启 worker。
- 建一个
CleaningRule模型，字段含source_name、column_mapping（JSONField）、drop_if_empty（BooleanField） - 清洗函数读取对应规则：
rule = CleaningRule.objects.get(source_name="shopify_orders") - Pandas 处理时用
df.astype(rule.column_dtypes)而不是写死df["price"].astype("float64") - 对日期列统一用
pd.to_datetime(df[col], errors="coerce")，避免因格式不一导致整列变NaT
上传 CSV 后自动触发清洗，但别卡住浏览器
用户点“上传”后立刻返回任务 ID，前端轮询 /api/clean-status/?task_id=xxx，而不是等清洗完再给响应。
- 视图函数只调用
clean_csv_task.delay(file_path, rule_id)，返回{"task_id": task.id} - 不要用
clean_csv_task.apply()（同步执行），也不要保存原始文件到media/后再读 —— 用TemporaryFile或直接存到/tmp/并设好清理策略 - 如果上传文件 >100MB，改用分块上传 + 合并，避免 Django 中间件耗尽内存；此时清洗任务应接收的是对象存储 URL（如 S3 presigned URL），而非本地路径
clean_csv_task.apply_async(..., countdown=1)。漏掉这个环节，管道就算跑起来，也迟早变成黑盒。Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
