企业官网建设流程全解析

• 大数据
• 流处理
• 批处理
• 数据工程

本文以 PyFlink 官方示例文档 window.rst 为骨架，完整讲解 DataStream API 中三类窗口（滚动窗口 Tumble、滑动窗口 Sliding、会话窗口 Session）的 Python 实现：包括事件时间（Event Time）与计数（Count）两种维度、固定间隔（Gap）与动态间隔（Dynamic Gap）两种会话窗口，以及 Watermark 分配、自定义ProcessWindowFunction、FileSink输出等配套技术。读完本文，你将能够直接复制运行 5 个开箱即用的窗口示例，并理解 PyFlink 窗口分配器（WindowAssigner）的底层分配逻辑与触发机制。

窗口（Window）是什么

流处理中的"窗口"是对无界数据流按时间或数量进行切分、形成有限计算批次的手段。PyFlink 的 DataStream API 沿用了 Flink 经典的窗口体系，核心思路是：key_by分组 → 指定窗口分配器（WindowAssigner）→ 在窗口上应用聚合或自定义处理函数。

本文涉及的窗口类型来自 PyFlink 官方示例目录 flink-python/pyflink/examples/datastream/windowing/，包括：

所有示例均以env.from_collection构造内存数据源，设置并行度为 1，可通过--output参数指定输出文件，否则直接打印到标准输出，非常适合本地快速验证窗口语义。

事件时间、Watermark 与 TimestampAssigner

时间窗口示例全部采用事件时间（Event Time）语义，因此必须先为数据分配时间戳与 Watermark。PyFlink 中通过WatermarkStrategy完成：

watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \ .with_timestamp_assigner(MyTimestampAssigner())

其中for_monotonous_timestamps()生成单调递增的 Watermark（适用于乱序程度可忽略的数据），MyTimestampAssigner继承自 TimestampAssigner：

class MyTimestampAssigner(TimestampAssigner): def extract_timestamp(self, value, record_timestamp) -> int: return int(value[1])

这里将二元组(word, ts)的第二个字段（毫秒时间戳）作为事件时间。随后通过assign_timestamps_and_watermarks(watermark_strategy)将策略应用到数据流上。若忘记分配时间戳而直接使用事件时间窗口，PyFlink 会抛出"Record has Java Long.MIN_VALUE timestamp"异常——这一保护逻辑可在 window.py 的assign_windows实现中看到。

Tumble Window（滚动窗口）

滚动窗口将数据流按固定大小切分为互不重叠的窗口，每条数据恰好属于一个窗口。官方示例文档给出了两种滚动窗口：基于事件时间的滚动时间窗口与基于元素数量的滚动计数窗口。

Tumbling Time Window（滚动时间窗口）

完整示例代码位于 tumbling_time_window.py，如下：

import sys import argparse from typing import Iterable from pyflink.datastream.connectors.file_system import FileSink, OutputFileConfig, RollingPolicy from pyflink.common import Types, WatermarkStrategy, Time, Encoder from pyflink.common.watermark_strategy import TimestampAssigner from pyflink.datastream import StreamExecutionEnvironment, ProcessWindowFunction from pyflink.datastream.window import TumblingEventTimeWindows, TimeWindow class MyTimestampAssigner(TimestampAssigner): def extract_timestamp(self, value, record_timestamp) -> int: return int(value[1]) class CountWindowProcessFunction(ProcessWindowFunction[tuple, tuple, str, TimeWindow]): def process(self, key: str, context: ProcessWindowFunction.Context[TimeWindow], elements: Iterable[tuple]) -> Iterable[tuple]: return [(key, context.window().start, context.window().end, len([e for e in elements]))] if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument( '--output', dest='output', required=False, help='Output file to write results to.') argv = sys.argv[1:] known_args, _ = parser.parse_known_args(argv) output_path = known_args.output env = StreamExecutionEnvironment.get_execution_environment() # write all the data to one file env.set_parallelism(1) # define the source data_stream = env.from_collection([ ('hi', 1), ('hi', 2), ('hi', 3), ('hi', 4), ('hi', 5), ('hi', 8), ('hi', 9), ('hi', 15)], type_info=Types.TUPLE([Types.STRING(), Types.INT()])) # define the watermark strategy watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \ .with_timestamp_assigner(MyTimestampAssigner()) ds = data_stream.assign_timestamps_and_watermarks(watermark_strategy) \ .key_by(lambda x: x[0], key_type=Types.STRING()) \ .window(TumblingEventTimeWindows.of(Time.milliseconds(5))) \ .process(CountWindowProcessFunction(), Types.TUPLE([Types.STRING(), Types.INT(), Types.INT(), Types.INT()])) # define the sink if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print() # submit for execution env.execute()

