Flink实时开发添加水印的案例分析
创始人
2025-01-08 01:33:14
0

在Flink中,处理时间序列数据时,通常需要考虑事件时间和水印(watermarks)的处理。以下是修改前后的代码对比分析:

修改前的代码:

val systemDS = unitDS.map(dp => {   dp.setDeviceCode(DeviceCodeEnum.fromPidToSystem(dp.getDeviceCode))   dp }).keyBy(_.getDeviceCode) .window(TumblingEventTimeWindows.of(Time.seconds(60))) .process(new MySystemWinF)
  1. unitDS 经过一个 map 操作,将每个元素的 deviceCode 转换为系统设备码。
  2. 使用 keyBy(_.getDeviceCode) 对转换后的设备码进行分组。
  3. 定义了一个基于事件时间的滚动窗口,窗口大小为60秒。
  4. 使用 process 操作应用自定义的窗口函数 HPageSystemWinF 来处理每个窗口中的数据。

注意:修改前的代码没有显示地处理水印(watermarks),这可能导致在处理乱序数据或延迟数据时出现问题。

修改后的代码:

val systemDS = unitDS.map(dp => {   dp.setDeviceCode(DeviceCodeEnum.fromPidToSystem(dp.getDeviceCode))   dp }).keyBy(_.getDeviceCode) .assignTimestampsAndWatermarks(   WatermarkStrategy     .forBoundedOutOfOrderness(Duration.ofSeconds(5)) // 假设这里应该是.forBoundedOutOfOrderness而不是.forBoundedOutOfOrdernessDaysPower     .withIdleness(Duration.ofSeconds(5))     .withTimestampAssigner(new SerializableTimestampAssigner[DaysPower] {       override def extractTimestamp(element: DaysPower, recordTimestamp: Long): Long = {         Math.max(element.getEventTime, recordTimestamp)       }     }) ).keyBy(_.getDeviceCode) .window(TumblingEventTimeWindows.of(Time.seconds(60))) .process(new MySystemWinF)
  1. 与修改前相同的部分:mapkeyBy, 和 window 操作。
  2. 添加了 assignTimestampsAndWatermarks 方法来处理事件时间和水印:
    • 使用 WatermarkStrategy.forBoundedOutOfOrderness 允许一定程度的乱序数据(这里是5秒)。
    • .withIdleness(Duration.ofSeconds(5)) 设置了空闲超时时间为5秒,用于处理不活跃的键。
    • 使用 withTimestampAssigner 自定义了时间戳分配器,确保使用的事件时间是元素中的 eventTime 和记录的 recordTimestamp 中的较大值。

不同点和适用场景:

  • 事件时间和水印处理:修改后的代码显式地处理了事件时间和水印,这对于处理乱序数据、延迟数据以及确保正确的时间窗口计算是非常重要的。如果您的数据流中存在乱序或延迟数据,或者您希望更严格地保证处理时间窗口的正确性,那么应该使用修改后的代码。
  • 空闲超时:通过设置空闲超时,可以处理那些长时间不活跃的键,避免因为某些键长时间没有新数据而导致整个程序挂起。
  • 延迟数据处理:如果数据有可能晚到,但仍然需要被纳入正确的窗口进行计算,水印可以帮助界定数据的“迟到”界限。
    精确的时间窗口分析:对于需要基于事件实际发生时间而非数据处理时间进行分析的场景,如实时监控、金融交易分析等,事件时间模型是必须的。

相关内容

热门资讯

现有关情况通报如下!哈灵永久辅... 现有关情况通报如下!哈灵永久辅助,佛手在线辅助器苹果版(好像有挂app)-哔哩哔哩;1、佛手在线辅助...
刚刚!老友广东麻将有没有假,打... 刚刚!老友广东麻将有没有假,打哈儿床将辅助最新(竟然存在有工具)-哔哩哔哩1、首先打开老友广东麻将有...
备受关注的!鱼虾蟹控制器辅助器... 备受关注的!鱼虾蟹控制器辅助器开挂,微信雀神挂件开挂辅助(一直有挂神器)-哔哩哔哩鱼虾蟹控制器辅助器...
最终!新道游辅助器,手机挂机软... 您好,新道游辅助器这款游戏可以开挂的,确实是有挂的,需要了解加去威信【136704302】很多玩家在...
昨日!微信小程序多乐跑得快破解... 昨日!微信小程序多乐跑得快破解,兴动互娱软件辅助下载(本来真的有工具)-哔哩哔哩1、在微信小程序多乐...
随着!河洛杠次脚本开发,斗棋辅... 随着!河洛杠次脚本开发,斗棋辅助脚本视频(好像真的是有器)-哔哩哔哩1、河洛杠次脚本开发辅助软件下载...
有玩家发现!乐逍遥脚本辅助,西... 有玩家发现!乐逍遥脚本辅助,西元红河辅助(一直有挂插件)-哔哩哔哩1、每一步都需要思考,不同水平的挑...
目前!人皇大厅控制牌型,花城棋... 目前!人皇大厅控制牌型,花城棋舍辅助(总是有挂脚本)-哔哩哔哩在进入人皇大厅控制牌型软件靠谱后,参与...
刚刚!卡农辅助,新九五辅助(一... 刚刚!卡农辅助,新九五辅助(一直是真的下载)-哔哩哔哩进入游戏-大厅左侧-新手福利-激活码辅助透视技...
据公告内容!衢州都莱辅助软件,... 据公告内容!衢州都莱辅助软件,天天微友挂(真是真的有下载)-哔哩哔哩1、首先打开衢州都莱辅助软件辅助...