FlinkSql详解

Flink SQL 中的执⾏流程总共包含了几个步骤

1. ⭐ 第⼀步：获取数据

CREATE TEMPORARY TABLE kafka_user_path (
  `user` string,
  `url` string
  `cTime` BIGINT,
  `my_time` TIMESTAMP(3) METADATA FROM 'timestamp',
  `my_date` AS CAST(`my_time` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '****',
  'properties.group.id' = '***',
  'topic' = '***',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

2. ⭐ 第⼆步：处理数据

CREATE TEMPORARY VIEW view_user_path AS
SELECT user
	  ,count(url)  as cnt
FROM kafka_user_path
group by user;

select   user
	  	,tumble_end(cTime, interval '1' hours) as endT
	  	,count(url) as cnt 
from kafka_user_path 
group by user
	   	,tumble(cTime, interval '1' hours)

3. ⭐ 第三步：⽣成的动态结果表被转换回数据输出流。

        连续查询（Continuous Query）的输出结果表是⼀个 changelog。其可以像普通数据库表⼀样通过 INSERT、UPDATE 和 DELETE 来不断修改。它可能是⼀个只有⼀⾏、不断更新 changelog 表，也可能是⼀个 insert-only 的 changelog 表。

CREATE TEMPORARY TABLE `mysql_user_path` (
  `user` string COMMENT '用户',
  `endT` string COMMENT '时间窗口',
  `cnt` BIGINT COMMENT 'url个数',
  PRIMARY KEY (`user`,`endT`,`cnt`) NOT ENFORCED
) WITH (
  'connector' = 'rds',
  'userName' = '****',
  'tableName' = '***',
  'batchSize' = '4096',
  'url' = '*****'
);
 
BEGIN STATEMENT SET;
 
insert into mysql_user_path select user,endT,cnt from view_user_path;
 
END;

FlinkSql中输出流的分类

        动态表可像传统表一样被INSERT、UPDATE、DELETE修改。可能只有一行的表被持续更新；或者是没有UPDATE、DELETE更改的只插入表。当将动态表转化为流或将其写入外部系统，这些更改(修改)需要被编码，Flink的Table API & SQL支持三种方式编码动态表上的更改(修改)。

Append-only流:仅使用INSERT更改进行修改的动态表可通过发出插入的行来转化为流。

Retract流:Retract流包含两种类型消息(add消息和retract消息)，通过将动态表的INSERT更改作为add消息、将DELETE更改作为retract消息、将UPDATE更改分解为旧记录的retract消息和新记录的add消息。

Upsert流:Upsert流包含两种类型消息(upset消息和delete消息)，动态表转化为upsert流需要有主键(可复合)，具有主键的动态表通过将INSERT、UPDATE更改编码为upset消息，将DELETE更改编码为delete消息。upset流与retract流主要区别是UPDATE更改使用单一消息(主键)进行编码，因此效率更高,但要求外部系统支持主键。