关键点拆解：

• 窗口定义：TumblingEventTimeWindows.of(Time.milliseconds(5))创建大小为 5 毫秒的滚动事件时间窗口。Time还支持seconds、minutes、hours、days等粒度。
• offset 偏移参数：of(size, offset)的第二个参数offset用于将窗口起点整体平移（例如处理 UTC+8 时区、让窗口对齐本地零点）。从 window.py 的实现可以看到，abs(offset)必须小于size，否则构造器直接抛异常。
• 窗口起点计算：分配器通过TimeWindow.get_window_start_with_offset(timestamp, offset, window_size)计算窗口起点，公式为timestamp - (timestamp - offset + window_size) % window_size（见 window.py）。
• 自定义处理函数：CountWindowProcessFunction继承ProcessWindowFunction[tuple, tuple, str, TimeWindow]，通过context.window().start/context.window().end拿到窗口起止时间，统计窗口内元素数量后输出四元组(key, window_start, window_end, count)。ProcessWindowFunction与增量聚合（如reduce/aggregate）相比，优势在于能访问窗口元数据并拿到完整元素集合。
• 输出类型：.process(..., Types.TUPLE([...]))显式声明输出类型，这是 PyFlink 类型推断的推荐做法。

上述数据的时间戳为 1、2、3、4、5、8、9、15（毫秒），窗口大小为 5，因此会切分出[1,5)、[5,10)、[10,15)、[15,20)等窗口区间（起始含、结尾不含），同 key 数据按各自时间戳落入对应窗口并输出计数。

Tumbling Count Window（滚动计数窗口）

计数窗口不依赖时间，而是按元素个数触发。示例 tumbling_count_window.py 完整代码如下：

import sys import argparse from typing import Iterable from pyflink.datastream.connectors.file_system import FileSink, OutputFileConfig, RollingPolicy from pyflink.common import Types, Encoder from pyflink.datastream import StreamExecutionEnvironment, WindowFunction from pyflink.datastream.window import CountWindow class SumWindowFunction(WindowFunction[tuple, tuple, str, CountWindow]): def apply(self, key: str, window: CountWindow, inputs: Iterable[tuple]): result = 0 for i in inputs: result += i[0] return [(key, result)] if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument( '--output', dest='output', required=False, help='Output file to write results to.') argv = sys.argv[1:] known_args, _ = parser.parse_known_args(argv) output_path = known_args.output env = StreamExecutionEnvironment.get_execution_environment() # write all the data to one file env.set_parallelism(1) # define the source data_stream = env.from_collection([ (1, 'hi'), (2, 'hello'), (3, 'hi'), (4, 'hello'), (5, 'hi'), (6, 'hello'), (6, 'hello')], type_info=Types.TUPLE([Types.INT(), Types.STRING()])) ds = data_stream.key_by(lambda x: x[1], key_type=Types.STRING()) \ .count_window(2) \ .apply(SumWindowFunction(), Types.TUPLE([Types.STRING(), Types.INT()])) # define the sink if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print() # submit for execution env.execute()

关键点拆解：

• 计数窗口 API：count_window(2)直接生成每 2 个元素触发一次的滚动计数窗口，无需时间戳与 Watermark。其对应的CountWindow在 window.py 中被定义为按唯一id标识的窗口，max_timestamp返回MAX_LONG_VALUE，表示其不受时间约束。
• 分组维度：本例按字符串字段x[1]分组，因此'hi'和'hello'两个 key 各自独立计数。
• 求和逻辑：SumWindowFunction继承WindowFunction[tuple, tuple, str, CountWindow]，在apply中对窗口内所有元素的整数字段求和。数据中(1,'hi'), (3,'hi'), (5,'hi')会形成两批（每批 2 个），输出('hi', 4)与('hi', 5)；'hello'依次为 2+4=6、6+6=12。
• 注意：计数窗口的触发条件是"每 n 个元素"，当 key 元素总数不是 n 的整数倍时，余数部分不会触发窗口，这是计数窗口的固有语义。

Sliding Window（滑动窗口）

滑动窗口有两个参数：窗口大小（size）与滑动步长（slide）。窗口之间允许重叠，每条数据可能同时属于多个窗口。当size == slide时退化为滚动窗口。

Sliding Time Window（滑动时间窗口）

示例 sliding_time_window.py 完整代码如下：

