1. Johngo学长首页
  2. 技术点详细复盘专题
  3. 大数据

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)

• 大数据 • 阅读 41
1、查询语句
1、hint

在对表进行查询的是偶动态修改表的属性

-- 创建表
CREATE TABLE word (
    lines STRING
)
WITH (
  'connector' = 'kafka',
  'topic' = 'word',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',-- 读取所有的数据
  'format' = 'csv',
  'csv.field-delimiter'='\t'
)
-- 加载hive函数
LOAD MODULE hive WITH ('hive-version' = '1.2.1');
--统计单词的数量
--不动态指定开始读取的参数
select word,count(1) from
word,
lateral table(explode(split(lines,','))) as t(word)
group by word

-- OPTIONS 动态指定参数
select word,count(1) from
word /*+ OPTIONS('scan.startup.mode'='latest-offset') */ ,
lateral table(explode(split(lines,','))) as t(word)
group by word

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
3、WITH
-- temp可以在后面的sql中使用多次
with temp as (
    select word from word,
    lateral table(explode(split(lines,','))) as t(word)
)
select * from  temp
 union all
select * from  temp

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
4、SELECT
SELECT order_id, price
FROM
(VALUES (1, 2.0), (2, 3.1))  AS t (order_id, price)

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
5、分组窗口聚合

老版本语法，新版本中不推荐使用

-- PROCTIME(): 获取处理时间的函数
CREATE TABLE words_window (
    lines STRING,
    proc_time as PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'words',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',-- 读取所有的数据
  'format' = 'csv',
  'csv.field-delimiter'='\t'
)

-- TUMBLE：滚动窗口
-- HOP": 滑动黄口
-- SESSION： 会话窗口

--TUMBLE：处理时间的滑动窗口
select
word,
TUMBLE_START(proc_time, INTERVAL '5' SECOND)  as s, -- 窗口开始时间
TUMBLE_END(proc_time, INTERVAL '5' SECOND) as e, -- 窗口开始使时间
count(1) as c
from
words_window,
lateral table(explode(split(lines,','))) as t(word)
group by
word,
TUMBLE(proc_time, INTERVAL '5' SECOND) -- 每5秒计算一次

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
  • 会话窗口

    一段时间没有数据开始计算 暂时只能在老板本api中使用

CREATE TABLE words_window (
    lines STRING,
    proc_time as PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'words',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',-- 读取所有的数据
  'format' = 'csv',
  'csv.field-delimiter'='\t'
)
select
word,
SESSION_START(proc_time, INTERVAL '5' SECOND)  as s, -- 窗口开始时间
SESSION_END(proc_time, INTERVAL '5' SECOND) as e, -- 窗口结束使时间
count(1) as c
from
words_window,
lateral table(explode(split(lines,','))) as t(word)
group by
word,
SESSION(proc_time, INTERVAL '5' SECOND) -- 会话超过5秒中没有发送消息，就开始进行计算

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
6、TVFs(重点)
  • 滚动窗口函数
CREATE TABLE words_window (
    lines STRING,
    proc_time as PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'words',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',-- 读取所有的数据
  'format' = 'csv',
  'csv.field-delimiter'='\t'
)

-- TUMBLE(TABLE words_window, DESCRIPTOR(proc_time), INTERVAL '5' SECOND)
-- TUMBLE: 窗口函数，可以给原表增加床i偶开始时间，窗口的结束时间，窗口时间
-- TABLE words_window ： 指定原表
-- DESCRIPTOR(proc_time) 指定时间字段，可以处理时间，也可以是事件时间
-- INTERVAL '5' SECOND 指定窗口大小

 SELECT lines,proc_time,window_start,window_end,window_time FROM TABLE(
  TUMBLE(TABLE words_window, DESCRIPTOR(proc_time), INTERVAL '5' SECOND)
 );

 -- 在划分和窗口之后进行聚合计算
 SELECT word,window_start,count(1) as c FROM
 TABLE(
  TUMBLE(TABLE words_window, DESCRIPTOR(proc_time), INTERVAL '5' SECOND)
 ),
 lateral table(explode(split(lines,','))) as t(word)
 group by word,window_start
  • 滑动窗口函数

    一条数据会出现在多个窗口中，所以输入一条数据，会输出多条数据

CREATE TABLE words_window (
    lines STRING,
    proc_time as PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'words',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',-- 读取所有的数据
  'format' = 'csv',
  'csv.field-delimiter'='\t'
)
-- HOP： 滑动窗口函数，需要指定窗口大小和滑动时间
-- 输入一条数据会输出多条数据
with temp as (
select * from words_window /*+ OPTIONS('scan.startup.mode'='latest-offset') */
)
SELECT * FROM
TABLE(
    HOP(TABLE temp , DESCRIPTOR(proc_time), INTERVAL '5' SECOND, INTERVAL '15' SECOND)
)
;

-- 窗口止呕进行聚合
with temp as (
select * from words_window /*+ OPTIONS('scan.startup.mode'='latest-offset') */
)
SELECT word ,window_start,count(1) as c FROM
TABLE(
    HOP(TABLE temp, DESCRIPTOR(proc_time), INTERVAL '5' SECOND, INTERVAL '15' SECOND)),
lateral table(explode(split(lines,','))) as t(word)
group by word,window_start
;

7、时间属性

1、处理时间

使用PROCTIME()函数给表增加一个时间字段

