Package org.apache.flink.table.runtime.operators.window.slicing

Class SlicingWindowOperator<K,​W>

  • All Implemented Interfaces:
    Serializable, org.apache.flink.api.common.state.CheckpointListener, org.apache.flink.streaming.api.operators.Input<org.apache.flink.table.data.RowData>, org.apache.flink.streaming.api.operators.KeyContext, org.apache.flink.streaming.api.operators.KeyContextHandler, org.apache.flink.streaming.api.operators.OneInputStreamOperator<org.apache.flink.table.data.RowData,​org.apache.flink.table.data.RowData>, org.apache.flink.streaming.api.operators.SetupableStreamOperator<org.apache.flink.table.data.RowData>, org.apache.flink.streaming.api.operators.StreamOperator<org.apache.flink.table.data.RowData>, org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.CheckpointedStreamOperator, org.apache.flink.streaming.api.operators.Triggerable<K,​W>
    @Internal
public final class SlicingWindowOperator<K,​W>
extends TableStreamOperator<org.apache.flink.table.data.RowData>
implements org.apache.flink.streaming.api.operators.OneInputStreamOperator<org.apache.flink.table.data.RowData,​org.apache.flink.table.data.RowData>, org.apache.flink.streaming.api.operators.Triggerable<K,​W>, org.apache.flink.streaming.api.operators.KeyContext
    The SlicingWindowOperator implements an optimized processing for aligned windows which can apply the slicing optimization. The core idea of slicing optimization is to divide all elements from a data stream into a finite number of non-overlapping chunks (a.k.a. slices).

    Concept of Aligned Window and Unaligned Window

    We divide windows into 2 categories: Aligned Windows and Unaligned Windows.

    Aligned Windows are windows have predetermined window boundaries and windows can be divided into finite number of non-overlapping chunks. The boundary of an aligned window is determined independently from the time characteristic of the data stream, or messages it receives. For example, hopping (sliding) window is an aligned window as the window boundaries are predetermined based on the window size and slide. Aligned windows include tumbling, hopping, cumulative windows.

    Unaligned Windows are windows determined dynamically based on elements. For example, session window is an unaligned window as the window boundaries are determined based on the messages timestamps and their correlations. Currently, unaligned windows include session window only.

    Because aligned windows can be divided into finite number of non-overlapping chunks (a.k.a. slices), which can apply efficient processing to share intermediate results.

    Concept of Slice

    Dividing a window of aligned windows into a finite number of non-overlapping chunks, where the chunks are slices. It has the following properties:

    • An element must only belong to a single slice.
    • Slices are non-overlapping, i.e. S_i and S_j should not have any shared elements if i != j.
    • A window is consist of a finite number of slices.

    Abstraction of Slicing Window Operator

    A slicing window operator is a simple wrap of SlicingWindowProcessor. It delegates all the important methods to the underlying processor, where the processor can have different implementation for aggregate and topk or others.

    A SlicingWindowProcessor usually leverages the SliceAssigner to assign slices and calculate based on the slices. See SliceSharedWindowAggProcessor as an example.

    Note: since SlicingWindowProcessor leverages slicing optimization for aligned windows, therefore, it doesn't support unaligned windows, e.g. session window.

    Note: currently, SlicingWindowOperator doesn't support early-fire and late-arrival. Thus late elements (elements belong to emitted windows) will be simply dropped.

    See Also:
    Serialized Form

    • Field Summary

      Fields 
      Modifier and Type Field Description
      protected org.apache.flink.streaming.api.operators.TimestampedCollector<org.apache.flink.table.data.RowData> collector
      This is used for emitting elements with a given timestamp.

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

        chainingStrategy, config, latencyStats, LOG, metrics, output, processingTimeService

    • Method Summary

      All Methods Instance Methods Concrete Methods 
      Modifier and Type Method Description
      void close()  
      org.apache.flink.metrics.Counter getNumLateRecordsDropped()  
      org.apache.flink.metrics.Gauge<Long> getWatermarkLatency()  
      void initializeState​(org.apache.flink.runtime.state.StateInitializationContext context)  
      void onEventTime​(org.apache.flink.streaming.api.operators.InternalTimer<K,​W> timer)  
      void onProcessingTime​(org.apache.flink.streaming.api.operators.InternalTimer<K,​W> timer)  
      void open()  
      void prepareSnapshotPreBarrier​(long checkpointId)  
      void processElement​(org.apache.flink.streaming.runtime.streamrecord.StreamRecord<org.apache.flink.table.data.RowData> element)  
      void processWatermark​(org.apache.flink.streaming.api.watermark.Watermark mark)  
      void snapshotState​(org.apache.flink.runtime.state.StateSnapshotContext context)  

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

        finish, getChainingStrategy, getContainingTask, getCurrentKey, getExecutionConfig, getInternalTimerService, getKeyedStateBackend, getKeyedStateStore, getMetricGroup, getOperatorConfig, getOperatorID, getOperatorName, getOperatorStateBackend, getOrCreateKeyedState, getPartitionedState, getPartitionedState, getProcessingTimeService, getRuntimeContext, getTimeServiceManager, getUserCodeClassloader, hasKeyContext1, hasKeyContext2, initializeState, isUsingCustomRawKeyedState, notifyCheckpointAborted, notifyCheckpointComplete, processLatencyMarker, processLatencyMarker1, processLatencyMarker2, processWatermark1, processWatermark2, processWatermarkStatus, processWatermarkStatus1, processWatermarkStatus2, reportOrForwardLatencyMarker, setChainingStrategy, setCurrentKey, setKeyContextElement1, setKeyContextElement2, setProcessingTimeService, setup, snapshotState

      • Methods inherited from interface org.apache.flink.api.common.state.CheckpointListener

        notifyCheckpointAborted, notifyCheckpointComplete

      • Methods inherited from interface org.apache.flink.streaming.api.operators.Input

        processLatencyMarker, processWatermarkStatus

      • Methods inherited from interface org.apache.flink.streaming.api.operators.KeyContext

        getCurrentKey, setCurrentKey

      • Methods inherited from interface org.apache.flink.streaming.api.operators.KeyContextHandler

        hasKeyContext

      • Methods inherited from interface org.apache.flink.streaming.api.operators.OneInputStreamOperator

        setKeyContextElement

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

        finish, getMetricGroup, getOperatorID, initializeState, setKeyContextElement1, setKeyContextElement2, snapshotState

    • Field Detail

      • collector

        protected transient org.apache.flink.streaming.api.operators.TimestampedCollector<org.apache.flink.table.data.RowData> collector
        This is used for emitting elements with a given timestamp.

    • Method Detail

      • open

        public void open()
          throws Exception
        Specified by:
        open in interface org.apache.flink.streaming.api.operators.StreamOperator<K>
        Overrides:
        open in class TableStreamOperator<org.apache.flink.table.data.RowData>
        Throws:
        Exception

      • initializeState

        public void initializeState​(org.apache.flink.runtime.state.StateInitializationContext context)
                     throws Exception
        Specified by:
        initializeState in interface org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.CheckpointedStreamOperator
        Overrides:
        initializeState in class org.apache.flink.streaming.api.operators.AbstractStreamOperator<org.apache.flink.table.data.RowData>
        Throws:
        Exception

      • snapshotState

        public void snapshotState​(org.apache.flink.runtime.state.StateSnapshotContext context)
                   throws Exception
        Specified by:
        snapshotState in interface org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.CheckpointedStreamOperator
        Overrides:
        snapshotState in class org.apache.flink.streaming.api.operators.AbstractStreamOperator<org.apache.flink.table.data.RowData>
        Throws:
        Exception

      • close

        public void close()
           throws Exception
        Specified by:
        close in interface org.apache.flink.streaming.api.operators.StreamOperator<K>
        Overrides:
        close in class org.apache.flink.streaming.api.operators.AbstractStreamOperator<org.apache.flink.table.data.RowData>
        Throws:
        Exception

      • processElement

        public void processElement​(org.apache.flink.streaming.runtime.streamrecord.StreamRecord<org.apache.flink.table.data.RowData> element)
                    throws Exception
        Specified by:
        processElement in interface org.apache.flink.streaming.api.operators.Input<K>
        Throws:
        Exception

      • processWatermark

        public void processWatermark​(org.apache.flink.streaming.api.watermark.Watermark mark)
                      throws Exception
        Specified by:
        processWatermark in interface org.apache.flink.streaming.api.operators.Input<K>
        Overrides:
        processWatermark in class TableStreamOperator<org.apache.flink.table.data.RowData>
        Throws:
        Exception

      • onEventTime

        public void onEventTime​(org.apache.flink.streaming.api.operators.InternalTimer<K,​W> timer)
                 throws Exception
        Specified by:
        onEventTime in interface org.apache.flink.streaming.api.operators.Triggerable<K,​W>
        Throws:
        Exception

      • onProcessingTime

        public void onProcessingTime​(org.apache.flink.streaming.api.operators.InternalTimer<K,​W> timer)
                      throws Exception
        Specified by:
        onProcessingTime in interface org.apache.flink.streaming.api.operators.Triggerable<K,​W>
        Throws:
        Exception

      • prepareSnapshotPreBarrier

        public void prepareSnapshotPreBarrier​(long checkpointId)
                               throws Exception
        Specified by:
        prepareSnapshotPreBarrier in interface org.apache.flink.streaming.api.operators.StreamOperator<K>
        Overrides:
        prepareSnapshotPreBarrier in class org.apache.flink.streaming.api.operators.AbstractStreamOperator<org.apache.flink.table.data.RowData>
        Throws:
        Exception

      • getNumLateRecordsDropped

        @VisibleForTesting
public org.apache.flink.metrics.Counter getNumLateRecordsDropped()

      • getWatermarkLatency

        @VisibleForTesting
public org.apache.flink.metrics.Gauge<Long> getWatermarkLatency()