你读一下 EvictingWindowOperator 相关代码或者说 Evictor#evictBefore 的调用链，里面关于 window
state 的处理是比较 hack 的，用文字说也起不到简练的作用
private void emitWindowContents(W window, Iterable<StreamRecord<IN>>
contents, ListState<StreamRecord<IN>> windowState) throws Exception {
timestampedCollector.setAbsoluteTimestamp(window.maxTimestamp());
// Work around type system restrictions...
FluentIterable<TimestampedValue<IN>> recordsWithTimestamp = FluentIterable
.from(contents)
.transform(new Function<StreamRecord<IN>, TimestampedValue<IN>>() {
@Override
public TimestampedValue<IN> apply(StreamRecord<IN> input) {
return TimestampedValue.from(input);
}
});
evictorContext.evictBefore(recordsWithTimestamp,
Iterables.size(recordsWithTimestamp));
FluentIterable<IN> projectedContents = recordsWithTimestamp
.transform(new Function<TimestampedValue<IN>, IN>() {
@Override
public IN apply(TimestampedValue<IN> input) {
return input.getValue();
}
});
processContext.window = triggerContext.window;
userFunction.process(triggerContext.key, triggerContext.window,
processContext, projectedContents, timestampedCollector);
evictorContext.evictAfter(recordsWithTimestamp,
Iterables.size(recordsWithTimestamp));
//work around to fix FLINK-4369, remove the evicted elements from
the windowState.
//this is inefficient, but there is no other way to remove elements
from ListState, which is an AppendingState.
windowState.clear();
for (TimestampedValue<IN> record : recordsWithTimestamp) {
windowState.add(record.getStreamRecord());
}
}
Best,
tison.
USERNAME <
[hidden email]> 于2020年1月21日周二 下午8:25写道：
Locked
