flink cdc读取mongo哪位老哥做过吗？有什么需要注意的吗？

在使用Flink CDC读取MongoDB数据时，有以下几点需要注意：

1. 并发度与增量快照：

   • 如果启用了scan.incremental.snapshot.enabled= 'true'配置，在全量数据读取的初始阶段会支持并发读取。这意味着在进行全量数据同步时可以利用多线程加快处理速度。
   • 确保在从全量读切换到增量读时，作业能够正确地等待至少一个Checkpoint完成，以避免数据不一致问题。

2. 集合订阅模式：

   • MongoDB CDC连接器目前仅支持订阅整个数据库下的所有集合，而不支持单独指定或筛选部分集合。因此，在配置时需通过设置database参数来指定需要监控的数据库，而不能直接指定单个集合。

3. 启动模式配置：

   • 通过startupOptions配置项可以选择不同的启动模式，包括从初始位点、当前位点或特定时间戳开始读取数据。确保根据业务需求选择合适的启动模式，例如，若只需增量数据，应配置为StartupOptions.latest-offset()。

4. MongoDB版本兼容性：

   • 确保使用的MongoDB版本为3.6及以上，最好是4.0或更高版本，因为Change Stream特性是在MongoDB 3.6中引入的。低版本可能因缺少Change Stream功能而导致错误。

5. 运行环境要求：

   • MongoDB CDC要求MongoDB运行在副本集或分片集群架构下，不支持单实例模式下的Change Stream功能。如果在开发测试环境中使用单节点，需初始化为单节点副本集并注意可能出现的限制。

6. 用户凭证与认证源：

   • 如果遇到用户名密码错误的问题，检查是否正确设置了connection.options中的authSource参数，确保指向了正确的用户所在的数据库。

7. Checkpoint策略调整：

   • 虽然上述信息主要针对MySQL CDC，但合理设置Checkpoint间隔对于任何CDC作业都至关重要。避免因Checkpoint间隔过长导致增量数据同步延迟。

综上所述，使用Flink CDC读取MongoDB数据时，重点在于正确配置连接器参数，确保MongoDB版本兼容性，以及理解其对数据库结构和运行环境的要求。同时，合理的Checkpoint策略也是保证数据同步效率和准确性的关键因素。