Spark+Django+MongoDB三层架构实战：批流一体数据管道设计

1. 项目概述：为什么要把 Spark、Django 和 MongoDB 拉到一张工作台上？

你有没有遇到过这样的场景：用户在 Django 后台点下“生成月度销售热力图”按钮，页面转圈 47 秒，数据库连接池告警，Celery worker 队列堆到 200+，而最终返回的却只是一张聚合了 300 万条订单记录的静态图表？这不是性能瓶颈，这是架构错配——用 Web 框架的“小刀”去切大数据的“整头牛”。我第一次在电商 SaaS 项目里撞上这个问题时，运维同事直接把监控截图甩进群：“Django 的 ORM 正在用SELECT * FROM orders WHERE created_at > '2023-01-01'扫描 87GB 表，PG 的 shared_buffers 已经在冒烟。”

这正是Apache Spark + Django + MongoDB组合要解决的核心矛盾：让实时交互层（Django）专注响应、路由与状态管理，让分析计算层（Spark）专注吞吐、迭代与容错，让文档存储层（MongoDB）专注灵活 schema、高并发读写与嵌套查询。它不是炫技，而是分工——就像餐厅里，Django 是前台点单收银员（快、准、稳），MongoDB 是食材冷柜（随时取用、分类清晰），Spark 是后厨中央厨房（批量备料、统一烹制、支持翻车重做）。

这个组合特别适合三类真实业务：

• 用户行为分析平台：MongoDB 存原始埋点（{user_id: "u123", event: "click_cart", props: {item_id: "p456", category: "electronics"}}），Spark 每小时跑一次会话路径聚类，Django 提供可视化看板；
• 推荐系统在线服务：MongoDB 存用户画像快照（{user_id: "u123", interests: ["AI", "photography"], last_active: ISODate("...")}），Spark 基于历史行为训练 ALS 模型并导出向量，Django 接口实时查向量做近邻召回；
• IoT 设备数据管道：MongoDB 接入设备上报的 JSON（{device_id: "d789", temp: 23.5, humidity: 65, ts: 1712345678}），Spark Streaming 实时统计异常温度频次，Django 管理员后台触发告警工单。

关键词 “Apache Spark”、“Django”、“MongoDB” 在标题里不是并列关系，而是分层协作关系：Spark 是计算引擎（不直接暴露给用户），Django 是应用网关（不碰原始数据流），MongoDB 是数据湖的“浅水区”（存清洗后、可即席查询的宽表）。接下来我会拆解这个三角如何真正咬合——不是教你怎么装三个软件，而是告诉你数据从 MongoDB 流进 Spark、结果回写进 MongoDB、Django 如何安全调用整个链路的每一步实操细节、参数陷阱和线上踩坑记录。

2. 架构设计与技术选型逻辑：为什么是这三个，而不是其他组合？

2.1 为什么不用 Pandas 替代 Spark？

新手常问：“我用 Django 调 Pandas 读 MongoDB 不就完了？”——这在 10 万行数据内确实丝滑。但当数据量突破 500 万行，问题立刻暴露：

• 内存爆炸：Pandas 单进程加载 500 万行 × 20 列的 DataFrame，实测内存占用超 3.2GB，Django worker 直接 OOM；
• 无容错机制：Pandas 计算中途断电，所有中间状态丢失，必须重跑；
• 无法水平扩展：加机器？Pandas 不认。而 Spark 的 DAG 调度器天然支持失败任务重试、Stage 级别恢复、Executor 动态扩缩。

我曾用 Pandas 处理日志分析任务，单次运行耗时 18 分钟，失败率 12%（因内存抖动）；换成 Spark 后，同样逻辑耗时 4.3 分钟，失败率归零——因为 Spark 把 18 分钟的长任务拆成 237 个 Task，其中 3 个失败后自动在其他节点重跑，总耗时仅增加 11 秒。

2.2 为什么 MongoDB 而不是 PostgreSQL 或 Elasticsearch？