import sys import argparse from typing import Iterable from pyflink.datastream.connectors.file_system import FileSink, OutputFileConfig, RollingPolicy from pyflink.common import Types, WatermarkStrategy, Time, Encoder from pyflink.common.watermark_strategy import TimestampAssigner from pyflink.datastream import StreamExecutionEnvironment, ProcessWindowFunction from pyflink.datastream.window import SlidingEventTimeWindows, TimeWindow class MyTimestampAssigner(TimestampAssigner): def extract_timestamp(self, value, record_timestamp) -> int: return int(value[1]) class CountWindowProcessFunction(ProcessWindowFunction[tuple, tuple, str, TimeWindow]): def process(self, key: str, context: ProcessWindowFunction.Context[TimeWindow], elements: Iterable[tuple]) -> Iterable[tuple]: return [(key, context.window().start, context.window().end, len([e for e in elements]))] if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument( '--output', dest='output', required=False, help='Output file to write results to.') argv = sys.argv[1:] known_args, _ = parser.parse_known_args(argv) output_path = known_args.output env = StreamExecutionEnvironment.get_execution_environment() # write all the data to one file env.set_parallelism(1) # define the source data_stream = env.from_collection([ ('hi', 1), ('hi', 2), ('hi', 3), ('hi', 4), ('hi', 5), ('hi', 8), ('hi', 9), ('hi', 15)], type_info=Types.TUPLE([Types.STRING(), Types.INT()])) # define the watermark strategy watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \ .with_timestamp_assigner(MyTimestampAssigner()) ds = data_stream.assign_timestamps_and_watermarks(watermark_strategy) \ .key_by(lambda x: x[0], key_type=Types.STRING()) \ .window(SlidingEventTimeWindows.of(Time.milliseconds(5), Time.milliseconds(2))) \ .process(CountWindowProcessFunction(), Types.TUPLE([Types.STRING(), Types.INT(), Types.INT(), Types.INT()])) # define the sink if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print() # submit for execution env.execute()

关键点拆解：

• 窗口定义：SlidingEventTimeWindows.of(Time.milliseconds(5), Time.milliseconds(2))表示窗口大小为 5 毫秒、每 2 毫秒滑动一次。因此同一时刻最多存在ceil(size / slide) = 3个重叠窗口。
• 窗口计算逻辑：从 window.py 的SlidingProcessingTimeWindows.assign_windows可以看到滑动窗口的分配算法：先以 slide 为步长计算最后一个窗口起点last_start，再向前枚举range(last_start, current_time - size, -slide)生成所有覆盖当前时间戳的窗口。SlidingEventTimeWindows遵循同样的多窗口分配逻辑，只是基于事件时间戳而非系统时间。
• 参数约束：SlidingProcessingTimeWindows构造器要求abs(offset) < slide 且 size > 0（见 window.py），内部还以math.gcd(size, slide)计算 pane 大小以优化状态管理。
• 结果语义：与滚动窗口示例相同的CountWindowProcessFunction会为每个重叠窗口各输出一条记录，因此同一数据会出现在多条输出中。例如时间戳 4 的数据会同时落在[2,7)、[4,9)两个窗口内。

Session Window（会话窗口）

会话窗口按"不活动间隔"切分：窗口在数据到达时创建，若两条数据间隔超过设定的 gap，则视为新的会话；相邻会话若被新数据"桥接"，会动态合并。会话窗口没有固定长度，天然适合用户活跃度、页面停留时长等场景。PyFlink 的会话窗口分为固定 gap 与动态 gap 两种。

Session With Gap Window（固定间隔会话窗口）

示例 session_with_gap_window.py 完整代码如下：

