当前位置：首页 > news >正文

pyflink的窗口

news 来源：原创 2024/9/20 4:44:58

PyFlink 中的窗口操作教程

在流处理应用中，窗口（Window）是一个非常重要的概念，它用于对无界的数据流进行切分，使得我们可以对流中的数据执行聚合、计数、排序等操作。PyFlink 提供了丰富的窗口类型和操作，可以对流数据进行时间和计数等维度的切片，进行实时的数据处理。

在本教程中，我们将介绍 PyFlink 中的几种常见窗口类型，并展示如何使用窗口进行数据处理。

1. 安装 PyFlink

在开始之前，确保你已经安装了 PyFlink：

pip install apache-flink

2. 什么是窗口？

窗口（Window）是 Flink 处理无界数据流的核心技术，它将无限的数据流划分为有限的块，这样可以对这些块进行聚合、计数等操作。常见的窗口类型包括：

滚动窗口（Tumbling Window）：将数据流划分为不重叠的固定长度时间段。
滑动窗口（Sliding Window）：将数据流划分为固定长度的时间段，这些时间段可以相互重叠。
会话窗口（Session Window）：基于数据的活动时间来划分数据流，窗口之间有间隔（即活动的间歇）。
计数窗口（Count Window）：基于事件的数量而非时间划分窗口。

3. PyFlink 中的窗口操作

在 PyFlink 中，窗口通常和时间、事件一起使用，通过对数据流应用窗口函数来执行聚合操作。以下是几种常见的窗口操作。

4. 流环境设置

在 PyFlink 中，窗口操作通常在流模式下进行。首先，我们需要设置流环境并定义一些基础数据流。

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings# 创建流执行环境
env = StreamExecutionEnvironment.get_execution_environment()# 创建 Table 环境
settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=settings)

5. 时间特性设置

时间特性分为两种类型：事件时间（Event Time）和 处理时间（Processing Time）。事件时间基于事件生成时的时间，而处理时间基于 Flink 系统处理事件的时间。

设置事件时间（Event Time）

事件时间需要通过在数据流中添加时间戳和水印（Watermark）来支持。

# 设置事件时间属性
t_env.get_config().set_local_timezone('UTC')  # 使用 UTC 时区

6. 创建窗口

6.1 滚动窗口（Tumbling Window）

滚动窗口会将数据流划分为固定长度的时间段，并且这些时间段互不重叠。

from pyflink.table.window import Tumble
from pyflink.table import expressions as expr# 创建示例表
t_env.execute_sql("""CREATE TEMPORARY TABLE source_table (user_id STRING,item STRING,amount DOUBLE,event_time TIMESTAMP(3),WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND) WITH ('connector' = 'datagen')
""")# 定义滚动窗口，窗口大小为10分钟
result_table = t_env.from_path("source_table") \.window(Tumble.over(expr.lit(10).minutes).on(expr.col("event_time")).alias("w")) \.group_by(expr.col("w"), expr.col("user_id")) \.select(expr.col("user_id"), expr.col("amount").sum.alias("total_spent"), expr.col("w").end.alias("window_end"))# 输出查询结果
result_table.execute().print()

6.2 滑动窗口（Sliding Window）

滑动窗口将数据划分为固定长度的时间段，这些时间段可以相互重叠。窗口的滑动步长定义了相邻窗口的开始时间。

from pyflink.table.window import Slide# 定义滑动窗口，窗口大小为10分钟，滑动步长为5分钟
result_table = t_env.from_path("source_table") \.window(Slide.over(expr.lit(10).minutes).every(expr.lit(5).minutes).on(expr.col("event_time")).alias("w")) \.group_by(expr.col("w"), expr.col("user_id")) \.select(expr.col("user_id"), expr.col("amount").sum.alias("total_spent"), expr.col("w").end.alias("window_end"))# 输出查询结果
result_table.execute().print()

6.3 会话窗口（Session Window）

会话窗口基于数据的活动时间和不活动时间来划分数据流。如果一段时间内没有新的事件到达，窗口会结束。

