org.apache.flink.streaming.runtime.operators.windowing

Class WindowOperator<K,IN,ACC,OUT,W extends Window>

java.lang.Object

org.apache.flink.streaming.api.operators.AbstractStreamOperator<OUT>

org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W>>

org.apache.flink.streaming.runtime.operators.windowing.WindowOperator<K,IN,ACC,OUT,W>

Type Parameters:

K - The type of key returned by the KeySelector.

IN - The type of the incoming elements.

OUT - The type of elements emitted by the InternalWindowFunction.

W - The type of Window that the WindowAssigner assigns.

All Implemented Interfaces:

Serializable, InputTypeConfigurable, OneInputStreamOperator<IN,OUT>, OutputTypeConfigurable<OUT>, StreamOperator<OUT>, Triggerable

Direct Known Subclasses:

EvictingWindowOperator

@Internal
public class WindowOperator<K,IN,ACC,OUT,W extends Window>
extends AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W>>
implements OneInputStreamOperator<IN,OUT>, Triggerable, InputTypeConfigurable

An operator that implements the logic for windowing based on a WindowAssigner and Trigger.

When an element arrives it gets assigned a key using a KeySelector and it gets assigned to zero or more windows using a WindowAssigner. Based on this, the element is put into panes. A pane is the bucket of elements that have the same key and same Window. An element can be in multiple panes if it was assigned to multiple windows by the WindowAssigner.

Each pane gets its own instance of the provided Trigger. This trigger determines when the contents of the pane should be processed to emit results. When a trigger fires, the given InternalWindowFunction is invoked to produce the results that are emitted for the pane to which the Trigger belongs.

See Also:

Serialized Form

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
class	WindowOperator.Context Context is a utility for handling Trigger invocations.
protected static class	WindowOperator.Timer<K,W extends Window> Internal class for keeping track of in-flight timers.

Nested classes/interfaces inherited from class org.apache.flink.streaming.api.operators.AbstractStreamOperator

AbstractStreamOperator.CountingOutput

Field Summary

Fields

Modifier and Type	Field and Description
protected long	allowedLateness The allowed lateness for elements.
protected WindowOperator.Context	context
protected long	currentWatermark To keep track of the current watermark so that we can immediately fire if a trigger registers an event time callback for a timestamp that lies in the past.
protected TypeSerializer<IN>	inputSerializer This is used to copy the incoming element because it can be put into several window buffers.
protected KeySelector<IN,K>	keySelector
protected TypeSerializer<K>	keySerializer For serializing the key in checkpoints.
protected Map<K,MergingWindowSet<W>>	mergingWindowsByKey
protected Map<Long,ScheduledFuture<?>>	processingTimeTimerFutures
protected Set<WindowOperator.Timer<K,W>>	processingTimeTimers
protected PriorityQueue<WindowOperator.Timer<K,W>>	processingTimeTimersQueue Processing time timers that are currently in-flight.
protected com.google.common.collect.Multiset<Long>	processingTimeTimerTimestamps
protected TimestampedCollector<OUT>	timestampedCollector This is given to the InternalWindowFunction for emitting elements with a given timestamp.
protected Trigger<? super IN,? super W>	trigger
protected Set<WindowOperator.Timer<K,W>>	watermarkTimers Current waiting watermark callbacks.
protected PriorityQueue<WindowOperator.Timer<K,W>>	watermarkTimersQueue
protected WindowAssigner<? super IN,W>	windowAssigner
protected WindowAssigner.WindowAssignerContext	windowAssignerContext
protected TypeSerializer<W>	windowSerializer For serializing the window in checkpoints.
protected StateDescriptor<? extends AppendingState<IN,ACC>,?>	windowStateDescriptor

Fields inherited from class org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator

userFunction

Fields inherited from class org.apache.flink.streaming.api.operators.AbstractStreamOperator

chainingStrategy, LOG, metrics, output

Constructor Summary

Constructors

Constructor and Description

