Flink CDC sql 版本，mongo connector 如何提升source的消费速率？

Flink CDC sql 版本，mongo connector 如何提升source的消费速率？【业务背景】有一批百亿级别的数据需要同步从mongo同步到doris，存量+增量都要同步，所以想使用CDC同步，但是发现速率太慢，目前QPS 2w，预计300亿要同步20天左右才能跑完。【当前 job 现状】UI上看到QPS在2万左右，并且source的并行度一直是1。请问如何提高source的并行度？以及并行度的提升是否有助于提升消费速率？sql 的 source table 配置如下，这里在CDC文档中没有找到可以配置source并行度的地方。提高了拉数据的batch size。发现对source的QPS没有提升效果。
之前尝试过这样的方法，先用data X同步存量数据，再用CDC消费mongo最新的offset到doris。出现的问题是后续mongo中删除存量数据，cdc在doris中不会完成删除操作，只有cdc自己增量同步的数据跟随mongo删除。 ps: 按文档说明doris开启了批量删除，并使用了unique key表模型。

展开

收起

真的很搞笑 2023-12-05 20:22:50 67 0

2 条回答

写回答

取消提交回答

芯在这

可以先用其他工具做全量同步，再用cdc增量，此回答整理自钉群“Flink CDC 社区”

2023-12-06 13:29:35

赞同展开评论打赏
小周sir

面对过去，不要迷离；面对未来，不必彷徨；活在今天，你只要把自己完全展示给别人看。
针对您的问题，可以尝试以下方法来提高Flink CDC MongoDB Source的消费速率：
1. 增加并行度：可以通过调整Flink作业的并行度来提高消费速率。在Flink SQL中，可以使用SET parallelism语句来设置并行度。例如，将并行度设置为4：
```
SET parallelism.default=4;
```
1. 调整拉取数据的batch size：虽然您已经尝试过提高batch size，但仍然可以尝试进一步调整它以找到最佳值。可以在Flink SQL中使用SET table.exec.buffer-timeout和SET table.exec.max.buffer-size语句来调整batch size。例如：
```
SET table.exec.buffer-timeout=5s;
SET table.exec.max.buffer-size=10000;
```
1. 优化源表配置：确保源表的配置是最优的，例如使用合适的索引、分区等。这有助于提高查询性能，从而提高消费速率。
2. 考虑使用其他连接器：如果当前的MongoDB Connector无法满足性能需求，可以考虑使用其他支持更高消费速率的连接器，如Debezium MongoDB Connector。
3. 优化目标端Doris：检查Doris集群的配置和性能，确保它们能够支持高并发的数据写入。可以考虑增加Doris集群的资源或优化其配置以提高写入性能。
2023-12-05 21:04:20

赞同展开评论打赏