• vs PostgreSQL：PostgreSQL 的 JSONB 字段虽支持嵌套查询，但复杂聚合（如“统计每个用户最近 3 次点击的品类分布”）需多层子查询 + LATERAL JOIN，SQL 写起来反人类，且 PG 的并行查询对 JSONB 支持有限。MongoDB 的$unwind+$group+$facet流水线，一行$group: { _id: "$user_id", categories: { $push: "$props.category" } }就搞定。更重要的是，MongoDB 的分片集群能线性扩展写入吞吐——我们生产环境用 3 分片集群，峰值写入达 42,000 docs/sec，PG 同配置下写入瓶颈在 8,000 tps。

• vs Elasticsearch：ES 擅长全文检索和实时分析，但不擅长事务性更新和复杂关联。比如“给用户 u123 的画像文档追加一个recommendation_scores数组”，ES 需用 Update API + 脚本，而 MongoDB 的$push、$set原子操作一行命令解决。且 ES 的_source存储开销大，同样数据量比 MongoDB 多占 35% 磁盘。

提示：MongoDB 版本必须 ≥ 4.2（支持聚合管道中的$lookup关联分片集合）、Spark 连接器必须用mongo-spark-connector_2.12:10.2.3（适配 Spark 3.4+ 和 Scala 2.12），Django 版本建议 4.2 LTS（原生支持异步视图，避免阻塞 Spark 任务）。

2.3 为什么 Django 不直接连 Spark？

Django 的 HTTP 请求生命周期极短（通常 < 5 秒），而 Spark 任务启动开销大（Driver 初始化、资源申请、Executor 启动平均 8~12 秒）。如果每次请求都spark-submit，用户会看到“504 Gateway Timeout”。正确姿势是：Django 只负责触发、轮询、展示结果，Spark 任务由独立调度器（如 Airflow）或后台服务（如 Spark Thrift Server）长期运行。我们采用后者——部署 Spark Thrift Server 作为 JDBC 网关，Django 通过pyhive连接，像查普通数据库一样发 SQL，Spark 自动将 SQL 编译为 DAG 执行。

2.4 整体数据流向与组件职责边界

[设备/前端] → [Django REST API] → [MongoDB 写入] ↓ [Airflow 定时触发] ↓ [Spark Job: 读 MongoDB → 计算 → 写 MongoDB] ↓ [Django Admin / Dashboard 查询 MongoDB 结果集]

关键设计原则：

• 零直连：Django 代码里绝不出现SparkSession.builder()或sc.parallelize()；
• 结果导向：Spark 不返回原始数据，只写计算结果到 MongoDB 的专用集合（如analytics_user_segments）；
• 幂等写入：Spark 任务每次运行前先drop()目标集合，避免脏数据累积；
• Django 异步化：用户点击“生成报告”后，Django 返回{"task_id": "spark_job_20240515_001"}，前端轮询/api/task-status/{id}/获取进度。

这个架构让每个组件回归本质：MongoDB 是数据容器，Spark 是计算工厂，Django 是服务接口。接下来，我们进入最硬核的部分——如何让这三者真正握手。

3. 核心实现与实操细节：从环境搭建到生产级部署

3.1 环境准备：版本锁死与依赖隔离

生产环境最怕“在我机器上能跑”。我们强制锁死以下版本（2024 年实测稳定组合）：

• MongoDB：6.0.14（分片集群，3 config server + 3 shard server + 1 mongos）
• Spark：3.4.1（Hadoop 3.3.4，Scala 2.12）
• Django：4.2.11（Python 3.11.8）
• 连接器：mongo-spark-connector_2.12:10.2.3（注意：_2.12 表示 Scala 版本，必须与 Spark 匹配）

注意：Spark 3.4 默认使用 Scala 2.12，若误装mongo-spark-connector_2.11，运行时抛java.lang.NoClassDefFoundError: scala/Function1——这是类加载器找不到 Scala 标准库，不是代码问题。

Django 项目依赖管理（requirements.txt关键行）：

Django==4.2.11 pymongo==4.6.1 pyhive[hive]==0.7.0 # 用于连接 Spark Thrift Server thrift==0.18.1 requests==2.31.0 # Spark 连接器不放这里！它只在 Spark 集群节点安装

Spark 集群依赖安装（在每个 Worker 节点执行）：

