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 get's 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 of 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.

This operator also needs a WindowBufferFactory to create a buffer for storing the elements of each pane.

See Also:

Serialized Form

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected 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.

Field Summary

Fields

Modifier and Type	Field and Description
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 Set<WindowOperator.Timer<K,W>>	processingTimeTimers Processing time timers that are currently in-flight.
protected PriorityQueue<WindowOperator.Timer<K,W>>	processingTimeTimersQueue
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 TypeSerializer<W>	windowSerializer For serializing the window in checkpoints.
protected StateDescriptor<? extends MergingState<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, 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 MergingState<IN,ACC>,?> windowStateDescriptor, InternalWindowFunction<ACC,OUT,K,W> windowFunction, Trigger<? super IN,? super W> trigger) 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).
KeySelector<IN,K>	getKeySelector()
StateDescriptor<? extends MergingState<IN,ACC>,?>	getStateDescriptor()
Trigger<? super IN,? super W>	getTrigger()
WindowAssigner<? super IN,W>	getWindowAssigner()
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.
protected void	processTriggerResult(TriggerResult triggerResult, K key, W window)
void	processWatermark(Watermark mark) Processes a Watermark.
void	restoreState(StreamTaskState taskState, long recoveryTimestamp) 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

dispose, getUserFunction, getUserFunctionParameters, notifyOfCompletedCheckpoint, setOutputType, setup

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

disableInputCopy, getChainingStrategy, getContainingTask, getExecutionConfig, getOperatorConfig, getPartitionedState, getPartitionedState, getRuntimeContext, getStateBackend, getUserCodeClassloader, isInputCopyingDisabled, 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

dispose, getChainingStrategy, isInputCopyingDisabled, 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 MergingState<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.

timestampedCollector

protected transient TimestampedCollector<OUT> timestampedCollector

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

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

processingTimeTimers

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

Processing time timers that are currently in-flight.

processingTimeTimersQueue

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

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

Constructor Detail

WindowOperator

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

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.

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.

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

processTriggerResult

protected void processTriggerResult(TriggerResult triggerResult,
                                    K key,
                                    W window)
                             throws Exception

Throws:

Exception

processWatermark

public final 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 final 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

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,
                         long recoveryTimestamp)
                  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.

recoveryTimestamp - Global recovery timestamp

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 MergingState<IN,ACC>,?> getStateDescriptor()