CREATE TABLE student_kafka_proc_time (
    id STRING,
    name STRING,
    age INT,
    gender STRING,
    clazz STRING,
    proc as PROCTIME() -- 处理时间字段
) WITH (
  'connector' = 'kafka',
  'topic' = 'student',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv',
  'csv.field-delimiter'=',', -- csv格式数据的分隔符
  'csv.ignore-parse-errors'='true', -- 如果出现脏数据据,补null
  'csv.allow-comments'='true'--跳过#注释行
)

-- 使用处理时间可以做窗口统计
 SELECT clazz,window_start,count(1) as c FROM
 TABLE(
  TUMBLE(TABLE student_kafka_proc_time, DESCRIPTOR(proc), INTERVAL '5' SECOND)
 )
 group by clazz,window_start

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
2、事件时间
  • 测试数据
1500100001,施笑槐,22,女,文科六班,2022-07-20 16:44:10
1500100001,施笑槐,22,女,文科六班,2022-07-20 16:44:11
1500100001,施笑槐,22,女,文科六班,2022-07-20 16:44:12
1500100001,施笑槐,22,女,文科六班,2022-07-20 16:44:20
1500100001,施笑槐,22,女,文科六班,2022-07-20 16:44:15
1500100001,施笑槐,22,女,文科六班,2022-07-20 16:44:25
  • 创建表指定时间字段和水位线
-- TIMESTAMP(3) flink的时间戳类型
-- ts - INTERVAL '5' SECOND 水位线前移5秒
CREATE TABLE student_kafka_event_time (
    id STRING,
    name STRING,
    age INT,
    gender STRING,
    clazz STRING,
    ts TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND -- 指定时间字段和水位线
) WITH (
  'connector' = 'kafka',
  'topic' = 'student_event_time',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

-- 使用事件时间  做窗口函数统计
-- 每一条数据都会计算出一个结果，会取更新之前已经输出的结果
-- 不存在数据丢失问题
-- 需要将统计结果保存在状态中
 SELECT clazz,window_start,count(1) as c FROM
 TABLE(
  TUMBLE(TABLE student_kafka_event_time, DESCRIPTOR(ts), INTERVAL '5' SECOND)
 )
 group by clazz,window_start

-- 分钟窗口统计
-- 如果数据乱序可能会丢失数据
-- 不需要将统计的结果保存在状态中
select
clazz,
TUMBLE_START(ts, INTERVAL '5' SECOND)  as s, -- 窗口开始时间
TUMBLE_END(ts, INTERVAL '5' SECOND) as e, -- 窗口开始使时间
count(1) as c
from
student_kafka_event_time
group by
clazz,
TUMBLE(ts, INTERVAL '5' SECOND) -- 没4秒计算一次

 -- 生产数据
 kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic student_event_time

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)
练习

统计单词的数量，
每隔5秒统计一次
每个窗口中取单词数量最多个两个单词

CREATE TABLE words_window_demo (
    lines STRING,
    proc_time as PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'words',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',-- 读取所有的数据
  'format' = 'csv',
  'csv.field-delimiter'='\t'
)
-- 在夫林卡 sql 流处理中row_number()必须要取topN
select * from (
    select
    word,
    window_start,
    c,
    row_number() over(partition by window_start order by c desc) as r
    from (
        select  word,window_start,count(1) as c from
        TABLE(
            TUMBLE(TABLE words_window_demo, DESCRIPTOR(proc_time), INTERVAL '5' SECOND)
        ),
        lateral table(explode(split(lines,','))) as t(word)
        group by word,window_start
    ) as a
) as b
where r
  • 统计每个城市中每个区县的车流量
  • 每隔5分钟统计一次，统计最近15分钟的数据
  • 每个城市中取车流量最大的前2个区县
  • 将统计好的结果保存到数据库中
-- 数据
{
    "car": "皖AK0H90",
    "city_code": "340100",
    "county_code": "340111",
    "card": 117303031813010,
    "camera_id": "00004",
    "orientation": "北",
    "road_id": 34130440,
    "time": 1614799929,
    "speed": 84.51
}

-- TIMESTAMP(3) flink的时间戳类型
-- ts - INTERVAL '5' SECOND 水位线前移5秒
-- 创建表读取kafka中的json数据
CREATE TABLE cars_kafka_event_time (
    car STRING,
    city_code STRING,
    county_code STRING,
    card BIGINT,
    camera_id STRING,
    orientation STRING,
    road_id BIGINT,
    time BIGINT,
    speed DOUBLE,
    ts_ltz AS TO_TIMESTAMP_LTZ(time, 3),
    WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND -- 指定时间字段和水位线
) WITH (
  'connector' = 'kafka',
  'topic' = 'car_test',
  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
  'properties.group.id' = 'carGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
)
-- 测试一下是否存在数据
select * from  cars_kafka_event_time

--  统计每个城市中每个区县的车流量,每隔5分钟统计一次，统计最近15分钟的数据,每个城市中取车流量最大的前2个区县
select *
from (
select
    county_code
    ,city_code
    ,window_start
    , c
    ,row_number() over(partition by window_start order by c desc) as r
    from
(
with temp as (
select * from cars_kafka_event_time  /*+ OPTIONS('scan.startup.mode'='latest-offset') */
)
SELECT
    county_code
    ,city_code
    ,window_start
    ,count(1) as c
    FROM
TABLE(
    HOP(TABLE temp, DESCRIPTOR(ts_ltz), INTERVAL '5' SECOND, INTERVAL '15' SECOND))
group by county_code,city_code,window_start
) as b ) as h
where r

Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)

Original: https://blog.csdn.net/weixin_48370579/article/details/126091927
Author: a-tao必须奥利给
Title: Flink SQl 语法(hint,with,select,分组窗口聚合，时间属性（处理，事件）)

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/817859/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球