# 下载 connector JAR（必须放 SPARK_HOME/jars/ 下） wget https://repo1.maven.org/maven2/org/mongodb/mongo-spark-connector_2.12/10.2.3/mongo-spark-connector_2.12-10.2.3.jar mv mongo-spark-connector_2.12-10.2.3.jar $SPARK_HOME/jars/ # 验证：spark-shell 启动后执行 # scala> spark.read.format("mongodb").option("uri","mongodb://localhost:27017/test.coll").load() # 若报错 "Failed to find data source: mongodb"，说明 JAR 未加载成功

3.2 MongoDB 数据建模：为 Spark 计算优化 Schema

MongoDB 不是“随便存”，Schema 设计直接影响 Spark 读取效率。以用户行为日志为例，错误设计：

// ❌ 反模式：嵌套过深 + 类型混杂 { "_id": "...", "user": { "id": "u123", "profile": { "name": "Alice", "age": 28 } }, "events": [ { "type": "page_view", "url": "/home", "ts": "2024-05-15T10:00:00Z" }, { "type": "click", "element": "cart_btn", "ts": "2024-05-15T10:00:05Z" } ] }

问题：Spark 读取时events数组需explode()才能展开，产生大量中间 shuffle；profile嵌套对象导致user.profile.name访问慢。

正确建模（宽表思想，预计算字段）：

// ✅ 生产模式：扁平化 + 时间分区 + 索引预置 { "_id": ObjectId("..."), "user_id": "u123", "user_name": "Alice", "user_age": 28, "event_type": "page_view", "event_url": "/home", "event_ts": {"$date": "2024-05-15T10:00:00Z"}, "hour_partition": "2024-05-15-10", // 按小时分区，便于 Spark filter 下推 "day_partition": "2024-05-15" }

必须创建的索引（在 MongoDB Shell 中执行）：

// 加速 Spark 的 filter 下推（WHERE hour_partition = '2024-05-15-10'） db.events.createIndex({ "hour_partition": 1 }) // 加速 groupBy（Spark 的 reduceByKey 会利用此索引） db.events.createIndex({ "user_id": 1, "event_type": 1 }) // 复合索引覆盖常用查询 db.events.createIndex({ "day_partition": 1, "user_id": 1, "event_ts": -1 })

实测：添加hour_partition索引后，Spark 读取单小时数据耗时从 210 秒降至 38 秒——因为 MongoDB 能直接定位到对应 chunk，无需扫描全量分片。

3.3 Spark 任务开发：从读取到写入的完整流水线

我们以“计算用户 7 日活跃度分层”为例（输出到analytics.user_segments集合）。核心代码（spark_job.py）：

from pyspark.sql import SparkSession from pyspark.sql.functions import col, count, when, lit, current_date, date_sub from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType # 1. 初始化 SparkSession（关键参数！） spark = SparkSession.builder \ .appName("user-segmentation-job") \ .config("spark.mongodb.input.uri", "mongodb://mongo1:27017/myapp.events") \ .config("spark.mongodb.output.uri", "mongodb://mongo1:27017/myapp.analytics_user_segments") \ .config("spark.sql.adaptive.enabled", "true") \ # 开启自适应查询执行，Spark 3.2+ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ .getOrCreate() # 2. 定义 Schema（强制指定！避免 Spark 推断错误类型） schema = StructType([ StructField("user_id", StringType(), False), StructField("event_type", StringType(), False), StructField("event_ts", TimestampType(), False), StructField("hour_partition", StringType(), False) ]) # 3. 读取 MongoDB（filter 下推到 MongoDB 层！） raw_df = spark.read \ .format("mongodb") \ .option("database", "myapp") \ .option("collection", "events") \ .option("pipeline", "[{'$match': {'hour_partition': {'$gte': '2024-05-15-00'}}}]") \ # 关键！MongoDB 原生 pipeline .schema(schema) \ .load() # 4. 业务逻辑：计算 7 日活跃用户数 & 最近登录时间 from pyspark.sql.window import Window window_spec = Window.partitionBy("user_id").orderBy(col("event_ts").desc()) active_users = raw_df \ .filter(col("event_type") == "login") \ .withColumn("rn", row_number().over(window_spec)) \ .filter(col("rn") == 1) \ .withColumn("last_login", col("event_ts")) \ .select("user_id", "last_login") # 5. 写入 MongoDB（覆盖模式，确保幂等） active_users.write \ .format("mongodb") \ .mode("overwrite") \ # ⚠️ 生产必须用 overwrite，避免数据叠加 .option("database", "myapp") \ .option("collection", "analytics_user_segments") \ .option("replaceDocument", "false") \ # false=按_id更新，true=全量替换（慎用） .save() spark.stop()

