PyFlink DataStream 窗口操作实战指南：Tumbling / Sliding / Session 窗口的 Python API 用法与源码解析

大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载本文以 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/包括窗口类型示例文件分配器Tumbling Time Window滚动时间窗口tumbling_time_window.pyTumblingEventTimeWindowsTumbling Count Window滚动计数窗口tumbling_count_window.pyCountWindow通过.count_window(n)触发Sliding Time Window滑动时间窗口sliding_time_window.pySlidingEventTimeWindowsSession With Gap Window固定间隔会话窗口session_with_gap_window.pyEventTimeSessionWindows.with_gapSession With Dynamic Gap Window动态间隔会话窗口session_with_dynamic_gap_window.pyEventTimeSessionWindows.with_dynamic_gap所有示例均以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继承自 TimestampAssignerclass 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, destoutput, requiredFalse, helpOutput 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_infoTypes.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_typeTypes.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( sinkFileSink.for_row_format( base_pathoutput_path, encoderEncoder.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用于将窗口起点整体平移例如处理 UTC8 时区、让窗口对齐本地零点。从 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, destoutput, requiredFalse, helpOutput 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_infoTypes.TUPLE([Types.INT(), Types.STRING()])) ds data_stream.key_by(lambda x: x[1], key_typeTypes.STRING()) \ .count_window(2) \ .apply(SumWindowFunction(), Types.TUPLE([Types.STRING(), Types.INT()])) # define the sink if output_path is not None: ds.sink_to( sinkFileSink.for_row_format( base_pathoutput_path, encoderEncoder.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()关键点拆解计数窗口 APIcount_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依次为 246、6612。注意计数窗口的触发条件是每 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, destoutput, requiredFalse, helpOutput 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_infoTypes.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_typeTypes.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( sinkFileSink.for_row_format( base_pathoutput_path, encoderEncoder.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, destoutput, requiredFalse, helpOutput 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_infoTypes.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_typeTypes.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( sinkFileSink.for_row_format( base_pathoutput_path, encoderEncoder.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、151~4之间间隔 ≤5 且连续合并为一个会话[1, 9)4 5 98、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_timestampend - 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, destoutput, requiredFalse, helpOutput 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_infoTypes.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_typeTypes.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( sinkFileSink.for_row_format( base_pathoutput_path, encoderEncoder.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 动态的选择固定 gapwith_gap语义简单、状态开销可控动态 gap 更贴近真实业务如不同支付渠道的超时阈值不同但 gap 计算逻辑需保证确定性以便故障恢复后合并结果一致。运行方式与输出配置运行示例所有示例均可直接以 Python 脚本方式运行需已安装 PyFlinkpython 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( sinkFileSink.for_row_format( base_pathoutput_path, encoderEncoder.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核心类结构如下类行号说明TimeWindowwindow.py表示[start, end)左闭右开时间区间提供max_timestampend-1、intersects、cover、merge_windows等方法CountWindowwindow.py按唯一 id 标识的计数窗口max_timestamp恒为MAX_LONG_VALUESessionWindowTimeGapExtractorwindow.py动态会话间隔提取抽象基类TumblingProcessingTimeWindowswindow.py滚动处理时间窗口TumblingEventTimeWindowswindow.py滚动事件时间窗口SlidingProcessingTimeWindowswindow.py滑动处理时间窗口SlidingEventTimeWindowswindow.py滑动事件时间窗口ProcessingTimeSessionWindowswindow.py处理时间会话窗口EventTimeSessionWindowswindow.py事件时间会话窗口含with_gap与with_dynamic_gap触发器的默认选择窗口分配器的get_default_trigger决定了窗口何时关闭计算事件时间窗口TumblingEventTimeWindows、SlidingEventTimeWindows、EventTimeSessionWindows默认使用EventTimeTriggerWatermark 越过窗口末尾即触发处理时间窗口默认使用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 值观察输出变化这是掌握流式窗口最有效的方式。赞分享大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载相关推荐PyFlink Table API 窗口操作实战Tumble / Slide / Session 窗口示例与源码解析PyFlink Table API 窗口操作实战Tumble / Slide / Session 窗口示例与源码解析 导读 本文以 Apache Flink大数据流处理批处理数据工程PyFlink DataStream 窗口机制全解析pyflink.datastream.window 模块 API 与源码实战指南PyFlink DataStream 窗口机制全解析pyflink.datastream.window 模块 API 与源码实战指南 导读 窗口Window大数据流处理批处理数据工程PyFlink Table Window 窗口 API 完全指南Tumble、Slide、Session 与 Over 窗口PyFlink Table Window 窗口 API 完全指南Tumble、Slide、Session 与 Over 窗口 窗口Window是流式数据处大数据流处理批处理数据工程上一篇LuckPerms Web编辑器完全指南可视化权限管理新体验下一篇RustDesk隐私模式如何解决企业远程管理中的安全与隐私平衡难题创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考