FlinkSQL学习笔记（三）常用连接器举例

写在前面

1、了解upsert kafka、JDBC、FileSystem、等连接器的使用场景，使用过程核心要点，

2、掌握cdc连接器的使用要点，完成源码编译过程；

3、问题：在flinkSQL中，端到端的一致性是通过什么方式保证的？

答：

1、upsert kafka本身是为了解决flink向kafka写数据是append-only的模式【只有+I这一种Changemode】，使用过程中注意需要制定key，作为update或者delete的标志

2、JDBC的连接器使用过程中注意source和sink的分别对应的两种不同模式，关键还是在于是否通过有key来做区分

3、通过ChangeLogStream的记录状态变化 ，参考链接：https://zhuanlan.zhihu.com/p/399739094

1、upsert-kafka连接器

1.1、背景

Upsert Kafka 连接器支持以 upsert 方式从 Kafka topic 中读取数据并将数据写入 Kafka topic。

作为 source，upsert-kafka 连接器生产 changelog 流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果有这个 key（如果不存在相应的 key，则该更新被视为 INSERT）。用表来类比，changelog 流中的数据记录被解释为 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同 key 的现有行都被覆盖。另外，value 为空的消息将会被视作为 DELETE 消息。

作为 sink，upsert-kafka 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。

1.3、group by条件下的【+I、-U、+U】

1. 产生socket流

DataStreamSource s1 = env.socketTextStream("123.56.100.37", 9999);
SingleOutputStreamOperator bean1 = s1.map(s -> { String[] arr = s.split(",");
    return new Bean1(Integer.parseInt(arr[0]), arr[1]);
});

2. socket流转表

tenv.createTemporaryView("bean1",bean1);

3. 创建upsert-kafka表

CREATE TABLE t_upsert_kafka (	
--	这里只能NOT ENFORCED，不像MySQL中可以进行强制主键					
  gender STRING PRIMARY KEY NOT ENFORCED,                  
  cnt BIGINT
) WITH (    
  'connector' = 'upsert-kafka',                            
  'topic' = 'upsert_kafka2',                               
  'properties.bootstrap.servers' = '123.56.100.37:9092',   
  'key.format' = 'csv',                                    
  'value.format' = 'csv'                                   
); 

4. 写入upsert-kafka

INSERT INTO t_upsert_kafka
SELECT gender,count(1) as cnt FROM bean1 GROUP BY gender

5. 执行结果

1.3、join条件下的【+I、-D、+I】

1. 产生socket流

//1,zs,18
DataStreamSource s1 = env.socketTextStream("123.56.100.37", 9999);
SingleOutputStreamOperator bean1 = s1.map(s -> { String[] arr = s.split(",");
    return new Bean1(Integer.parseInt(arr[0]),arr[1],Integer.parseInt(arr[2]));
});
//1,beijing,1388888
DataStreamSource s2 = env.socketTextStream("123.56.100.37", 9998);
SingleOutputStreamOperator bean2 = s2.map(s -> { String[] arr = s.split(",");
    return new Bean2(Integer.parseInt(arr[0]),arr[1],Integer.parseInt(arr[2]));
});

2. socket流转表

tenv.createTemporaryView("bean1",bean1);
tenv.createTemporaryView("bean2",bean2);

3. 执行建表语句

CREATE TABLE t_upsert_kafka_join3 (						
  id     INT PRIMARY KEY NOT ENFORCED,                  
  name   STRING,                                        
  age    INT,                                           
  addr   STRING,                                        
  phone  BIGINT                                         
) WITH ( 
  'connector' = 'upsert-kafka',                         
  'topic' = 'upsert_kafka2_join',                       
  'properties.bootstrap.servers' = '123.56.100.37:9092',
  'key.format' = 'csv',                                 
  'value.format' = 'json'                               
) 

4. 执行join语句插入

INSERT INTO t_upsert_kafka_join3		    
SELECT A1.id,A1.name,A1.age,A2.addr,A2.phone
  FROM bean1 AS A1                          
  LEFT JOIN bean2 AS A2                     
    ON A1.id = A2.id 

5. 查看结果

tenv.executeSql("SELECT * FROM t_upsert_kafka_join3").print();