关键参数解析：

• .option("pipeline", "...")：将$match下推到 MongoDB，Spark 只拉匹配数据，网络传输量降 92%；
• .mode("overwrite")：生产环境唯一安全模式，每次任务先清空目标集合再写入；
• .option("replaceDocument", "false")：设为false时，Spark 用_id字段做 upsert（存在则更新，不存在则插入），避免重复 ID 冲突。

提交命令（生产环境必须用--master yarn或--master k8s://）：

spark-submit \ --master yarn \ --deploy-mode cluster \ --conf spark.yarn.submit.waitAppCompletion=false \ # 异步提交，Django 不阻塞 --jars $SPARK_HOME/jars/mongo-spark-connector_2.12-10.2.3.jar \ spark_job.py

3.4 Django 集成：安全调用与状态管理

Django 不直接运行 Spark，而是通过REST API 触发 Airflow DAG或调用 Spark Thrift Server。我们选择后者（更轻量）：

步骤 1：启动 Spark Thrift Server

# 在 Spark 集群 Master 节点执行 $SPARK_HOME/sbin/start-thriftserver.sh \ --master yarn \ --deploy-mode cluster \ --conf spark.sql.hive.thriftServer.singleSession=true \ --conf spark.sql.adaptive.enabled=true \ --jars $SPARK_HOME/jars/mongo-spark-connector_2.12-10.2.3.jar

默认监听thrift://spark-master:10000。

步骤 2：Django 视图调用 HiveServer2

# views.py from pyhive import hive from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt import json @csrf_exempt def trigger_segmentation(request): if request.method == 'POST': # 1. 提交 Spark SQL 任务（非阻塞） conn = hive.Connection(host="spark-master", port=10000, username="admin") cursor = conn.cursor() # 执行预编译的 SQL（实际应存入 DB，防注入） cursor.execute(""" INSERT OVERWRITE TABLE analytics_user_segments SELECT user_id, COUNT(*) as login_count, MAX(event_ts) as last_login FROM myapp.events WHERE event_type = 'login' AND hour_partition >= '2024-05-15-00' GROUP BY user_id """) cursor.close() conn.close() # 2. 返回任务 ID 供前端轮询 task_id = f"seg_{int(time.time())}" # 3. 将 task_id 和状态存入 Django DB（models.TaskStatus） TaskStatus.objects.create(task_id=task_id, status="RUNNING") return JsonResponse({"task_id": task_id, "status": "RUNNING"})

步骤 3：前端轮询状态（Vue 示例）：

// 轮询 /api/task-status/{id}/ async checkTaskStatus(taskId) { const res = await fetch(`/api/task-status/${taskId}/`); const data = await res.json(); if (data.status === "SUCCESS") { this.segments = await this.fetchSegments(); // 从 MongoDB 读结果 } else if (data.status === "RUNNING") { setTimeout(() => this.checkTaskStatus(taskId), 2000); // 2秒后重试 } }

Django 读取结果（直接查 MongoDB，零 Spark 依赖）：

# models.py class UserSegment(models.Model): user_id = models.CharField(max_length=50) login_count = models.IntegerField() last_login = models.DateTimeField() class Meta: managed = False # 不由 Django 管理表结构 db_table = 'analytics_user_segments' # 对应 MongoDB 集合名 # views.py def get_segments(request): segments = list(UserSegment.objects.all().values()) # pymongo 驱动自动映射 return JsonResponse({"segments": segments})

3.5 生产部署 checklist：12 项必须验证

4. 常见问题与排查技巧实录：线上故障的 7 个真实现场