WindowOperator(WindowAssigner<? super IN,W> windowAssigner, TypeSerializer<W> windowSerializer, KeySelector<IN,K> keySelector, TypeSerializer<K> keySerializer, StateDescriptor<? extends AppendingState<IN,ACC>,?> windowStateDescriptor, InternalWindowFunction<ACC,OUT,K,W> windowFunction, Trigger<? super IN,? super W> trigger, long allowedLateness) Creates a new WindowOperator based on the given policies and user functions.

Method Summary

Modifier and Type	Method and Description
void	close() This method is called after all records have been added to the operators via the methods OneInputStreamOperator.processElement(StreamRecord), or TwoInputStreamOperator.processElement1(StreamRecord) and TwoInputStreamOperator.processElement2(StreamRecord).
protected void	deleteCleanupTimer(W window) Deletes the cleanup timer set for the contents of the provided window.
void	dispose() This method is called at the very end of the operator's life, both in the case of a successful completion of the operation, and in the case of a failure and canceling.
KeySelector<IN,K>	getKeySelector()
protected MergingWindowSet<W>	getMergingWindowSet() Retrieves the MergingWindowSet for the currently active key.
StateDescriptor<? extends AppendingState<IN,ACC>,?>	getStateDescriptor()
Trigger<? super IN,? super W>	getTrigger()
WindowAssigner<? super IN,W>	getWindowAssigner()
protected boolean	isCleanupTime(W window, long time) Decides if it is time to clean up the window state.
protected boolean	isLate(W window) Decides if a window is currently late or not, based on the current watermark, i.e.
void	open() This method is called immediately before any elements are processed, it should contain the operator's initialization logic.
void	processElement(StreamRecord<IN> element) Processes one element that arrived at this operator.
void	processWatermark(Watermark mark) Processes a Watermark.
protected void	registerCleanupTimer(W window) Registers a timer to cleanup the content of the window.
void	restoreState(StreamTaskState taskState) Restores the operator state, if this operator's execution is recovering from a checkpoint.
void	setInputType(TypeInformation<?> type, ExecutionConfig executionConfig) Method that is called on an OutputFormat when it is passed to the DataSet's output method.
StreamTaskState	snapshotOperatorState(long checkpointId, long timestamp) Called to draw a state snapshot from the operator.
void	trigger(long time) This method is invoked with the timestamp for which the trigger was scheduled.

Methods inherited from class org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator

getUserFunction, getUserFunctionParameters, notifyOfCompletedCheckpoint, setOutputType, setup

Methods inherited from class org.apache.flink.streaming.api.operators.AbstractStreamOperator

getChainingStrategy, getContainingTask, getCurrentProcessingTime, getExecutionConfig, getMetricGroup, getOperatorConfig, getOperatorName, getPartitionedState, getPartitionedState, getRuntimeContext, getStateBackend, getUserCodeClassloader, registerTimer, setChainingStrategy, setKeyContext, setKeyContextElement1, setKeyContextElement2

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.flink.streaming.api.operators.StreamOperator

getChainingStrategy, getMetricGroup, notifyOfCompletedCheckpoint, setChainingStrategy, setKeyContextElement1, setKeyContextElement2, setup

Field Detail

windowAssigner

protected final WindowAssigner<? super IN,W extends Window> windowAssigner

keySelector

protected final KeySelector<IN,K> keySelector

trigger

protected final Trigger<? super IN,? super W extends Window> trigger

windowStateDescriptor

protected final StateDescriptor<? extends AppendingState<IN,ACC>,?> windowStateDescriptor

inputSerializer

protected TypeSerializer<IN> inputSerializer

This is used to copy the incoming element because it can be put into several window buffers.

keySerializer

protected final TypeSerializer<K> keySerializer

For serializing the key in checkpoints.

windowSerializer

protected final TypeSerializer<W extends Window> windowSerializer

For serializing the window in checkpoints.

allowedLateness

protected final long allowedLateness

The allowed lateness for elements. This is used for:

• Deciding if an element should be dropped from a window due to lateness.
• Clearing the state of a window if the system time passes the window.maxTimestamp + allowedLateness landmark.

timestampedCollector

