public class StreamingSemiAntiJoinOperator extends AbstractStreamingJoinOperator
AbstractStreamingJoinOperator.AssociatedRecords, AbstractStreamingJoinOperator.OuterRecord
AbstractStreamOperator.CountingOutput<OUT>
collector, joinCondition, LEFT_RECORDS_STATE_NAME, leftInputSideSpec, leftType, minRetentionTime, RIGHT_RECORDS_STATE_NAME, rightInputSideSpec, rightType, stateCleaningEnabled
chainingStrategy, config, latencyStats, LOG, metrics, output, timeServiceManager
Constructor and Description |
---|
StreamingSemiAntiJoinOperator(boolean isAntiJoin,
BaseRowTypeInfo leftType,
BaseRowTypeInfo rightType,
GeneratedJoinCondition generatedJoinCondition,
JoinInputSideSpec leftInputSideSpec,
JoinInputSideSpec rightInputSideSpec,
boolean[] filterNullKeys,
long minRetentionTime) |
Modifier and Type | Method and Description |
---|---|
void |
open()
This method is called immediately before any elements are processed, it should contain the
operator's initialization logic, e.g.
|
void |
processElement1(StreamRecord<BaseRow> element)
Process an input element and output incremental joined records, retraction messages will
be sent in some scenarios.
|
void |
processElement2(StreamRecord<BaseRow> element)
Process an input element and output incremental joined records, retraction messages will
be sent in some scenarios.
|
close
dispose, getChainingStrategy, getContainingTask, getCurrentKey, getExecutionConfig, getInternalTimerService, getKeyedStateBackend, getKeyedStateStore, getMetricGroup, getOperatorConfig, getOperatorID, getOperatorName, getOperatorStateBackend, getOrCreateKeyedState, getPartitionedState, getPartitionedState, getProcessingTimeService, getRuntimeContext, getUserCodeClassloader, initializeState, initializeState, notifyCheckpointComplete, numEventTimeTimers, numProcessingTimeTimers, prepareSnapshotPreBarrier, processLatencyMarker, processLatencyMarker1, processLatencyMarker2, processWatermark, processWatermark1, processWatermark2, reportOrForwardLatencyMarker, setChainingStrategy, setCurrentKey, setKeyContextElement1, setKeyContextElement2, setup, snapshotState, snapshotState
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
processLatencyMarker1, processLatencyMarker2, processWatermark1, processWatermark2
dispose, getChainingStrategy, getMetricGroup, getOperatorID, initializeState, prepareSnapshotPreBarrier, setChainingStrategy, setKeyContextElement1, setKeyContextElement2, snapshotState
notifyCheckpointComplete
getCurrentKey, setCurrentKey
public StreamingSemiAntiJoinOperator(boolean isAntiJoin, BaseRowTypeInfo leftType, BaseRowTypeInfo rightType, GeneratedJoinCondition generatedJoinCondition, JoinInputSideSpec leftInputSideSpec, JoinInputSideSpec rightInputSideSpec, boolean[] filterNullKeys, long minRetentionTime)
public void open() throws Exception
AbstractStreamOperator
The default implementation does nothing.
open
in interface StreamOperator<BaseRow>
open
in class AbstractStreamingJoinOperator
Exception
- An exception in this method causes the operator to fail.public void processElement1(StreamRecord<BaseRow> element) throws Exception
Following is the pseudo code to describe the core logic of this method.
if there is no matched rows on the other side if anti join, send input record if there are matched rows on the other side if semi join, send input record if the input record is accumulate, state.add(record, matched size) if the input record is retract, state.retract(record)
Exception
public void processElement2(StreamRecord<BaseRow> element) throws Exception
Following is the pseudo code to describe the core logic of this method.
if input record is accumulate | state.add(record) | if there is no matched rows on the other side, skip | if there are matched rows on the other side | | if the matched num in the matched rows == 0 | | if anti join, send -[other]s | | if semi join, send +[other]s | | if the matched num in the matched rows > 0, skip | | otherState.update(other, old+1) | endif endif if input record is retract | state.retract(record) | if there is no matched rows on the other side, skip | if there are matched rows on the other side | | if the matched num in the matched rows == 0, this should never happen! | | if the matched num in the matched rows == 1 | | if semi join, send -[other] | | if anti join, send +[other] | | if the matched num in the matched rows > 1, skip | | otherState.update(other, old-1) | endif endif
Exception
Copyright © 2014–2020 The Apache Software Foundation. All rights reserved.