Flink CDC 2.4 DataStream API配置增量快照禁用及快照查询覆盖问题

要让snapshot.select.statement.overrides生效，核心是正确关闭增量快照模式——注意scan.incremental.snapshot.enabled是Flink CDC自身的配置项，不能放在Debezium的Properties中，需要通过MySqlSourceBuilder的configuration()方法直接设置。

正确配置步骤及代码示例 #

修改你的代码，关键改动点如下：

1. 保留初始快照模式配置（你已通过StartupOptions.initial()完成）
2. 通过builder.configuration()添加Flink CDC的增量快照关闭配置
3. 保留Debezium的snapshot.select.statement.overrides相关配置

修改后的完整代码：

String dbName="db1";
String tableName="table11";
String query = "select * from db1.table11 where id='1' ";

Properties debeziumProperties = new Properties();
// 配置Debezium的快照查询覆盖规则
debeziumProperties.put("snapshot.select.statement.overrides", "db1.table11");
debeziumProperties.put("snapshot.select.statement.overrides.db1.table11", query);

String configFile =  "application-batch.conf";
ConfigParser.init(configFile);
ConfigParser.MySqlConfig config = ConfigParser.getMysqlConfig();
Map<String, Object> jsonConvertConfig = new HashMap<>();
jsonConvertConfig.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, DecimalFormat.NUMERIC.name());

MySqlSourceBuilder<String> builder = MySqlSource.<String>builder()
        .hostname(config.getHost())
        .port(config.getPort())
        .databaseList(dbName)
        .tableList(String.format("%s.%s", dbName, tableName))
        .username(config.getUserName())
        .password(config.getPassword())
        .debeziumProperties(debeziumProperties)
        // 关键：通过configuration设置Flink CDC的增量快照关闭属性
        .configuration(Map.of("scan.incremental.snapshot.enabled", "false"))
        .deserializer(new JsonDebeziumDeserializationSchema(true, jsonConvertConfig))
        .startupOptions(StartupOptions.initial());

注意事项 #

• scan.incremental.snapshot.enabled默认值为true，开启增量快照时会直接忽略snpshot.select.statement.overrides配置，必须关闭才能生效
• 确保查询语句query中的表名与tableList配置的完全一致（包含库名）
• 运行程序的MySQL账号需要具备对应表的查询权限，以及Debezium所需的binlog读取等权限

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