protected transient TimestampedCollector<OUT> timestampedCollector

This is given to the InternalWindowFunction for emitting elements with a given timestamp.

processingTimeTimerFutures

protected transient Map<Long,ScheduledFuture<?>> processingTimeTimerFutures

currentWatermark

protected transient long currentWatermark

To keep track of the current watermark so that we can immediately fire if a trigger registers an event time callback for a timestamp that lies in the past.

context

protected transient WindowOperator.Context context

windowAssignerContext

protected transient WindowAssigner.WindowAssignerContext windowAssignerContext

processingTimeTimersQueue

protected transient PriorityQueue<WindowOperator.Timer<K,W extends Window>> processingTimeTimersQueue

Processing time timers that are currently in-flight.

processingTimeTimers

protected transient Set<WindowOperator.Timer<K,W extends Window>> processingTimeTimers

processingTimeTimerTimestamps

protected transient com.google.common.collect.Multiset<Long> processingTimeTimerTimestamps

watermarkTimers

protected transient Set<WindowOperator.Timer<K,W extends Window>> watermarkTimers

Current waiting watermark callbacks.

watermarkTimersQueue

protected transient PriorityQueue<WindowOperator.Timer<K,W extends Window>> watermarkTimersQueue

mergingWindowsByKey

protected transient Map<K,MergingWindowSet<W extends Window>> mergingWindowsByKey

Constructor Detail

WindowOperator

public WindowOperator(WindowAssigner<? super IN,W> windowAssigner,
                      TypeSerializer<W> windowSerializer,
                      KeySelector<IN,K> keySelector,
                      TypeSerializer<K> keySerializer,
                      StateDescriptor<? extends AppendingState<IN,ACC>,?> windowStateDescriptor,
                      InternalWindowFunction<ACC,OUT,K,W> windowFunction,
                      Trigger<? super IN,? super W> trigger,
                      long allowedLateness)

Creates a new WindowOperator based on the given policies and user functions.

Method Detail

setInputType

public final void setInputType(TypeInformation<?> type,
                               ExecutionConfig executionConfig)

Description copied from interface: InputTypeConfigurable

Method that is called on an OutputFormat when it is passed to the DataSet's output method. May be used to configures the output format based on the data type.

Specified by:

setInputType in interface InputTypeConfigurable

Parameters:

type - The data type of the input.

executionConfig - The execution config for this parallel execution.

open

public final void open()
                throws Exception

Description copied from class: AbstractStreamOperator

This method is called immediately before any elements are processed, it should contain the operator's initialization logic.

The default implementation does nothing.

Specified by:

open in interface StreamOperator<OUT>

Overrides:

open in class AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W extends Window>>

Throws:

Exception - An exception in this method causes the operator to fail.

close

public final void close()
                 throws Exception

Description copied from class: AbstractStreamOperator

This method is called after all records have been added to the operators via the methods OneInputStreamOperator.processElement(StreamRecord), or TwoInputStreamOperator.processElement1(StreamRecord) and TwoInputStreamOperator.processElement2(StreamRecord).

The method is expected to flush all remaining buffered data. Exceptions during this flushing of buffered should be propagated, in order to cause the operation to be recognized asa failed, because the last data items are not processed properly.

Specified by:

close in interface StreamOperator<OUT>

Overrides:

close in class AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W extends Window>>

Throws:

Exception - An exception in this method causes the operator to fail.

dispose

public void dispose()

Description copied from class: AbstractStreamOperator

This method is called at the very end of the operator's life, both in the case of a successful completion of the operation, and in the case of a failure and canceling. This method is expected to make a thorough effort to release all resources that the operator has acquired.

Specified by:

dispose in interface StreamOperator<OUT>

Overrides:

dispose in class AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W extends Window>>

processElement

public void processElement(StreamRecord<IN> element)
                    throws Exception

Description copied from interface: OneInputStreamOperator

Processes one element that arrived at this operator. This method is guaranteed to not be called concurrently with other methods of the operator.

Specified by:

processElement in interface OneInputStreamOperator<IN,OUT>

Throws:

Exception

processWatermark

