Flink Watermark详解 - ToB企服应用市场:ToB评测及商务社交产业平台

package com.wfg.flink.example.watermark;
import com.wfg.flink.example.watermark.data.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
import java.time.Instant;
public class FlinkWatermarkDemo {
public static void main(String[] args) throws Exception {
// 设置执行环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 假设我们有一个数据源，这里使用 fromElements 模拟
DataStream<Event> eventStream = env.fromElements(
new Event(1L, Instant.now().minusSeconds(10).toEpochMilli()),
new Event(2L, Instant.now().minusSeconds(8).toEpochMilli()),
new Event(3L, Instant.now().minusSeconds(12).toEpochMilli()),
new Event(4L, Instant.now().minusSeconds(15).toEpochMilli()),
new Event(5L, Instant.now().minusSeconds(19).toEpochMilli()),
new Event(6L, Instant.now().minusSeconds(18).toEpochMilli()),
new Event(7L, Instant.now().minusSeconds(22).toEpochMilli())
);
// 定义 Watermark 策略，允许 5 秒的乱序
WatermarkStrategy<Event> watermarkStrategy = WatermarkStrategy
.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((event, timestamp) -> {
// 从事件中提取时间戳
return event.getTimestamp();
// timestamp.assignTimestamp(event.getTimestamp());
});
// 应用 Watermark 策略，并处理数据流
DataStream<String> resultStream = eventStream
.assignTimestampsAndWatermarks(watermarkStrategy)
// 根据事件 ID 进行分区（这只是一个示例，实际可能根据业务需求分区）
.keyBy(Event::getId)
// 接下来可以进行窗口操作、时间聚合等操作
.map(new MapFunction<Event, String>() {
@Override
public String map(Event event) throws Exception {
return "Event ID: " + event.getId() + ", Timestamp: " + Instant.ofEpochMilli(event.getTimestamp());
}
});
// 输出结果
resultStream.print();
// 执行任务
env.execute("Flink Watermark Demo");
}
}

复制代码

package com.wfg.flink.example.watermark.data;
import lombok.Data;
/**
* @author wfg
*/
@Data
public class Event {
private final long id;
private final long timestamp;
public Event(long id, long timestamp) {
this.id = id;
this.timestamp = timestamp;
}
}

复制代码

// 定义 Watermark 策略，允许 5 秒的乱序
WatermarkStrategy<Event> watermarkStrategy = WatermarkStrategy
.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((event, timestamp) -> {
// 从事件中提取时间戳
return event.getTimestamp();
// timestamp.assignTimestamp(event.getTimestamp());
});

复制代码

// ...
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 假设有一个名为 eventStream 的 DataStream，其中包含具有时间戳的事件
DataStream<MyEvent> eventStream = ...; // 获取或创建事件流
// 创建一个 WatermarkStrategy，这里使用了一个允许一定乱序的 BoundedOutOfOrdernessTimestampExtractor
WatermarkStrategy<MyEvent> watermarkStrategy = WatermarkStrategy
.<MyEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner(new BoundedOutOfOrdernessTimestampExtractor<MyEvent>() {
@Override
public long extractTimestamp(MyEvent element) {
return element.getTimestamp(); // 假设 MyEvent 有一个 getTimestamp() 方法返回事件的时间戳
}
@Override
public long getMaxAllowedLatency(MyEvent element) {
return Duration.ofSeconds(10).toMillis(); // 最大允许乱序时间为10秒
}
});
// 为事件流分配时间戳和 Watermarks
DataStream<MyEvent> timestampedStream = eventStream.assignTimestampsAndWatermarks(watermarkStrategy);
// 现在可以基于事件时间进行窗口操作或其他时间感知的操作了
timestampedStream
.keyBy(event -> event.getKey()) // 假设 MyEvent 有一个 getKey() 方法
.timeWindow(Time.seconds(30)) // 使用基于事件时间的30秒窗口
.apply(new WindowFunction<MyEvent, String, String, TimeWindow>() {
// ... 实现 WindowFunction
})
.print(); // 打印结果或其他后续操作
// ...

复制代码