[大数据]pyflink滑动窗口实例

写在前头：

更多大数据相关精彩内容请进我的知识星球，每周定期更新

?正篇

技术路线：模拟kafka生产者发送数据——>flink对kafka数据实时计算处理——>处理后的数据发送到kafka

1、模拟客流数据的生产者，参考2021-08-19，flink:kafka——>mysql

2、flink聚合操作原理介绍，参考2021-08-25，flink窗口函数原理深入

滑动窗口：不同于滚动窗口，滑动窗口的窗口可以重叠

滑动窗口有两个参数：slide和size。slide为每次滑动的步长，size为窗口的大小。

• slide < size，则窗口会重叠，每个元素会被分配到多个窗口。
• slide = size，则等同于滚动窗口（TUMBLE）。
• slide > size，则为跳跃窗口，窗口之间不重叠且有间隙。

参考自：https://help.aliyun.com/document_detail/62512.html

滑动窗口示例：

以客流数据为例，20s为一个窗口，10s为一个步长，计算该事件窗口内的客流数和

kafka中数据样式为：

{'stationid':1, 'stationname':'车站1', 'time':'2021-08-01 00:00:00', 'equipmentid':10, 'num':10, 'inorout':0}

这里滑动窗口演示代码如下：

# -*- coding: UTF-8 -*-
import os
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings


# 创建Table Environment， 并选择使用的Planner
env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(
   env,
   environment_settings=EnvironmentSettings.new_instance().use_blink_planner().build())

sourceKafkaDdl = """
    create table SourceKafka(
        stationId int,
        stationName varchar,
        CreateTime varchar,
        rt as TO_TIMESTAMP(CreateTime, 'yyyyMMdd HH:mm:ss'),
        equipmentId int,
        passengerNum int,
        inOrOut int,
        WATERMARK FOR rt as rt - INTERVAL '10' SECOND
    )
    with(
        'connector.type' = 'kafka',
        'connector.version' = 'universal',
        'connector.topic' = 'passenger-flow2',
        'connector.properties.zookeeper.connect' = 'tdh09:2181',
        'connector.properties.bootstrap.servers' = 'tdh09:9092',
        'connector.startup-mode' = 'latest-offset',
        'format.type' = 'json',
        'format.ignore-parse-errors' = 'true'
    )
    """

KafkaSinkDdl = """
   CREATE TABLE SinkKafka (
    stationId INT ,
    stationName VARCHAR ,
    passengerNumSum INT,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3)
) WITH (
    'connector.type' = 'kafka',
    'connector.version' = 'universal',
    'connector.topic' = 'passenger-flow3',
    'connector.properties.zookeeper.connect' = 'tdh09:2181',
    'connector.properties.bootstrap.servers' = 'tdh09:9092',
    'connector.startup-mode' = 'latest-offset',
    'format.type' = 'json',
    'format.ignore-parse-errors' = 'true'
)
"""

# 创建Kafka数据源表
t_env.execute_sql(sourceKafkaDdl)
# 创建MySql结果表
t_env.execute_sql(KafkaSinkDdl)

query = """
select stationId, stationName , sum(passengerNum) as passengerNumSum, HOP_START(rt, INTERVAL '10' SECOND, INTERVAL '20' SECOND), HOP_END(rt, INTERVAL '10' SECOND, INTERVAL '20' SECOND)
from SourceKafka
group by stationId, stationName, HOP(rt, INTERVAL '10' SECOND, INTERVAL '20' SECOND)
"""


t_env.sql_query(query=query).insert_into("SinkKafka")
t_env.execute("pyFlink_mysql")

处理前数据：

处理后数据：