import sys import argparse from typing import Iterable from pyflink.datastream.connectors.file_system import FileSink, RollingPolicy, OutputFileConfig from pyflink.common import Types, WatermarkStrategy, Time, Encoder from pyflink.common.watermark_strategy import TimestampAssigner from pyflink.datastream import StreamExecutionEnvironment, ProcessWindowFunction from pyflink.datastream.window import EventTimeSessionWindows, \ SessionWindowTimeGapExtractor, TimeWindow class MyTimestampAssigner(TimestampAssigner): def extract_timestamp(self, value, record_timestamp) -> int: return int(value[1]) class MySessionWindowTimeGapExtractor(SessionWindowTimeGapExtractor): def extract(self, element: tuple) -> int: return element[1] class CountWindowProcessFunction(ProcessWindowFunction[tuple, tuple, str, TimeWindow]): def process(self, key: str, context: ProcessWindowFunction.Context[TimeWindow], elements: Iterable[tuple]) -> Iterable[tuple]: return [(key, context.window().start, context.window().end, len([e for e in elements]))] if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument( '--output', dest='output', required=False, help='Output file to write results to.') argv = sys.argv[1:] known_args, _ = parser.parse_known_args(argv) output_path = known_args.output env = StreamExecutionEnvironment.get_execution_environment() # write all the data to one file env.set_parallelism(1) # define the source data_stream = env.from_collection([ ('hi', 1), ('hi', 2), ('hi', 3), ('hi', 4), ('hi', 8), ('hi', 9), ('hi', 15)], type_info=Types.TUPLE([Types.STRING(), Types.INT()])) # define the watermark strategy watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \ .with_timestamp_assigner(MyTimestampAssigner()) ds = data_stream.assign_timestamps_and_watermarks(watermark_strategy) \ .key_by(lambda x: x[0], key_type=Types.STRING()) \ .window(EventTimeSessionWindows.with_gap(Time.milliseconds(5))) \ .process(CountWindowProcessFunction(), Types.TUPLE([Types.STRING(), Types.INT(), Types.INT(), Types.INT()])) # define the sink if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print() # submit for execution env.execute()

关键点拆解：

• 窗口定义：EventTimeSessionWindows.with_gap(Time.milliseconds(5))设定会话间隔为 5 毫秒。数据时间戳为 1、2、3、4、8、9、15：1~4之间间隔 ≤5 且连续，合并为一个会话[1, 9)（4 + 5 = 9）；8、9与前一会话首尾衔接（4 到 8 间隔 4 < 5），继续并入同一会话；15与9间隔 6 > 5，开启新会话[15, 20)。最终输出两条记录：('hi', 1, 9, 6)与('hi', 15, 20, 1)。
• 会话合并机制：EventTimeSessionWindows继承自MergingWindowAssigner，其assign_windows为每个元素生成TimeWindow(timestamp, timestamp + gap)（见 window.py），随后通过TimeWindow.merge_windows（见 window.py）对相交（intersects）窗口执行合并，合并后取两个窗口起止的最小/最大值（cover）。
• 触发条件：默认使用EventTimeTrigger，即 Watermark 越过窗口max_timestamp（end - 1）时窗口关闭并触发计算。

Session With Dynamic Gap Window（动态间隔会话窗口）

当每个元素所需的会话间隔不同（例如依据用户等级、请求类型动态变化）时，使用with_dynamic_gap+ 自定义SessionWindowTimeGapExtractor。示例 session_with_dynamic_gap_window.py 完整代码如下：

import sys import argparse from typing import Iterable from pyflink.datastream.connectors.file_system import FileSink, OutputFileConfig, RollingPolicy from pyflink.common import Types, WatermarkStrategy, Encoder from pyflink.common.watermark_strategy import TimestampAssigner from pyflink.datastream import StreamExecutionEnvironment, ProcessWindowFunction from pyflink.datastream.window import EventTimeSessionWindows, \ SessionWindowTimeGapExtractor, TimeWindow class MyTimestampAssigner(TimestampAssigner): def extract_timestamp(self, value, record_timestamp) -> int: return int(value[1]) class MySessionWindowTimeGapExtractor(SessionWindowTimeGapExtractor): def extract(self, element: tuple) -> int: return element[1] class CountWindowProcessFunction(ProcessWindowFunction[tuple, tuple, str, TimeWindow]): def process(self, key: str, context: ProcessWindowFunction.Context[TimeWindow], elements: Iterable[tuple]) -> Iterable[tuple]: return [(key, context.window().start, context.window().end, len([e for e in elements]))] if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument( '--output', dest='output', required=False, help='Output file to write results to.') argv = sys.argv[1:] known_args, _ = parser.parse_known_args(argv) output_path = known_args.output env = StreamExecutionEnvironment.get_execution_environment() # write all the data to one file env.set_parallelism(1) # define the source data_stream = env.from_collection([ ('hi', 1), ('hi', 2), ('hi', 3), ('hi', 4), ('hi', 8), ('hi', 9), ('hi', 15)], type_info=Types.TUPLE([Types.STRING(), Types.INT()])) # define the watermark strategy watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \ .with_timestamp_assigner(MyTimestampAssigner()) ds = data_stream.assign_timestamps_and_watermarks(watermark_strategy) \ .key_by(lambda x: x[0], key_type=Types.STRING()) \ .window(EventTimeSessionWindows.with_dynamic_gap(MySessionWindowTimeGapExtractor())) \ .process(CountWindowProcessFunction(), Types.TUPLE([Types.STRING(), Types.INT(), Types.INT(), Types.INT()])) # define the sink if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print() # submit for execution env.execute()

