Unable to display incremental data in Flink SQL 在Flink SQL中无法显示增量数据

[KTurnura]

Enviroment 环境

Operating System ： Mac OS

Mysql: 8.0.33

Java : zulu-8

hadoop: 3.2.2

flink: 1.16.1

Configuration 基本配置

1. Jar in $FLINK_HOME\lib Flink 所含外部lib包

   commons-cli-1.5.0.jar
   flink-connector-jdbc-1.16.0.jar
   flink-shaded-hadoop-3-3.1.1.7.2.9.0-173-9.0.jar
   flink-sql-connector-mysql-cdc-2.3.0.jar
   flink-table-api-java-uber-1.16.1.jar
   flink-table-planner-loader-1.16.1.jar
   flink-table-runtime-1.16.1.jar
   guava-27.0-jre.jar
   hudi-flink1.16-bundle-0.14.0-SNAPSHOT.jar
   hudi-hadoop-mr-bundle-0.14.0-SNAPSHOT.jar
   

2. flink checkpoint location(位置) hdfs://hostname:port/flink/checkpoints

3. mysql my.cnf setting 设置

   basedir=/Users/kturnura/Packages/mysql8
   default_authentication_plugin=mysql_native_password
   datadir=/Users/kturnura/Packages/mysql8/data
   
   server-id=2
   log-bin=mysql-bin
   binlog_format="ROW"
   expire_logs_days=15
   binlog_row_image=full
   binlog-do-db=flink_CDC_Hudi
   

4. mysql binlog test 测试

   mysql> show master logs;
   +------------------+-----------+-----------+
   | Log_name         | File_size | Encrypted |
   +------------------+-----------+-----------+
   | mysql-bin.000002 |       180 | No        |
   | mysql-bin.000003 |      7512 | No        |
   +------------------+-----------+-----------+
   2 rows in set (0.00 sec)
   
   mysql> insert into flink_CDC_Hudi.tbl_users (name) values ('wanger2');
   Query OK, 1 row affected (0.00 sec)
   
   mysql> show master logs;
   +------------------+-----------+-----------+
   | Log_name         | File_size | Encrypted |
   +------------------+-----------+-----------+
   | mysql-bin.000002 |       180 | No        |
   | mysql-bin.000003 |      7844 | No        |
   +------------------+-----------+-----------+
   2 rows in set (0.00 sec)
   

flink CDC query mysql Data 查询mysql数据

start Flink SQL

export HADOOP_CLASSPATH=`hadoop classpath`
-- start Flik SQL  启动 Flink SQL
bin/sql-client.sh embedded

-- 在flink sql  内设置属性
SET  'execution.checkpointing.interval' = '3s';
SET 'sql-client.execution.result-mode' = 'tableau';

-- 建表
create table users_source_mysql(
	id BIGINT primary key not enforced,
	name string,
	birthday timestamp(3),
	ts timestamp(3)
)with(
	'connector' = 'mysql-cdc',
	'hostname' = 'localhost',
	'port' = '3306',
	'username' = 'root',
	'password' = 'password',
	'database-name'='flink_CDC_Hudi',
	'table-name'='tbl_users',
	'server-id'='2'
);

-- 查询
select * from users_source_mysql;

When the data was finally queried, the incremental data could not be queried. During the query process of Flink, mysql was added and deleted, but the changed data was not successfully captured. Insert data, Flink Web Logs run log

最终查询数据时，并不能查询增量数据，在Flink 运行 该查询过程中，对mysql 进行增删，没有成功捕获变更数据。插入数据，Flink Web Logs 运行日志

2023-09-09 16:22:16,848 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Marking checkpoint 25 as completed for source Source: users_source_mysql[9].
2023-09-09 16:22:19,396 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Triggering checkpoint 26 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1694247739395 for job 5962eee782a28935494f859ea08245a5.
2023-09-09 16:22:19,852 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Completed checkpoint 26 for job 5962eee782a28935494f859ea08245a5 (3607 bytes, checkpointDuration=457 ms, finalizationTime=0 ms).
2023-09-09 16:22:19,856 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Marking checkpoint 26 as completed for source Source: users_source_mysql[9].
2023-09-09 16:22:22,394 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Triggering checkpoint 27 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1694247742393 for job 5962eee782a28935494f859ea08245a5.
2023-09-09 16:22:22,835 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Completed checkpoint 27 for job 5962eee782a28935494f859ea08245a5 (3607 bytes, checkpointDuration=441 ms, finalizationTime=1 ms).
2023-09-09 16:22:22,835 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Marking checkpoint 27 as completed for source Source: users_source_mysql[9].

Expect performance 期待表现

Flink SQL> select * from users_source_mysql;

op	id	name	birthday	ts
+	1	test1	time1	time1
+	2	test2	time2	Time2

Delete data in mysql

mysql>delete from users_source_mysql where id = 1;

Expect Result

op	id	name	birthday	ts
+	1	test1	time1	time1
+	2	test2	time2	time2
-	1	test1	time3	Time

However, our actual effect is that we cannot see our changed data records in real time during the initial query execution.
When the last query execution is closed and the same query statement is reopened, the result will become:

然而我们实际效果确实并不能实时在最开始的查询执行中看到我们的变更数据记录，
当新关闭上个查询执行，重新开启一个相同的查询语句，结果就会变成：

op	id	name	birthday	ts
+	2	test2	time2	time2