前言
以前写Flink从kafka入hdfs因为业务需求和老版本缘故都是自定义BucketSink入动态目录中，对于简单的需求可以直接用Flink SQL API进行输出。Flink版本1.13.1。
Flink官网示例
准备
本地下载个kafka（单机即可），新建个桌面目录文件夹k2f。
输入输出源
按照建表有：
执行操作语句：
String opSql ="insert into fileOut select id,name,age,sum(score) from kafkaInput group by id";
报错如下，原因是这样数据是增量（不支持）,需要进行开窗：
Exception in thread "main" org.apache.flink.table.api.TableException: Table sink 'default_catalog.default_database.fileOut' doesn't support consuming update changes which is produced by node GroupAggregate(groupBy=[id], select=[id, SUM(score) AS EXPR$1])
开窗需要指定水印字段（这里我们采用kafka自动生成的eventTine时间戳<kafka0.10.1.0后>，除此之前外我们还能获取offset和partition等元数据信息水印相关具体可见：Flink事件时间和水印详解），指定字段eventTime为kafka元数据的timestamp，以及生成水印时间1s；
// 创建flink流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 1.3 基于Blink的流处理
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
//.inStreamingMode()
.build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);
String kafkaInput = "CREATE TABLE kafkaInput( \n" +
"id string, \n" +
"name string, \n" +
"eventTime TIMESTAMP(3) METADATA FROM 'timestamp' ,\n" +
"WATERMARK FOR eventTime AS eventTime - INTERVAL '1' SECOND,\n" +
"score BIGINT \n" +
")WITH( \n" +
"'connector' = 'kafka', \n" +
"'topic' = 'k2f_topic_1', \n" +
"'properties.bootstrap.servers' = '127.0.0.1:9092', \n" +
"'properties.group.id' = 'testGroup', \n" +
"'format' = 'csv' \n" +
")\n";
String fileOut = "CREATE TABLE fileOut( \n" +
"id string, \n" +
"score BIGINT, \n" +
"window_start TIMESTAMP(3),\n" +
"window_end TIMESTAMP(3)\n" +
") WITH ( \n" +
"'connector' = 'filesystem', \n" +
"'format' = 'csv',"+
"'path' = 'file:\\C:\\Users\\cbry\\Desktop\\k2f' \n" +
")";
user_action_time AS PROCTIME() – 声明一个额外的列作为处理时间属性，这个事件是系统计算的时间，不需要从我们的源头数据进行提供只需要声明
指定操作语句
前置数据准备好了，水印策略和字段也准备好了，开窗的窗口策略设置在操作语句，如下指定TUMBLE滚动窗口和
String opSql ="insert into fileOut select id,sum(score),window_start, window_end from TABLE(" +
"TUMBLE(TABLE kafkaInput, DESCRIPTOR(eventTime), INTERVAL '10' SECONDS)) group by window_start, window_end,id";
tableEnv.executeSql(kafkaInput);
TableResult outTable = tableEnv.executeSql(fileOut);
tableEnv.executeSql(opSql);
outTable.print();
env.execute("k2F");
若不指定水印报错：
Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. The window function TUMBLE(TABLE table_name, DESCRIPTOR(timecol), datetime interval) requires the timecol is a time attribute type, but is TIMESTAMP(3).
效果
kafka生产：
11020102,cbry,1
11020102,cbry,1
11020102,cbry,1
11020102,cbry,1
… 省略 …
11020102,cbry,100
11020102,cbry,1000
11020102,cbry,10000
11020102,cbry,1
11020102,cbry,1
11020102,cbry,1
11020102,cbry,1
… 省略 …
文件输出：
11020102,2,“2022-08-12 14:43:50”,“2022-08-12 14:44:00”
11020102,2,“2022-08-12 14:44:00”,“2022-08-12 14:44:10”
11020102,1,“2022-08-12 14:44:30”,“2022-08-12 14:44:40”
11020102,1,“2022-08-12 14:44:50”,“2022-08-12 14:45:00”
11020102,1,“2022-08-12 14:46:40”,“2022-08-12 14:46:50”
11020102,100,“2022-08-12 14:46:50”,“2022-08-12 14:47:00”
11020102,1000,“2022-08-12 14:47:10”,“2022-08-12 14:47:20”
11020102,10000,“2022-08-12 14:50:00”,“2022-08-12 14:50:10”
11020102,5,“2022-08-12 14:58:10”,“2022-08-12 14:58:20”
11020102,2,“2022-08-12 14:58:30”,“2022-08-12 14:58:40”
11020102,1,“2022-08-12 14:59:00”,“2022-08-12 14:59:10”
11020102,3,“2022-08-12 14:59:30”,“2022-08-12 14:59:40”
11020102,1,“2022-08-12 14:59:40”,“2022-08-12 14:59:50”
FlinkSQL的窗口类型
窗口函数TVF（table-valued functions）一共有三种：滚动(TUMBLE )、滑动（HOP）、累计（CUMULATE）：Flink官网SQLAPI窗口函数TVF。
简单描述下：窗口函数除去原始source表的所有列外，额外有三个列：“window_start”、“window_end”、“window_time”。其中window_time指的是窗口生成的时间属性列（窗口计算时间），且window_time总是等于window_end - 1ms。所以使用FlinkSQL开窗必须要source表中有时间字段`。
加入一个window_time到最后：
11020102,2,“2022-08-12 16:46:20”,“2022-08-12 16:46:30”,“2022-08-12 16:46:29.999”
11020102,1,“2022-08-12 16:46:30”,“2022-08-12 16:46:40”,“2022-08-12 16:46:39.999”
目前不支持单窗口函数，必须跟：
group by window_start, window_end, 。。。
一起使用聚合。
开窗规范（以滚动为例）
本质上上将窗口的结果封装称一张动态表。
TUMBLE(TABLE data, DESCRIPTOR(timecol), size , [（可选） offset ])
data: 数据原表；timecol:作为窗口的常规时间字段纳入窗口表中。size:开窗宽度（大小）offset:窗口开始的偏移量
timecol不太好理解，简单的说就是：窗口采用的时间类型的时间字段，据此滚动。比如说采用source表中的eventTime，那么开窗时间就是第一条数据的eventTime的值。
当使用EventTime的时候必须指定水印。所以在Flink事件时间和水印详解中，水印的时间字段和窗口时间字段保持一致，因为创建水印的时候指定了EvenTime在元数据中的字段。
举个例子：当前是2022-08-12 19:15:20 ，将eventTime免去kafka的时间戳关联：
"eventTime TIMESTAMP(3) ,\n" +
模拟数据：
关窗
在操作的过程中现象，要有新的数据才能刷新统计结果：新的数据：插入新的水印；
参考Flink窗口详解和各示例使用 触发窗口条件：
窗口Window会在以下的条件满足时被触发执行：
- watermark时间 >= window_end_time（闭窗）；
- 在[window_start_time,window_end_time)中有数据存在（入窗）；
所以说内存中始终有上一个窗口没关闭，而这个窗口需要等待新的数据关窗。
实际应用场景中：kafka写入生产多个分区的时候（kafka的topic多分区），如果写入对应的分区没有足够的数据量来触发窗口的闭合，会导致数据结果迟迟不出现和结果偏差（所以需要实时数据不断涌入），这个问题在测试的时候因为测试数据量不够导致查询了很久。。。mark一下。
阿里的blink：DATE_FORMAT
问题
kafka只能增量
Exception in thread “main” org.apache.flink.table.api.TableException: Table sink ‘default_catalog.default_database.sink_LLRZ’ doesn’t support consuming update changes which is produced by node GroupAggregate(groupBy=[host, app, stream, dev_name, server_ip, $f5, batch_time, to_date_time, id, window_time], select=[host, app, stream, dev_name, server_ip, $f5, batch_time, to_date_time, id, window_time, SUM(cash_out_flow) AS EXPR$1, SUM(total_duration) AS EXPR$2, MIN(start_time) AS EXPR$8, MAX(end_time) AS EXPR$9])
原因：需要group by
目前不支持单窗口函数，必须跟：
group by window_start, window_end, 。。。
一起使用聚合。
