如何在无第三方SaaS工具下实现MongoDB Atlas到Google BigQuery的IoT数据持续复制？

1. 排查MongoDB Atlas Change Streams配置 #

• 确认集群规格：必须是M10及以上层级（M0/M2/M5免费/低阶集群不支持Change Streams）。
• 验证集合类型：避免使用capped集合，此类集合不支持Change Streams。
• 检查用户权限：Dataflow使用的MongoDB账号需配置readChangeStream和find权限，确保能读取变更流和集合数据。

2. 修正Dataflow CDC模板参数 #

• 调整mongoDBChangeStreamStartAtOption：设置为latest，确保从当前时间点开始捕获新增数据；若需从历史断点恢复，可设为resume并指定有效的checkpoint存储路径。
• 核对mongoDBUri格式：需包含完整认证信息、集群地址、目标库和集合，示例：mongodb+srv://<username>:<password>@cluster0.mongodb.net/<dbname>?authSource=admin。
• 优化BigQuery输出配置：若存储原始数据，可将表字段设为JSON类型，避免Schema不匹配导致的写入失败；同时确保Dataflow服务账号拥有BigQuery数据集的写入权限。

3. 自定义Dataflow作业（全量+增量结合） #

如果官方模板仍无法满足需求，可基于Apache Beam编写自定义作业，同时实现全量同步和持续CDC捕获：

• 核心逻辑：先一次性同步现有所有数据，再监听MongoDB Change Stream捕获新增/更新文档，持续写入BigQuery。
• 示例代码（Python）：

  import apache_beam as beam
  from apache_beam.io.mongodbio import ReadFromMongoDB, ReadFromMongoDBChangeStream
  
  def main():
      pipeline_options = beam.options.pipeline_options.PipelineOptions()
      with beam.Pipeline(options=pipeline_options) as p:
          # 全量同步现有数据
          full_sync = p | "Read Full MongoDB Data" >> ReadFromMongoDB(
              uri="mongodb+srv://<user>:<pass>@cluster.mongodb.net/",
              db="your_db",
              coll="your_collection"
          )
          full_sync | "Write Full to BigQuery" >> beam.io.WriteToBigQuery(
              "your-project:your_dataset.your_table",
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
              create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
          )
  
          # 持续监听Change Stream
          cdc_stream = p | "Read CDC Events" >> ReadFromMongoDBChangeStream(
              uri="mongodb+srv://<user>:<pass>@cluster.mongodb.net/",
              db="your_db",
              coll="your_collection",
              start_options={"start_at_operation_time": None}
          )
          # 提取完整文档（过滤Change Stream元数据）
          full_docs = cdc_stream | "Extract Full Document" >> beam.Map(lambda event: event["fullDocument"])
          full_docs | "Write CDC to BigQuery" >> beam.io.WriteToBigQuery(
              "your-project:your_dataset.your_table",
              write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
          )
  
  if __name__ == "__main__":
      main()
  

• 部署时选择Dataflow的流式运行模式，确保作业持续运行以捕获新增数据。

4. 手动验证Change Streams可用性 #

在MongoDB Atlas Shell中执行以下命令，验证Change Streams是否正常产生事件：

db.your_collection.watch()

插入一条测试数据后，若Shell输出包含fullDocument的变更事件，说明Change Streams正常工作；否则需回到第一步排查集群规格和权限配置。

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