FlinkSql开窗实例：消费kafka写入文本

前言

以前写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(TABLEdata, 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, 。。。

一起使用聚合。

标签： kafka 大数据 flink

本文转载自: https://blog.csdn.net/qq_37334150/article/details/128458087
版权归原作者 长臂人猿 所有，如有侵权，请联系我们删除。

FlinkSql开窗实例：消费kafka写入文本

前言

准备

输入输出源

指定操作语句

效果

FlinkSQL的窗口类型

开窗规范（以滚动为例）

关窗

问题

kafka只能增量

发表评论

“FlinkSql开窗实例：消费kafka写入文本”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航