Flink中测量事件时间的进度的机制就是watermark(水印). watermark作为数据流的一部分在流动, 并且携带一个时间戳t。
一个Watermark(t)表示在这个流里面事件时间已经到了时间t, 意味着此时, 流中不应该存在这样的数据: 他的时间戳t2<=t (时间比较旧或者等于时间戳)。
1、衡量事件时间的进展。
2、是一个特殊的时间戳，生成之后随着流的流动而向后传递(广播方式)。
3、用来处理数据乱序的问题。
4、触发窗口等的计算、关闭。
5、单调递增的（时间不能倒退），与key无关。
6、Flink认为，小于Watermark时间戳的数据处理完了，不应该再出现。
事件是有序的(生成数据的时间和被处理的时间顺序是一致的), watermark是流中一个简单的周期性的标记。
有序场景： 1、底层调用的也是乱序的Watermark生成器，只是乱序程度传了一个0ms。 2、Watermark = maxTimestamp – outOfOrdernessMills – 1ms = maxTimestamp – 0ms – 1ms =>事件时间 – 1ms乱序流中的水印
按照时间戳来看, 事件是乱序的, 则watermark对于这些乱序的流来说至关重要。
乱序场景：
1、什么是乱序 => 时间戳大的比时间戳小的先来
2、乱序程度设置多少比较合适？
a)经验值 => 对自身集群和数据的了解，大概估算。
b)对数据进行抽样。 --->抽样，算最大乱序程度
c)肯定不会设置为几小时，一般设为 秒 或者 分钟。
3、Watermark = maxTimestamp – outOfOrdernessMills – 1ms
=>当前最大的事件时间 – 乱序程度（等待时间）- 1ms
EventTime
事件时间是指的这个事件发生的时间.
事件时间程序必须制定如何产生Event Time Watermarks(水印) . 在事件时间体系中, 水印是表示时间进度的标志(作用就相当于现实时间的时钟).
注意：在Flink1.12之前默认的时间语义是处理时间, 从1.12开始, Flink内部已经把默认的语义改成了事件时间。
//分配时间戳和水印 waterSensorDStream.assignTimestampsAndWatermarks(WatermarkStrategy. //水印策略:时间戳单调增长2.Fixed Amount of Lateness(允许固定时间的延迟) —>乱序forMonotonousTimestamps() //指定时间戳 .withTimestampAssigner(new SerializableTimestampAssigner () { @Override public long extractTimestamp(WaterSensor element, long recordTimestamp) { return element.getTs() * 1000; } }) );
//分配时间戳和水印 waterSensorDStream.assignTimestampsAndWatermarks(WatermarkStrategy. //水印策略:固定时间的延迟 .多并行度下WaterMark的传递 多并行度下WaterMark生成位置与传递测试forBoundedOutOfOrderness(Duration.ofSeconds(3)) //最大容忍的延迟时间 //指定时间戳 .withTimestampAssigner(new SerializableTimestampAssigner () { @Override public long extractTimestamp(WaterSensor element, long recordTimestamp) { return element.getTs() * 1000; } }) );
测试声明：全局并行度2
测试流程：source —> map —> keyby
1.测试数据准备
1001,1,12 1002,6,23
2.测试用JavaBean准备：WaterSensor
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class WaterSensor {
private String id;
private Long ts;
private Double vc;
}
3.测试程序1：MaxVC1 source（WaterMark） —> map —> keyby
import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class MaxVC1 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
DataStreamSource dataStreamSource = env.socketTextStream("hadoop102", 9999);
SingleOutputStreamOperator outputStreamOperator = dataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner() {
@Override
public long extractTimestamp(String element, long recordTimestamp) {
String[] fields = element.split(",");
return Long.parseLong(fields[1]) * 1000L;
}
}));
SingleOutputStreamOperator waterSensorDS = outputStreamOperator.map(line -> {
String[] fields = line.split(",");
return new WaterSensor(fields[0], Long.parseLong(fields[1]), Double.parseDouble(fields[2]));
});
SingleOutputStreamOperator result = waterSensorDS.keyBy(WaterSensor::getId)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.reduce(new ReduceFunction() {
@Override
public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
return new WaterSensor(value1.getId(), value1.getTs(), Math.max(value1.getVc(), value2.getVc()));
}
});
result.print();
env.execute();
}
}
| 输入 | 控制台打印输出 |
|---|---|
| 1001,1,12 | |
| 1002,6,23 | 1> WaterSensor(id=1001, ts=1, vc=12.0) |
4.测试程序2：MaxVC2 source —> map（WaterMark） —> keyby
import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class MaxVC2 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
DataStreamSource dataStreamSource = env.socketTextStream("hadoop102", 9999);
SingleOutputStreamOperator waterSensorDS = dataStreamSource.map(line -> {
String[] fields = line.split(",");
return new WaterSensor(fields[0], Long.parseLong(fields[1]), Double.parseDouble(fields[2]));
}).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner() {
@Override
public long extractTimestamp(WaterSensor element, long recordTimestamp) {
return element.getTs() * 1000L;
}
}));
SingleOutputStreamOperator result = waterSensorDS.keyBy(WaterSensor::getId)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.reduce(new ReduceFunction() {
@Override
public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
return new WaterSensor(value1.getId(), value1.getTs(), Math.max(value1.getVc(), value2.getVc()));
}
});
result.print();
env.execute();
}
}
| 输入 | 控制台打印输出 |
|---|---|
| 1001,1,12 | |
| 1002,6,23 | |
| 1002,6,23 | 1> WaterSensor(id=1001, ts=1, vc=12.0) |
1.多并行度的条件下, 向下游传递WaterMark的时候是以广播的方式传递的
2.总是以最小的那个WaterMark为准! 木桶原理!
3.并且当watermark值没有增长的时候不会向下游传递，注意：生成不变。
