flink cdc 相应的数据丢失怎么办？

展开

收起

真的很搞笑 2023-05-28 20:48:12 218 版权

阿里云 AI 助理回答

在使用Flink CDC时，如果出现数据丢失问题，可以从以下几个方面进行排查和解决：

1. 确认维表缓存策略是否有问题

维表的缓存策略可能会影响数据匹配的准确性。如果使用了Cache All缓存策略，并且启用了SHUFFLE_HASH、REPLICATED_SHUFFLE_HASH或SKEW策略，可能会导致部分连接器（如MaxCompute、OTS）无法正确匹配维表数据。

解决方案： - 检查维表的缓存策略是否合理。 - 如果问题存在，尝试调整缓存策略或优化数据匹配逻辑。

2. 检查数据乱序问题

如果作业中存在乱序数据，这些数据的Watermark可能不在新窗口的开窗和关窗时间范围内，从而被系统认为是迟到数据并丢弃。例如，11秒的数据在16秒进入15~20秒的窗口，其Watermark为11，会被丢弃。

解决方案： - 使用Print Sink或Log4j方式确认数据源中是否存在乱序数据。 - 根据乱序的程度，合理设置Watermark生成策略。例如，可以定义Watermark = Event time - 5s，以延迟触发窗口计算，确保乱序数据能够被正确处理。 - 建议以整天整时整分开窗口求聚合，避免因乱序严重而导致数据丢失。

3. 检查全量转增量阶段的数据丢失

在MySQL CDC从全量读取切换到增量读取时，如果Checkpoint间隔时间设置过大（如20分钟），可能导致增量数据同步延迟，甚至出现数据丢失的情况。

解决方案： - 根据业务需求，合理设置Checkpoint间隔时间，确保全量数据写入下游后再开始增量读取。 - 如果使用的是RDS MySQL 5.6只读实例，注意该类型实例的Binlog文件不包含数据，无法读取增量阶段的数据。建议使用可写实例或升级至更高版本。

4. 检查网络连通性和带宽消耗

Flink CDC通过Debezium或Flink CDC连接器过滤特定表的变更记录，但Binlog本身是整个实例级别的，会记录所有数据库和表的变更。这可能导致即使数据量不大，也会消耗大量带宽。

解决方案： - 开启Source复用功能，减少Binlog连接数，降低带宽消耗。可以通过以下命令启用：

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

注意：开启Source复用后，需要无状态启动作业，因为作业拓扑会发生变化。

5. 检查时间戳字段的时区问题

在增量阶段读取的timestamp字段可能出现时区相差8小时的问题，原因是CDC作业中配置的server-time-zone参数与MySQL服务器时区不一致。

解决方案： - 确保CDC作业中配置的server-time-zone参数与MySQL服务器时区一致。 - 在DataStream中使用自定义序列化器时，明确指定serverTimeZone。

6. 检查Binlog过滤器配置

如果某个库的表无法同步增量数据，而其他库可以，可能是MySQL服务器配置了Binlog过滤器，忽略了某些库的Binlog。

解决方案： - 使用以下命令检查Binlog过滤器配置：

SHOW MASTER STATUS;

如果发现Binlog_Ignore_DB或Binlog_Do_DB配置有问题，调整MySQL服务器的Binlog过滤器设置。

7. 检查作业失败后的恢复机制

在全量读取阶段，如果作业失败，可能导致数据丢失。对于MongoDB CDC，可以通过配置scan.incremental.snapshot.enabled='true'参数，从Checkpoint恢复读取数据。

解决方案： - 确保作业启用了Checkpoint机制，并合理设置Checkpoint间隔时间。 - 对于支持增量快照的连接器（如MongoDB CDC），启用scan.incremental.snapshot.enabled参数。

8. 检查权限问题

如果Flink作业中涉及的表分布在不同数据库，且使用的账号没有某些数据库的权限，可能导致数据丢失。

解决方案： - 检查SQL逻辑是否存在retract相关操作。如果存在，确保结果表的操作用户具有DELETE权限。 - 为账号添加作业中所有数据库的对应权限。

9. 检查DDL事件的捕获

如果需要捕获数据库中的DDL事件，可以使用社区版CDC连接器，并通过DataStream API配置includeSchemaChanges(true)参数。

解决方案： - 配置示例：

MySqlSource<xxx> mySqlSource =
    MySqlSource.<xxx>builder()
        .hostname(...)
        .port(...)
        .databaseList("<databaseName>")
        .tableList("<databaseName>.<tableName>")
        .username(...)
        .password(...)
        .serverId(...)
        .deserializer(...)
        .includeSchemaChanges(true) // 配置获取DDL事件的参数
        .build();