Apache Beam 固定时间窗口（Fixed Time Window）实战：用 1 天窗口统计事件数量

大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载窗口Windowing是 Apache Beam 统一批流编程模型的核心抽象之一它让无界数据流可以被切分为一个个有界的、可聚合的逻辑单元。本文以 Beam Katas 中 Fixed Time Window 练习见 learning/katas/java/Windowing/Fixed Time Window/Fixed Time Window/task.md为主线系统讲解固定时间窗口的工作原理、FixedWindows的底层实现并给出一个可直接运行、可被测试验证的完整 Java 示例按 1 天窗口统计事件数量。读完本文你将理解 Beam 中元素时间戳与窗口的对应关系掌握Window.into(FixedWindows.of(...))与Count.perElement()的组合用法并学会用PAssert验证窗口化的聚合结果。为什么需要窗口把无界数据切成有界片段在 Beam 的编程模型中窗口化Windowing根据 PCollection 中每个元素的时间戳将整个集合细分为多个逻辑窗口。这一点之所以关键是因为像GroupByKey、Combine这类聚合变换天然是有界的——它们需要看到一批完整的输入才能产生输出而流式场景下的无界 PCollection 永远不会结束。窗口机制给出的答案是聚合变换隐式地按窗口工作——它们把每个 PCollection 当作一连串有限的窗口来处理即使整个集合本身可能是无界的。具体来说任意 PCollection包括无界 PCollection都可以被细分为逻辑窗口每个元素根据所属 PCollection 的窗口化函数windowing function被分配到一个或多个窗口中每个窗口内部包含有限数量的元素因而可以安全地执行聚合分组类变换按键 窗口两个维度组织元素。例如GroupByKey隐式地将一个 PCollection 的元素按 key 和 window 同时分组——这意味着即使两个元素拥有相同的 key只要它们落在不同的窗口中就不会被聚合到同一条输出中。Beam 提供的四种窗口化函数Beam SDK 内置了多种窗口化函数覆盖不同的业务场景对应练习的 Windowing 课程见 learning/katas/java/Windowing/Fixed Time Window/lesson-info.yaml窗口类型语义典型场景Fixed Time Windows固定时间窗口将时间轴切分为固定时长、互不重叠的区间按天/按小时统计流量、交易额Sliding Time Windows滑动时间窗口固定时长但可重叠每隔一定间隔产生一个新窗口滚动平均值、近 1 小时 UVPer-Session Windows会话窗口按元素间的间隔动态切分间隔超过阈值即开启新会话用户活跃会话、点击流分析Single Global Window全局窗口整个 PCollection 只有一个窗口需配合触发器使用对无界数据做全局聚合如总数其中固定时间窗口是最简单、最常用的一种它把数据流表示为一系列时长一致、互不重叠的时间区间。每个元素恰好属于一个固定窗口窗口边界由元素时间戳唯一确定。Kata 实战按 1 天固定窗口统计事件数量本次练习的目标非常明确Kata请基于持续时间为 1 天的固定窗口统计每个窗口中发生的事件数量。即输入一批带时间戳的事件输出每个 1 天窗口内的事件计数。任务描述与提示见 task.md完整实现位于 Task.java。第一步构造带时间戳的输入数据固定窗口依赖元素的事件时间event time因此输入不能是普通的字符串集合而必须使用Create.timestamped为每个元素显式指定时间戳PCollectionString events pipeline.apply( Create.timestamped( TimestampedValue.of(event, Instant.parse(2019-06-01T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-01T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-01T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-01T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-05T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-05T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-08T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-08T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-08T00:00:0000:00)), TimestampedValue.of(event, Instant.parse(2019-06-10T00:00:0000:00)) ) );TimestampedValue.of(value, instant)将普通值与其事件时间绑定Instant.parse使用 ISO-8601 格式解析时间戳基于 Joda-Time。这 10 个事件分布在 4 个不同的日期上便于我们直观验证窗口切分结果。第二步核心变换——窗口 计数applyTransform是本次练习的关键只需两步static PCollectionKVString, Long applyTransform(PCollectionString events) { return events .apply(Window.into(FixedWindows.of(Duration.standardDays(1)))) .apply(Count.perElement()); }拆解如下Window.into(FixedWindows.of(Duration.standardDays(1)))将 PCollection 的窗口化函数设置为1 天固定窗口。Duration.standardDays(1)来自 Joda-Time表示 24 小时FixedWindows.of(...)是工厂方法默认窗口起始偏移为 0即从 Unix 纪元对齐。Count.perElement()统计每个元素值出现的次数。由于它发生在窗口化之后计数会自动在每个窗口内部独立进行——这正是聚合变换按窗口工作的直接体现。最终输出的PCollectionKVString, Long中key 是事件名本示例中均为 eventvalue 是该窗口内该事件出现的次数。第三步输出与运行在main方法中通过Log.ofElements()将结果打印出来然后运行管道PipelineOptions options PipelineOptionsFactory.fromArgs(args).create(); Pipeline pipeline Pipeline.create(options); PCollectionString events /* 第一步中的带时间戳数据 */; PCollectionKVString, Long output applyTransform(events); output.apply(Log.ofElements()); pipeline.run();Log.ofElements()是 Katas 工具包提供的调试输出变换位于 learning/katas/java/util 目录会打印每个元素及其所在窗口非常适合观察窗口切分效果。窗口分配原理解读 FixedWindows 源码为了深入理解事件被分到哪个窗口需要阅读FixedWindows的底层实现sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/FixedWindows.java。数据结构与工厂方法FixedWindows继承自PartitioningWindowFnObject, IntervalWindow即每个元素恰好属于一个窗口的分区型窗口函数。它内部只有两个字段size窗口大小时长offset窗口起始偏移默认Duration.ZERO。两个构造入口public static FixedWindows of(Duration size) { return new FixedWindows(size, Duration.ZERO); } public FixedWindows withOffset(Duration offset) { return new FixedWindows(size, offset); }withOffset用于对齐窗口边界例如让自然日窗口从某个时区偏移开始其约束为0 offset size否则构造函数会抛出IllegalArgumentException。窗口边界计算半开区间核心方法assignWindow(Instant timestamp)将时间戳空间划分为如下形式的半开区间[N * size offset, (N 1) * size offset) 其中 0 为 Unix 纪元窗口的起始时间通过取模计算Instant start new Instant( timestamp.getMillis() - timestamp.plus(size).minus(offset).getMillis() % size.getMillis()); Instant end start.plus(size); return new IntervalWindow(start, end);注意两个细节半开区间[start, end)窗口包含起始边界、不包含结束边界。因此2019-06-01T00:00:00属于[2019-06-01, 2019-06-02)窗口而2019-06-02T00:00:00精确落在边界上时属于下一个窗口尾部截断若start size超出全局窗口的最大可表示时间则窗口终点被截断为全局窗口末端保证边界合法源码注释中甚至幽默地提到只有当数据来自公元 294247 年时才会真正遇到这种情况。窗口兼容性isCompatible/verifyCompatibility规定只有size与offset完全相同的两个FixedWindows才是兼容的否则在管道合并窗口化策略时会抛出IncompatibleWindowException。这一约束保证了同一 PCollection 的所有元素遵循一致的窗口划分。用测试验证窗口化结果Katas 为每个练习都配了隐藏的单元测试见 TaskTest.java它用TestPipelinePAssert精确断言每个窗口的计数是理解固定窗口语义的最好教材。测试将Task.applyTransform的结果再经过一个ParDo借助BoundedWindow参数读出每个元素所属窗口的字符串表示然后断言PAssert.that(windowedResults) .containsInAnyOrder( new WindowedEvent(event, 4L, [2019-06-01T00:00:00.000Z..2019-06-02T00:00:00.000Z)), new WindowedEvent(event, 2L, [2019-06-05T00:00:00.000Z..2019-06-06T00:00:00.000Z)), new WindowedEvent(event, 3L, [2019-06-08T00:00:00.000Z..2019-06-09T00:00:00.000Z)), new WindowedEvent(event, 1L, [2019-06-10T00:00:00.000Z..2019-06-11T00:00:00.000Z)) );对应窗口与计数的对应关系如下输入时间戳所属窗口[start, end)事件数2019-06-01×4[2019-06-01T00:00:00Z .. 2019-06-02T00:00:00Z)42019-06-05×2[2019-06-05T00:00:00Z .. 2019-06-06T00:00:00Z)22019-06-08×3[2019-06-08T00:00:00Z .. 2019-06-09T00:00:00Z)32019-06-10×1[2019-06-10T00:00:00Z .. 2019-06-11T00:00:00Z)1测试还展示了如何在DoFn中访问当前窗口在ProcessElement方法里声明一个BoundedWindow类型的参数Beam 会自动注入该元素所在的窗口对象。WindowedEvent见 WindowedEvent.java是一个可序列化的简单 POJO用于承载事件名 计数 窗口三元组以便断言。练习环境与扩展学习完成练习本练习的task-info.yaml见 task-info.yaml声明了Task.java中需要你填写的TODO()占位符即上述Window.into(FixedWindows.of(Duration.standardDays(1)))一行代码测试文件默认不可见用于自动评判你的实现。运行管道使用./gradlew runJava 版 Katas 提供 Gradle Wrapper见 learning/katas/java或直接在 IDE 中运行Task.main即可看到按窗口输出的计数日志。继续进阶掌握固定窗口后可以依次挑战同一课程中的其余窗口类型——滑动窗口、会话窗口与全局窗口Katas 目录 learning/katas/java/Windowing 下均有对应练习并结合 sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing 下的SlidingWindows、Sessions、GlobalWindows源码对照学习。小结与易错点窗口化必须发生在聚合之前Window.into(...)要施加在待聚合的 PCollection 上聚合变换Count、GroupByKey、Combine才能按窗口独立执行顺序写反则聚合结果不分窗口。输入必须携带时间戳固定窗口依据元素的事件时间切分使用Create.timestamped或WithTimestamps显式设置没有时间戳的元素无法被正确分窗可参考同课程的 Adding Timestamp 练习。窗口是半开区间边界元素归属下一个窗口IntervalWindow的字符串表示[start..end)直观反映了这一点。调整边界用withOffset默认窗口从纪元对齐若需要按特定时区或业务零点对齐使用withOffset并确保偏移量满足0 offset size。固定时间窗口虽然简单却是理解 Beam 事件时间、窗口与聚合三者关系的基石。掌握它之后滑动窗口、会话窗口乃至触发器Trigger等更高级的流处理能力都将建立在这套一致的模型之上。赞分享大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载相关推荐Apache Beam Katas 实战用 Kotlin 实现 Fixed Time Window 固定时间窗口按天聚合计数Apache Beam Katas 实战用 Kotlin 实现 Fixed Time Window 固定时间窗口按天聚合计数 本指南以 Apache Beam大数据批处理流处理数据工程Apache Beam Go SDK 实战使用固定时间窗口Fixed Time Window处理带时间戳的 PCollectionApache Beam Go SDK 实战使用固定时间窗口Fixed Time Window处理带时间戳的 PCollection 固定时间窗口Fixe大数据批处理流处理数据工程Apache Beam 事件时间触发器Event Time Triggers实战用 AfterWatermark 实现 5 秒固定窗口计数Kotlin KataApache Beam 事件时间触发器Event Time Triggers实战用 AfterWatermark 实现 5 秒固定窗口计数Kotlin大数据批处理流处理数据工程上一篇日志迷雾中的导航者LogViewer如何重塑你的调试体验下一篇如何构建可维护和可扩展的自然语言理解系统Snips NLU最佳实践指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考