from pyflink.table.window import Session# 定义会话窗口，不活动间隔为30分钟
result_table = t_env.from_path("source_table") \.window(Session.with_gap(expr.lit(30).minutes).on(expr.col("event_time")).alias("w")) \.group_by(expr.col("w"), expr.col("user_id")) \.select(expr.col("user_id"), expr.col("amount").sum.alias("total_spent"), expr.col("w").end.alias("window_end"))# 输出查询结果
result_table.execute().print()

6.4 计数窗口（Count Window）

计数窗口是基于记录的数量来划分窗口，而不是基于时间。例如，每 1000 条记录形成一个窗口。

from pyflink.table.window import Tumble# 定义计数窗口，每 1000 条记录形成一个窗口
result_table = t_env.from_path("source_table") \.window(Tumble.over(expr.lit(1000).rows).on(expr.col("event_time")).alias("w")) \.group_by(expr.col("w"), expr.col("user_id")) \.select(expr.col("user_id"), expr.col("amount").sum.alias("total_spent"), expr.col("w").end.alias("window_end"))# 输出查询结果
result_table.execute().print()

7. 自定义窗口聚合函数

除了使用内置的窗口聚合函数（如 SUM, COUNT 等），你还可以自定义窗口聚合逻辑。

自定义聚合函数

from pyflink.table.udf import AggregateFunction, udafclass AvgAggregateFunction(AggregateFunction):def get_value(self, accumulator):return accumulator[0] / accumulator[1] if accumulator[1] > 0 else 0def create_accumulator(self):return [0, 0]  # sum, countdef accumulate(self, accumulator, value):accumulator[0] += valueaccumulator[1] += 1# 注册自定义聚合函数
avg_udaf = udaf(AvgAggregateFunction(), result_type='DOUBLE')# 使用自定义聚合函数计算窗口内平均值
result_table = t_env.from_path("source_table") \.window(Tumble.over(expr.lit(10).minutes).on(expr.col("event_time")).alias("w")) \.group_by(expr.col("w"), expr.col("user_id")) \.select(expr.col("user_id"), avg_udaf(expr.col("amount")).alias("average_spent"), expr.col("w").end.alias("window_end"))# 输出查询结果
result_table.execute().print()

8. 完整示例

以下是一个包含窗口操作的完整 PyFlink 示例：

from pyflink.table.window import Tumble
from pyflink.table import expressions as expr
from pyflink.table.udf import AggregateFunction, udaf
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings# 设置流执行环境
env = StreamExecutionEnvironment.get_execution_environment()
settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=settings)# 创建示例表
t_env.execute_sql("""CREATE TEMPORARY TABLE source_table (user_id STRING,item STRING,amount DOUBLE,event_time TIMESTAMP(3),WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND) WITH ('connector' = 'datagen')
""")# 定义滚动窗口和自定义聚合函数
class AvgAggregateFunction(AggregateFunction):def get_value(self, accumulator):return accumulator[0] / accumulator[1] if accumulator[1] > 0 else 0def create_accumulator(self):return [0, 0]def accumulate(self, accumulator, value):accumulator[0] += valueaccumulator[1] += 1avg_udaf = udaf(AvgAggregateFunction(), result_type='DOUBLE')# 使用滚动窗口和自定义聚合函数
result_table = t_env.from_path("source_table") \.window(Tumble.over(expr.lit(10).minutes).on(expr.col("event_time")).alias("w")) \.group_by(expr.col("w"), expr.col("user_id")) \.select(expr.col("user_id"), avg_udaf(expr.col("amount")).alias("average_spent"), expr.col("w").end.alias("window_end"))# 输出结果
result_table.execute().print()

9. 总结

在 PyFlink 中，窗口是流处理的核心概念之一，允许你对无界数据流进行聚合、计算和操作。Flink 提供了丰富的窗口类型，包括滚动窗口、滑动窗口、会话窗口和计数窗口，以满足不同场景下的需求。通过本教程，你可以学习如何在 PyFlink 中使用窗口对流数据进行处理，并通过自定义函数来实现更复杂的计算逻辑。