Package org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn

Class ReadChangeStreamPartitionDoFn

java.lang.Object

org.apache.beam.sdk.transforms.DoFn<PartitionMetadata,DataChangeRecord>

org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFn

All Implemented Interfaces:

Serializable, HasDisplayData

@UnboundedPerElement
public class ReadChangeStreamPartitionDoFn
extends DoFn<PartitionMetadata,DataChangeRecord>
implements Serializable

A SDF (Splittable DoFn) class which is responsible for performing a change stream query for a given partition. A different action will be taken depending on the type of record received from the query. This component will also reflect the partition state in the partition metadata tables.

The processing of a partition is delegated to the QueryChangeStreamAction.

See Also:

• Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.beam.sdk.transforms.DoFn

DoFn.AlwaysFetched, DoFn.BoundedPerElement, DoFn.BundleFinalizer, DoFn.Element, DoFn.FieldAccess, DoFn.FinishBundle, DoFn.FinishBundleContext, DoFn.GetInitialRestriction, DoFn.GetInitialWatermarkEstimatorState, DoFn.GetRestrictionCoder, DoFn.GetSize, DoFn.GetWatermarkEstimatorStateCoder, DoFn.Key, DoFn.MultiOutputReceiver, DoFn.NewTracker, DoFn.NewWatermarkEstimator, DoFn.OnTimer, DoFn.OnTimerContext, DoFn.OnTimerFamily, DoFn.OnWindowExpiration, DoFn.OnWindowExpirationContext, DoFn.OutputReceiver<T>, DoFn.ProcessContext, DoFn.ProcessContinuation, DoFn.ProcessElement, DoFn.RequiresStableInput, DoFn.RequiresTimeSortedInput, DoFn.Restriction, DoFn.Setup, DoFn.SideInput, DoFn.SplitRestriction, DoFn.StartBundle, DoFn.StartBundleContext, DoFn.StateId, DoFn.Teardown, DoFn.TimerFamily, DoFn.TimerId, DoFn.Timestamp, DoFn.TruncateRestriction, DoFn.UnboundedPerElement, DoFn.WatermarkEstimatorState, DoFn.WindowedContext

Constructor Summary

Constructors

Constructor	Description
ReadChangeStreamPartitionDoFn(DaoFactory daoFactory, MapperFactory mapperFactory, ActionFactory actionFactory, ChangeStreamMetrics metrics)	This class needs a DaoFactory to build DAOs to access the partition metadata tables and to perform the change streams query.

Method Summary

Modifier and Type	Method	Description
Instant	getInitialWatermarkEstimatorState(PartitionMetadata partition)
double	getSize(PartitionMetadata partition, TimestampRange range)
TimestampRange	initialRestriction(PartitionMetadata partition)	The restriction for a partition will be defined from the start and end timestamp to query the partition for.
ReadChangeStreamPartitionRangeTracker	newTracker(PartitionMetadata partition, TimestampRange range)
ManualWatermarkEstimator<Instant>	newWatermarkEstimator(Instant watermarkEstimatorState)
DoFn.ProcessContinuation	processElement(PartitionMetadata partition, RestrictionTracker<TimestampRange,com.google.cloud.Timestamp> tracker, DoFn.OutputReceiver<DataChangeRecord> receiver, ManualWatermarkEstimator<Instant> watermarkEstimator, DoFn.BundleFinalizer bundleFinalizer)	Performs a change stream query for a given partition.
void	setThroughputEstimator(BytesThroughputEstimator<DataChangeRecord> throughputEstimator)	Sets the estimator to calculate the backlog of this function.
void	setup()	Constructs instances for the PartitionMetadataDao, ChangeStreamDao, ChangeStreamRecordMapper, PartitionMetadataMapper, DataChangeRecordAction, HeartbeatRecordAction, ChildPartitionsRecordAction, PartitionStartRecordAction, PartitionEndRecordAction, PartitionEventRecordAction and QueryChangeStreamAction.

Methods inherited from class org.apache.beam.sdk.transforms.DoFn