4.1 问题 1：Spark 读 MongoDB 报com.mongodb.MongoTimeoutException: Timed out after 30000 ms

现象：Spark Driver 日志显示Failed to connect to mongodb://mongo1:27017，但mongoshell 能连通。
根因：Spark 连接器默认使用maxConnectionLifeTimeMS=0（永不过期），而 MongoDB 分片集群的 mongos 进程会主动关闭空闲连接（默认 60 分钟）。当 Spark 任务长时间运行，连接被 mongos 断开，后续请求超时。
解决方案：在 SparkSession 配置中显式设置连接生命周期：

spark = SparkSession.builder \ .config("spark.mongodb.input.uri", "mongodb://mongo1:27017/myapp.events?maxConnectionLifeTimeMS=300000") \ .config("spark.mongodb.output.uri", "mongodb://mongo1:27017/myapp.result?maxConnectionLifeTimeMS=300000") \ .getOrCreate()

300000ms = 5分钟，确保连接在 mongos 关闭前主动刷新。

4.2 问题 2：Django 调用 Spark Thrift Server 报TTransportException: Could not connect to spark-master:10000

现象：Django 日志ConnectionRefusedError: [Errno 111] Connection refused。
排查路径：

1. kubectl get pods -n spark（K8s）或yarn node -list（YARN）确认 Thrift Server Pod/Container 是否 Running；
2. kubectl logs -n spark thrift-server-pod查看启动日志，是否报Failed to bind to 10000（端口被占）；
3. kubectl exec -it thrift-server-pod -- netstat -tuln | grep 10000确认端口监听地址是0.0.0.0:10000而非127.0.0.1:10000（后者只允许本地访问）。
   修复：在start-thriftserver.sh中添加--conf spark.sql.hive.server2.thrift.bind.host=0.0.0.0。

4.3 问题 3：Spark 写入 MongoDB 后，Django 读不到新数据

现象：Spark 任务日志显示Saved 12,456 rows to mongodb://...，但 DjangoUserSegment.objects.count()返回 0。
根因：MongoDB 集合名大小写敏感！Spark 写入analytics_user_segments，而 Django model 的db_table = 'Analytics_User_Segments'（首字母大写）。
验证：在 MongoDB Shell 中执行show collections，确认集合名完全一致。
修复：Django model 中db_table必须与 Spark 写入的 collection 名逐字符相同（包括下划线、大小写）。

4.4 问题 4：Spark 任务耗时突增 300%，CPU 使用率 100%

现象：原本 5 分钟的任务，某天突然跑 22 分钟，YARN UI 显示所有 Executor CPU 持续 100%。
根因：MongoDB 的hour_partition字段类型从string被误改为int（如2024051510），Spark 无法下推$match，被迫拉取全量数据。
排查：

• Spark UI 的 SQL tab 查看 Physical Plan，若Scan mongodb下没有PushedFilters，说明 filter 未下推；
• MongoDB 日志grep "find" /var/log/mongodb/mongod.log，确认查询是否带hour_partition条件。
  修复：将hour_partition字段统一为字符串，并重建索引。

4.5 问题 5：Django Admin 中 MongoDB 数据显示为None

现象：Django Admin 列表页所有字段显示None，但pymongo直连能查到数据。
根因：Django 的pymongo驱动默认将 MongoDB 的ObjectId当作str，而 Django model 字段定义为CharField，类型不匹配导致解析失败。
修复：在 model 中显式处理_id：

class UserSegment(models.Model): _id = models.CharField(primary_key=True, max_length=24) # ObjectId 是 24 位 hex string user_id = models.CharField(max_length=50) # ... 其他字段

并在admin.py中注册：

@admin.register(UserSegment) class UserSegmentAdmin(admin.ModelAdmin): list_display = ('_id', 'user_id', 'login_count') # 显式列出字段

4.6 问题 6：Spark Thrift Server 内存溢出（OOM Killed）

现象：kubectl describe pod thrift-server显示State: Terminated Reason: OOMKilled。
根因：Thrift Server 默认内存 1G，但执行复杂 SQL（如多表 join）需更多内存。
修复：启动时增加内存：