public void processWatermark(Watermark mark)
                      throws Exception

Description copied from interface: OneInputStreamOperator

Processes a Watermark. This method is guaranteed to not be called concurrently with other methods of the operator.

Specified by:

processWatermark in interface OneInputStreamOperator<IN,OUT>

Throws:

Exception

See Also:

Watermark

trigger

public void trigger(long time)
             throws Exception

Description copied from interface: Triggerable

This method is invoked with the timestamp for which the trigger was scheduled.

If the triggering is delayed for whatever reason (trigger timer was blocked, JVM stalled due to a garbage collection), the timestamp supplied to this function will still be the original timestamp for which the trigger was scheduled.

Specified by:

trigger in interface Triggerable

Parameters:

time - The timestamp for which the trigger event was scheduled.

Throws:

Exception

getMergingWindowSet

protected MergingWindowSet<W> getMergingWindowSet()
                                           throws Exception

Retrieves the MergingWindowSet for the currently active key. The caller must ensure that the correct key is set in the state backend.

Throws:

Exception

isLate

protected boolean isLate(W window)

Decides if a window is currently late or not, based on the current watermark, i.e. the current event time, and the allowed lateness.

Parameters:

window - The collection of windows returned by the WindowAssigner.

Returns:

The windows (among the eligibleWindows) for which the element should still be considered when triggering.

registerCleanupTimer

protected void registerCleanupTimer(W window)

Registers a timer to cleanup the content of the window.

Parameters:

window - the window whose state to discard

deleteCleanupTimer

protected void deleteCleanupTimer(W window)

Deletes the cleanup timer set for the contents of the provided window.

Parameters:

window - the window whose state to discard

isCleanupTime

protected final boolean isCleanupTime(W window,
                                      long time)

Decides if it is time to clean up the window state. Clean up time for a window is:

if it is event time, after the watermark passes the end of the window plus the user-specified allowed lateness if it is processing time, after the processing time at the node passes the end of the window.

Parameters:

window - the window to clean

time - the current time (event or processing depending on the WindowAssigner

Returns:

true if it is time to clean up the window state, false otherwise.

snapshotOperatorState

public StreamTaskState snapshotOperatorState(long checkpointId,
                                             long timestamp)
                                      throws Exception

Description copied from interface: StreamOperator

Called to draw a state snapshot from the operator. This method snapshots the operator state (if the operator is stateful) and the key/value state (if it is being used and has been initialized).

Specified by:

snapshotOperatorState in interface StreamOperator<OUT>

Overrides:

snapshotOperatorState in class AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W extends Window>>

Parameters:

checkpointId - The ID of the checkpoint.

timestamp - The timestamp of the checkpoint.

Returns:

The StreamTaskState object, possibly containing the snapshots for the operator and key/value state.

Throws:

Exception - Forwards exceptions that occur while drawing snapshots from the operator and the key/value state.

restoreState

public void restoreState(StreamTaskState taskState)
                  throws Exception

Description copied from interface: StreamOperator

Restores the operator state, if this operator's execution is recovering from a checkpoint. This method restores the operator state (if the operator is stateful) and the key/value state (if it had been used and was initialized when the snapshot occurred).

This method is called after StreamOperator.setup(StreamTask, StreamConfig, Output) and before StreamOperator.open().

Specified by:

restoreState in interface StreamOperator<OUT>

Overrides:

restoreState in class AbstractUdfStreamOperator<OUT,InternalWindowFunction<ACC,OUT,K,W extends Window>>

Parameters:

taskState - The state of operator that was snapshotted as part of checkpoint from which the execution is restored.

Throws:

Exception - Exceptions during state restore should be forwarded, so that the system can properly react to failed state restore and fail the execution attempt.

getTrigger

public Trigger<? super IN,? super W> getTrigger()

getKeySelector

public KeySelector<IN,K> getKeySelector()

getWindowAssigner

public WindowAssigner<? super IN,W> getWindowAssigner()

getStateDescriptor

public StateDescriptor<? extends AppendingState<IN,ACC>,?> getStateDescriptor()