2、Jdbc连接器

JDBC 连接器允许使用 JDBC 驱动向任意类型的关系型数据库读取或者写入数据。如果在 DDL 中定义了主键，JDBC sink 将以 upsert 模式与外部系统交换 UPDATE/DELETE 消息；否则，它将以 append 模式与外部系统交换消息且不支持消费 UPDATE/DELETE 消息。Jdbc作为source的时候，支持scan模式和lookup模式，look模式的意思参考lookup join。

2.1、Scan Source

CREATE TABLE flink_users (									
  id int primary key,                                       
  name string,                                              
  age int,   
  gender string                                             
) WITH (     
   'connector' = 'jdbc',                                    
   'url' = 'jdbc:mysql://123.56.100.37:3306/flinktest?useSSL=false', 
   'table-name' = 'users',                                  
   'username' = 'root',                                     
   'password' = '123456'                                    
) 

相当于有界流，只scan一次

2.2、Sink: Streaming Append & Upsert Mode

INSERT INTO flink_users		         
SELECT A1.id,A1.name,A1.age,A2.gender
  FROM bean1 AS A1                   
  LEFT JOIN bean2 AS A2              
    ON A1.id = A2.id 

查询结果，select的之后只scan一次

3、CDC连接器

适用于 Apache Flink 的 CDC 连接器是一组适用于 Apache Flink 的源连接器，使用更改数据捕获 （CDC） 从不同的数据库引入更改。 适用于 Apache Flink 的 CDC 连接器集成了 Debezium 作为捕获数据更改的引擎。所以它可以充分利用Debezium的能力。

github地址：https://github.com/ververica/flink-cdc-connectors?tab=readme-ov-file

3.1、Flink-CDC源码编译

1. 下载源码

   https://github.com/apache/flink-cdc/blob/release-2.0/pom.xml

   选择合适的版本、文章中用的事1.14.4，这里选择1.13.1版本

2. 在idea中，通过project from version control导入maven项目，源码修改如下：

   ①：删除包括测试等不用的项目，避免编译多余的项目

   ②：修改flink版本和Scala版本，为了匹配我们现在使用的flink版本

   ③：修改项目的groupId和version，定义新的flink版本

   这一步其实就是相当于修改源码，实际编程过程中，遇到很多细节性的错误，这里不做列举

3. 编译源码，打包

   这里直接在idea中选择install，从而可以成功的安装到本地

3.2 、mysql开启binlog

1. 修改配置文件/etc/mysql/my.cnf

   github参考连接：https://github.com/apache/flink-cdc/blob/master/docs/content/docs/connectors/cdc-connectors/mysql-cdc.md

   [mysqld]：https://stackoverflow.com/questions/44298071/how-to-fix-mysql-error-found-option-without-preceding-group-in-config-file

2. 查看binlog是否开启

show variables like 'log_%';

3. 查看master状态，即binlog的状态

show master status;

4. 建表后查看binlog信息

show binlog events in 'mysql-bin.000001';

3.3、mysql cdc连接器测试

使用方式参考连接：https://github.com/apache/flink-cdc/blob/master/docs/content/docs/connectors/cdc-connectors/mysql-cdc.md

1. 开启Checkpoint

env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointStorage("file:///c:/checkpoint");

2. 创建SQL表

CREATE TABLE flink_score (
 id INT,
 name STRING,
 gender STRING,
 score DOUBLE,
 PRIMARY KEY(id) NOT ENFORCED
) WITH (
 'connector' = 'mysql-cdc',
 'hostname' = '123.56.100.37',
 'port' = '3306',
 'username' = 'root',
 'password' = '123456',
 'database-name' = 'flinktest',
 'table-name' = 'score'
)

3. 测试是否可以查询到MySQL更新得到的数据

4. 测试复杂的SQL逻辑—分组topN

SELECT gender
     , name
	 , score_amt
	 , rn
FROM (
	 SELECT gender
		  , name
		  , score_amt
		  , row_number() over(partition by gender order by score_amt) as rn
	   FROM (
		    SELECT gender
				 , name
				 , sum(score) as score_amt
			  FROM flink_score
			 GROUP BY gender,name
	   )O1
     
	)O2
WHERE rn<=2