关键点拆解：

• 动态间隔提取器：MySessionWindowTimeGapExtractor继承SessionWindowTimeGapExtractor（抽象基类定义见 window.py），在extract(element)中返回该元素自身的间隔值。本示例中元素二元组第二个字段恰好同时充当"时间戳"与"会话间隔"，便于演示；实际业务中二者通常是不同的字段。
• API 关联：EventTimeSessionWindows.with_dynamic_gap(extractor)（见 window.py）内部构造DynamicEventTimeSessionWindows，为每个元素独立计算窗口区间。因此同 key 下相邻元素若间隔小于等于各自动态 gap 之和，会话就会合并。
• 固定 vs 动态的选择：固定 gap（with_gap）语义简单、状态开销可控；动态 gap 更贴近真实业务（如不同支付渠道的超时阈值不同），但 gap 计算逻辑需保证确定性，以便故障恢复后合并结果一致。

运行方式与输出配置

运行示例

所有示例均可直接以 Python 脚本方式运行（需已安装 PyFlink）：

python tumbling_time_window.py python tumbling_time_window.py --output /tmp/flink_output

• 不传--output时，结果通过ds.print()打印到标准输出，并提示 "Printing result to stdout. Use --output to specify output path."。
• 传--output时，结果写入FileSink管理的输出目录。

FileSink 输出细节

五个示例的输出段完全一致，使用FileSink按行格式写出：

ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() )

参数说明：

• for_row_format(base_path, encoder)：以行为单位写出，base_path是输出目录；Encoder.simple_string_encoder()将每个元素序列化为字符串行。
• with_output_file_config：通过OutputFileConfig.builder()定制输出文件名，with_part_prefix("prefix")与with_part_suffix(".ext")生成形如prefix-<uuid>.ext的文件名。
• with_rolling_policy(RollingPolicy.default_rolling_policy())：使用默认滚动策略（按文件大小与不活动时间滚动落盘），控制文件切分节奏。

FileSink、OutputFileConfig、RollingPolicy均来自pyflink.datastream.connectors.file_system，相关实现位于 flink-python/pyflink/datastream/connectors/file_system.py。由于示例设置了env.set_parallelism(1)，所有数据会写入单个文件，便于核对窗口计算结果。

窗口 API 源码速览与进阶指引

窗口分配器体系

PyFlink 的窗口逻辑集中在 flink-python/pyflink/datastream/window.py，核心类结构如下：

触发器的默认选择

窗口分配器的get_default_trigger决定了窗口何时关闭计算：事件时间窗口（TumblingEventTimeWindows、SlidingEventTimeWindows、EventTimeSessionWindows）默认使用EventTimeTrigger（Watermark 越过窗口末尾即触发）；处理时间窗口默认使用ProcessingTimeTrigger（系统时钟到达窗口末尾即触发）。会话窗口额外依赖合并回调，EventTimeSessionWindows.merge_windows直接委托给TimeWindow.merge_windows。

进阶扩展方向

• 处理时间版本：把示例中的*EventTimeWindows换成*ProcessingTimeWindows，并去掉assign_timestamps_and_watermarks，即可切换为处理时间语义，适合对精确性要求不高、追求低延迟的场景。
• 增量聚合：ProcessWindowFunction需要缓存全部窗口元素，数据量大时可改用.reduce()/.aggregate()配合ProcessWindowFunction做增量聚合。
• allowed_lateness 与旁路输出：事件时间窗口可设置允许迟到时间，并通过侧输出收集迟到数据，实现更稳健的乱序处理。
• 更完整的示例集合：窗口之外，PyFlink 官方文档还提供 basic_operations.rst、state.rst、timer.rst、process_json_data.rst 等配套示例，可组合阅读以构建完整的 DataStream 应用能力。

小结

本文以官方示例文档 window.rst 为主线，完整呈现了 PyFlink DataStream 的 5 个窗口示例：滚动事件时间窗口、滚动计数窗口、滑动事件时间窗口、固定间隔会话窗口与动态间隔会话窗口。每个示例都配套讲解了 Watermark 分配、key_by分组、自定义窗口函数、FileSink输出与窗口分配器源码，读者既可以直接复制运行验证窗口语义，也可以基于源码理解 Flink 窗口从分配到合并再到触发的完整生命周期。建议动手修改窗口大小、滑动步长与 gap 值，观察输出变化，这是掌握流式窗口最有效的方式。

• 大数据
• 流处理
• 批处理
• 数据工程