getAllowedTimestampSkew, getInputTypeDescriptor, getOutputTypeDescriptor, populateDisplayData, prepareForProcessing

Methods inherited from class java.lang.Object

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

Constructor Details

ReadChangeStreamPartitionDoFn

public ReadChangeStreamPartitionDoFn(DaoFactory daoFactory,
 MapperFactory mapperFactory,
 ActionFactory actionFactory,
 ChangeStreamMetrics metrics)

This class needs a DaoFactory to build DAOs to access the partition metadata tables and to perform the change streams query. It uses mappers to transform database rows into the ChangeStreamRecord model. It uses the ActionFactory to construct the action dispatchers, which will perform the change stream query and process each type of record received. It emits metrics for the partition using the ChangeStreamMetrics.

Parameters:

daoFactory - the DaoFactory to construct PartitionMetadataDaos and ChangeStreamDaos

mapperFactory - the MapperFactory to construct ChangeStreamRecordMappers

actionFactory - the ActionFactory to construct actions

metrics - the ChangeStreamMetrics to emit partition related metrics

Method Details

getInitialWatermarkEstimatorState

@GetInitialWatermarkEstimatorState
public Instant getInitialWatermarkEstimatorState(@Element
 PartitionMetadata partition)

newWatermarkEstimator

@NewWatermarkEstimator
public ManualWatermarkEstimator<Instant> newWatermarkEstimator(@WatermarkEstimatorState
 Instant watermarkEstimatorState)

initialRestriction

@GetInitialRestriction
public TimestampRange initialRestriction(@Element
 PartitionMetadata partition)

The restriction for a partition will be defined from the start and end timestamp to query the partition for. The TimestampRange restriction represents a closed-open interval, while the start / end timestamps represent a closed-closed interval, so we add 1 nanosecond to the end timestamp to convert it to closed-open.

In this function we also update the partition state to PartitionMetadata.State.RUNNING.

Parameters:

partition - the partition to be queried

Returns:

the timestamp range from the partition start timestamp to the partition end timestamp + 1 nanosecond

getSize

@GetSize
public double getSize(@Element
 PartitionMetadata partition,
 @Restriction
 TimestampRange range)
               throws Exception

Throws:

Exception

newTracker

@NewTracker
public ReadChangeStreamPartitionRangeTracker newTracker(@Element
 PartitionMetadata partition,
 @Restriction
 TimestampRange range)

setup

@Setup
public void setup()

Constructs instances for the PartitionMetadataDao, ChangeStreamDao, ChangeStreamRecordMapper, PartitionMetadataMapper, DataChangeRecordAction, HeartbeatRecordAction, ChildPartitionsRecordAction, PartitionStartRecordAction, PartitionEndRecordAction, PartitionEventRecordAction and QueryChangeStreamAction.

processElement

@ProcessElement
public DoFn.ProcessContinuation processElement(@Element
 PartitionMetadata partition,
 RestrictionTracker<TimestampRange,com.google.cloud.Timestamp> tracker,
 DoFn.OutputReceiver<DataChangeRecord> receiver,
 ManualWatermarkEstimator<Instant> watermarkEstimator,
 DoFn.BundleFinalizer bundleFinalizer)

Performs a change stream query for a given partition. A different action will be taken depending on the type of record received from the query. This component will also reflect the partition state in the partition metadata tables.

The processing of a partition is delegated to the QueryChangeStreamAction.

Parameters:

partition - the partition to be queried

tracker - an instance of ReadChangeStreamPartitionRangeTracker

receiver - a DataChangeRecord DoFn.OutputReceiver

watermarkEstimator - a ManualWatermarkEstimator of Instant

bundleFinalizer - the bundle finalizer

Returns:

a DoFn.ProcessContinuation.stop() if a record timestamp could not be claimed or if the partition processing has finished

setThroughputEstimator

public void setThroughputEstimator(BytesThroughputEstimator<DataChangeRecord> throughputEstimator)

Sets the estimator to calculate the backlog of this function. Must be called after the initialization of this DoFn.

Parameters:

throughputEstimator - an estimator to calculate local throughput.