Flink读取mysql cdc有办法过滤掉 update_before类型的数据嘛？

在Apache Flink中，当读取MySQL CDC时，可以通过配置选项或编程方式来过滤掉update_before类型的数据。

1. 使用Flink SQL：
   如果您正在使用Flink SQL，可以在创建表源时指定只处理特定的变更事件类型。例如：

   CREATE TABLE my_table (
     ...
   ) WITH (
     'connector' = 'mysql-cdc',
     'hostname' = '<hostname>',
     'port' = '<port>',
     'username' = '<username>',
     'password' = '<password>',
     'database-name' = '<database>',
     'table-name' = '<table>',
     'server-id' = '<server_id>',
     -- 只处理INSERT和UPDATE_AFTER类型的事件
     'include-events' = 'insert,update_after'
   );
   

2. 编程方式：
   如果您正在使用Java/Scala API进行编程，可以自定义一个ChangelogMode，并将其传递给TableSource的构造函数。例如：

   ChangelogMode changelogMode = ChangelogMode.newBuilder()
       .addContainedKind(Kind.INSERT)
       .addContainedKind(Kind.UPDATE_AFTER)
       .build();
   
   TableSource<?> source = ...;
   source.configure(
       new Configuration(),
       new HashMap<>(),
       changelogMode);
   

没有办法。此回答整理自钉群“实时计算Flink产品交流群”