序
本文主要研究一下flink的BoundedOutOfOrdernessTimestampExtractor
BoundedOutOfOrdernessTimestampExtractor
flink-streaming-java_2.11-1.7.0-sources.jar!/org/apache/flink/streaming/api/functions/timestamps/BoundedOutOfOrdernessTimestampExtractor.java
/**
* This is a {@link AssignerWithPeriodicWatermarks} used to emit Watermarks that lag behind the element with
* the maximum timestamp (in event time) seen so far by a fixed amount of time, <code>t_late</code>. This can
* help reduce the number of elements that are ignored due to lateness when computing the final result for a
* given window, in the case where we know that elements arrive no later than <code>t_late</code> units of time
* after the watermark that signals that the system event-time has advanced past their (event-time) timestamp.
* */
public abstract class BoundedOutOfOrdernessTimestampExtractor<T> implements AssignerWithPeriodicWatermarks<T> {
private static final long serialVersionUID = 1L;
/** The current maximum timestamp seen so far. */
private long currentMaxTimestamp;
/** The timestamp of the last emitted watermark. */
private long lastEmittedWatermark = Long.MIN_VALUE;
/**
* The (fixed) interval between the maximum seen timestamp seen in the records
* and that of the watermark to be emitted.
*/
private final long maxOutOfOrderness;
public BoundedOutOfOrdernessTimestampExtractor(Time maxOutOfOrderness) {
if (maxOutOfOrderness.toMilliseconds() < 0) {
throw new RuntimeException("Tried to set the maximum allowed " +
"lateness to " + maxOutOfOrderness + ". This parameter cannot be negative.");
}
this.maxOutOfOrderness = maxOutOfOrderness.toMilliseconds();
this.currentMaxTimestamp = Long.MIN_VALUE + this.maxOutOfOrderness;
}
public long getMaxOutOfOrdernessInMillis() {
return maxOutOfOrderness;
}
/**
* Extracts the timestamp from the given element.
*
* @param element The element that the timestamp is extracted from.
* @return The new timestamp.
*/
public abstract long extractTimestamp(T element);
@Override
public final Watermark getCurrentWatermark() {
// this guarantees that the watermark never goes backwards.
long potentialWM = currentMaxTimestamp - maxOutOfOrderness;
if (potentialWM >= lastEmittedWatermark) {
lastEmittedWatermark = potentialWM;
}
return new Watermark(lastEmittedWatermark);
}
@Override
public final long extractTimestamp(T element, long previousElementTimestamp) {
long timestamp = extractTimestamp(element);
if (timestamp > currentMaxTimestamp) {
currentMaxTimestamp = timestamp;
}
return timestamp;
}
}
- BoundedOutOfOrdernessTimestampExtractor抽象類實現(xiàn)AssignerWithPeriodicWatermarks接口的extractTimestamp及getCurrentWatermark方法亡资，同時聲明抽象方法extractAscendingTimestamp供子類實現(xiàn)
- BoundedOutOfOrdernessTimestampExtractor的構造器接收maxOutOfOrderness參數(shù)用于指定element允許滯后(
t-t_w棵介，t為element的eventTime谊却，t_w為前一次watermark的時間)的最大時間倦春，在計算窗口數(shù)據(jù)時私痹，如果超過該值則會被忽略 - BoundedOutOfOrdernessTimestampExtractor的extractTimestamp方法會調用子類的extractTimestamp方法抽取時間容握，如果該時間大于currentMaxTimestamp逢艘，則更新currentMaxTimestamp鳞青；getCurrentWatermark先計算potentialWM，如果potentialWM大于等于lastEmittedWatermark則更新lastEmittedWatermark(
currentMaxTimestamp - lastEmittedWatermark >= maxOutOfOrderness铃肯，這里表示lastEmittedWatermark太小了所以差值超過了maxOutOfOrderness患亿，因而調大lastEmittedWatermark)，最后返回Watermark(lastEmittedWatermark)
實例
public static void main(String[] args) throws Exception {
final int popThreshold = 20; // threshold for popular places
// set up streaming execution environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.getConfig().setAutoWatermarkInterval(1000);
// configure the Kafka consumer
Properties kafkaProps = new Properties();
kafkaProps.setProperty("zookeeper.connect", LOCAL_ZOOKEEPER_HOST);
kafkaProps.setProperty("bootstrap.servers", LOCAL_KAFKA_BROKER);
kafkaProps.setProperty("group.id", RIDE_SPEED_GROUP);
// always read the Kafka topic from the start
kafkaProps.setProperty("auto.offset.reset", "earliest");
// create a Kafka consumer
FlinkKafkaConsumer011<TaxiRide> consumer = new FlinkKafkaConsumer011<>(
"cleansedRides",
new TaxiRideSchema(),
kafkaProps);
// assign a timestamp extractor to the consumer
consumer.assignTimestampsAndWatermarks(new TaxiRideTSExtractor());
// create a TaxiRide data stream
DataStream<TaxiRide> rides = env.addSource(consumer);
// find popular places
DataStream<Tuple5<Float, Float, Long, Boolean, Integer>> popularPlaces = rides
// match ride to grid cell and event type (start or end)
.map(new GridCellMatcher())
// partition by cell id and event type
.keyBy(0, 1)
// build sliding window
.timeWindow(Time.minutes(15), Time.minutes(5))
// count ride events in window
.apply(new RideCounter())
// filter by popularity threshold
.filter((Tuple4<Integer, Long, Boolean, Integer> count) -> (count.f3 >= popThreshold))
// map grid cell to coordinates
.map(new GridToCoordinates());
popularPlaces.print();
// execute the transformation pipeline
env.execute("Popular Places from Kafka");
}
/**
* Assigns timestamps to TaxiRide records.
* Watermarks are a fixed time interval behind the max timestamp and are periodically emitted.
*/
public static class TaxiRideTSExtractor extends BoundedOutOfOrdernessTimestampExtractor<TaxiRide> {
public TaxiRideTSExtractor() {
super(Time.seconds(MAX_EVENT_DELAY));
}
@Override
public long extractTimestamp(TaxiRide ride) {
if (ride.isStart) {
return ride.startTime.getMillis();
}
else {
return ride.endTime.getMillis();
}
}
}
- 該實例使用的是AssignerWithPeriodicWatermarks押逼，通過env.getConfig().setAutoWatermarkInterval(1000)設置了watermark的時間間隔步藕，通過assignTimestampsAndWatermarks指定了AssignerWithPeriodicWatermarks為TaxiRideTSExtractor，它繼承了BoundedOutOfOrdernessTimestampExtractor抽象類
小結
- flink為了方便開發(fā)提供了幾個內置的Pre-defined Timestamp Extractors / Watermark Emitters挑格，其中一個就是BoundedOutOfOrdernessTimestampExtractor
- BoundedOutOfOrdernessTimestampExtractor抽象類實現(xiàn)AssignerWithPeriodicWatermarks接口的extractTimestamp及getCurrentWatermark方法咙冗，同時聲明抽象方法extractAscendingTimestamp供子類實現(xiàn)
- BoundedOutOfOrdernessTimestampExtractor的構造器接收maxOutOfOrderness參數(shù)用于指定element允許滯后(
t-t_w，t為element的eventTime漂彤，t_w為前一次watermark的時間)的最大時間雾消，在計算窗口數(shù)據(jù)時，如果超過該值則會被忽略