$SPARK_HOME/sbin/start-thriftserver.sh \ --driver-memory 4g \ --executor-memory 8g \ --executor-cores 4 \ ...

4.7 问题 7：Airflow 触发 Spark 任务后，Django 轮询永远返回RUNNING

现象：Spark 任务已成功写入 MongoDB，但 Django 的TaskStatus表中状态仍为RUNNING。
根因：Airflow DAG 中未调用TaskStatus.objects.filter(task_id=xxx).update(status="SUCCESS")。
修复模板（Airflow PythonOperator）：

def mark_task_success(**context): task_id = context['dag_run'].conf.get('task_id') TaskStatus.objects.filter(task_id=task_id).update(status="SUCCESS") mark_success_task = PythonOperator( task_id='mark_success', python_callable=mark_task_success, trigger_rule='all_success' # 仅在上游任务成功后执行 )

5. 性能调优与扩展实践：从单机到千节点集群

5.1 Spark 读取 MongoDB 的 5 个加速技巧

1. Pipeline 下推优先级最高：

   # ✅ 最佳：MongoDB 原生 pipeline，过滤 99% 数据 .option("pipeline", "[{'$match': {'hour_partition': '2024-05-15-10', 'event_type': 'login'}}, {'$project': {'user_id': 1, 'event_ts': 1}}]") # ❌ 次佳：Spark filter，全量数据进 Spark .filter((col("hour_partition") == "2024-05-15-10") & (col("event_type") == "login"))

2. Projection 减少网络传输：在 pipeline 中用$project只取必要字段，实测减少 68% 网络流量。

3. Batch Size 调优：默认batchSize=0（MongoDB 自动决定），生产环境设为10000：

   .option("batchSize", "10000")

   避免小包频繁传输，提升吞吐。

4. 并行度控制：Spark 会根据 MongoDB 的 chunk 数量自动分片，但可通过partitioner强制：

   .option("partitioner", "MongoShardedPartitioner") \ # 分片集群用此 .option("partitionerOptions.partitionKey", "user_id") \ # 指定分片键

5. Kryo 序列化必开：比 Java 序列化快 10 倍，内存占用少 50%：

   .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ .config("spark.kryo.registrator", "com.mongodb.spark.sql.DefaultMongoKryoRegistrator")

5.2 Django 高并发下的 MongoDB 连接池优化

默认pymongo连接池maxPoolSize=100，但在 1000 QPS 下仍可能耗尽。优化方案：

• 连接复用：Django 中全局单例 MongoClient：

  # utils.py MONGO_CLIENT = MongoClient( os.getenv('MONGO_URI'), maxPoolSize=500, # 提升至 500 minPoolSize=50, # 预热 50 连接 maxIdleTimeMS=600000, # 空闲 10 分钟释放 connectTimeoutMS=5000, socketTimeoutMS=30000 )

• 读写分离：为分析集合（analytics.*）单独配置只读副本集 URI，减轻主库压力。

5.3 从单机 Spark 到 YARN/K8s 集群的平滑迁移

单机开发时用--master local[*]，上线必须切集群。关键差异：

迁移检查清单：

• ✅ 确保 HDFS 或 K8s 存储中存在 connector JAR；
• ✅ Spark 配置spark.hadoop.fs.defaultFS指向正确 HDFS 地址；
• ✅ K8s 中为 Spark ServiceAccount 绑定spark-roleRBAC 权限。

5.4 业务扩展：加入实时流处理（Spark Streaming + Kafka）

当需求从“小时级批处理”升级到“秒级实时分析”，架构演进为：

[设备] → [Kafka] → [Spark Streaming] → [MongoDB] → [Django] ↓ [实时告警规则引擎]

关键改造点：

• Spark Streaming 用foreachBatch将微批次写入 MongoDB：

  def write_to_mongo(batch_df, batch_id): batch_df.write \ .format("mongodb") \ .mode("append") \ .option("database", "myapp") \ .option("collection", "realtime_alerts") \ .save() query = streaming_df.writeStream \ .foreachBatch(write_to_mongo) \ .start()

• Django 用 Server-Sent Events（SSE）推送实时告警，替代轮询。

5.5 成本监控