Flink之Watermark策略代码模板

public class FlinkWaterMark throws Exception {
  public static void main(String[] args) throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      // 获取数据源
      DataStreamSource<String> socketSource = env.socketTextStream("localhost", 8888);
      // 构造watermark生成策略,选择不生成watermark
      WatermarkStrategy<UserEvent2> watermark = WatermarkStrategy.noWatermarks();
      // 将构造完成的watermark分配给数据流
      SingleOutputStreamOperator<UserEvent2> source = socketSource.assignTimestampsAndWatermarks(watermark);
      // ...
      env.execute();
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

public class FlinkWaterMark throws Exception {
  public static void main(String[] args) throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      // 获取数据源
      DataStreamSource<String> socketSource = env.socketTextStream("localhost", 8888);
      // 构造watermark, 使用紧跟最大事件时间策略
      WatermarkStrategy<String> watermark = WatermarkStrategy.<String>forMonotonousTimestamps()
              // 抽取时间时间, 根据数据中实际情况选择
              .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                  @Override
                  public long extractTimestamp(String element, long recordTimestamp) {
                      /**
                       * 这里是样例代码,实际情况根据具体业务具体数据特性抽取对应的时间
                       **/
                      String time = element.split(",")[0];
                      long timestamp = Long.parseLong(time);
                      return timestamp;
                  }
              });
      // 将构造完成的watermark分配给数据流
      SingleOutputStreamOperator<UserEvent2> source = socketSource.assignTimestampsAndWatermarks(watermark);
      // ...
      env.execute();
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// 首先看这里,继承的BoundedOutOfOrdernessWatermarks
public class AscendingTimestampsWatermarks<T> extends BoundedOutOfOrdernessWatermarks<T> {

  /** Creates a new watermark generator with for ascending timestamps. */
  public AscendingTimestampsWatermarks() {
      super(Duration.ofMillis(0)); // 这里将容错时间设置为了0
  }
}
1
2
3
4
5
6
7
8

public class FlinkWaterMark throws Exception {
  public static void main(String[] args) throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      // 获取数据源
      DataStreamSource<String> socketSource = env.socketTextStream("localhost", 8888);
      // 构造watermark, 使用允许水位线乱序策略,并设置最大容错时间为2s
      WatermarkStrategy<String> watermark = WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(2000))
              // 抽取时间时间, 根据数据中实际情况选择
              .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                  @Override
                  public long extractTimestamp(String element, long recordTimestamp) {
                      /**
                       * 这里是样例代码,实际情况根据具体业务具体数据特性抽取对应的时间
                       **/
                      String time = element.split(",")[0];
                      long timestamp = Long.parseLong(time);
                      return timestamp;
                  }
              });
      // 将构造完成的watermark分配给数据流
      SingleOutputStreamOperator<UserEvent2> source = socketSource.assignTimestampsAndWatermarks(watermark);
      // ...
      env.execute();
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25