如何在Spark Streaming上下文中获取MongoDB集合的更新数据

你可以直接使用MongoDB官方提供的Spark连接器对接MongoDB Change Stream功能实现需求，无需自己开发Custom Receiver，整体实现门槛很低：

• 前置要求
  • MongoDB服务版本≥3.6，且部署为副本集或分片集群（Change Stream依赖oplog实现，单节点MongoDB不支持该功能）
  • 使用的MongoDB Spark连接器版本≥3.0，需和你当前的Spark版本、MongoDB服务版本兼容
• 核心实现（Structured Streaming场景，Spark 2.3+支持）
  直接通过连接器提供的流式数据源读取变更流即可，参考代码如下：

  import org.apache.spark.sql.SparkSession
  
  val spark = SparkSession.builder()
    .appName("MongoChangeStreamConsumer")
    .config("spark.mongodb.input.uri", "mongodb://你的MongoDB地址/数据库名.集合名")
    // 配置更新操作返回完整变更后的文档，不需要可以删除该配置
    .config("spark.mongodb.change.stream.publish.full.document", "updateLookup")
    .getOrCreate()
  
  // 加载MongoDB变更流
  val changeStreamDF = spark.readStream
    .format("mongodb")
    .load()
  
  // 此处替换为你自己的业务处理逻辑，示例为打印到控制台
  val processingQuery = changeStreamDF.writeStream
    .format("console")
    // 配置检查点，任务重启后可从断点继续消费
    .option("checkpointLocation", "你的检查点存储路径")
    .start()
  
  processingQuery.awaitTermination()
  

• 变更数据字段说明
  读取到的DataFrame中包含所有变更相关信息，常用字段如下：
  • operationType：标识操作类型，可选值包括insert/update/delete/replace等
  • fullDocument：存储变更后的完整文档内容
  • documentKey：存储变更文档的_id字段值
  • updateDescription：仅更新操作存在，存储具体修改的字段和变更前后值
• 老版本Spark Streaming（DStream API）适配
  如果你还在使用DStream API，也可以直接基于MongoDB Java异步客户端封装Change Stream的拉取逻辑，配合StreamingContext.queueStream生成对应的DStream，社区已有大量成熟的封装实现可以直接复用，不需要自己完整实现Custom Receiver。

• 可以在读取变更流时传入过滤Pipeline，只将你需要的变更事件推送到Spark侧，大幅减少不必要的数据传输
• 对于数据一致性要求高的场景，必须配置检查点路径，同时保证MongoDB的oplog保留时长大于Spark任务的最长故障恢复时间，避免位点过期导致数据丢失
• 单文档高频更新场景可以配合Spark窗口算子做聚合去重，避免重复处理同一个文档的多次临时变更

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