请问一下有人遇到flink cdc用mysql source重复消费的情况吗？

在使用 Flink CDC 时，确实可能会遇到 MySQL Source 表重复消费的问题。以下是可能导致该问题的原因及解决方案：

1. 重复消费的原因

Flink CDC 在正常操作环境下能够以 exactly-once 的语义投递每条变更事件。但在非正常情况下（例如作业故障或重启），CDC 工具（如 Debezium、Canal 或 Maxwell）只能保证 at-least-once 的语义。这可能导致以下情况： - 当作业从 Kafka 中消费数据时，可能会接收到重复的变更事件。 - 如果未对这些重复事件进行处理，可能会导致 Flink 查询结果错误或出现非预期异常。

2. 解决方案

为了避免重复消费问题，可以采取以下措施：

(1) 配置去重参数

建议将作业参数 table.exec.source.cdc-events-duplicate 设置为 true，并在源表上定义主键（PRIMARY KEY）。这样，Flink 系统会生成一个额外的有状态算子，利用主键对变更事件进行去重，并生成一个规范化的 changelog 流。

示例配置：

SET 'table.exec.source.cdc-events-duplicate' = 'true';

(2) 使用高效去重方案

实时计算 Flink 版提供了两种高效的去重方案： - 保留第一条（Deduplicate Keep FirstRow）：仅保留每个主键的第一条记录。 - 保留最后一条（Deduplicate Keep LastRow）：仅保留每个主键的最后一条记录。

根据业务需求选择合适的去重策略。

(3) 检查 Binlog 消费位置

在作业重启时，MySQL CDC 源表可以选择从配置的位置重新消费，或者从作业停止时的位置继续消费。如果选择了从最新状态恢复，则需要确保所需的 Binlog 文件未被清理，否则会导致重复消费或报错。

示例： - 全新启动：从配置的 Binlog 位点重新消费。 - 最新状态恢复：从作业停止时的 Binlog 位点继续消费。

(4) 开启 Source 复用

当作业中包含多个 MySQL CDC 源表时，开启 Source 复用功能可以减少数据库的压力，并避免重复拉取 Binlog 数据。通过以下命令启用 Source 复用：

SET 'table.optimizer.source-merge.enabled' = 'true';

注意：启用 Source 复用后，需无状态启动作业，因为复用会导致作业拓扑改变。

3. 其他注意事项

• Binlog 格式：确保 MySQL 的 Binlog 格式设置为 ROW，否则可能导致数据无法正确解析。
• Server ID 冲突：为每个 MySQL CDC 源表配置不同的 Server ID，避免因冲突导致的重复消费问题。
• Checkpoint 配置：合理设置 Checkpoint 间隔时间，避免因 Checkpoint 失败导致的数据重复写入问题。

总结

通过以上方法，您可以有效解决 Flink CDC 使用 MySQL Source 表时的重复消费问题。建议优先配置去重参数和开启 Source 复用功能，同时检查 Binlog 消费位置和 Server ID 配置，确保作业稳定运行。