CdcTrigger
class CdcTrigger(
conn: Conn,
tables: QualifiedTablesSpec,
cdc_format: CdcFormat = CdcFormat(values_format='columns', flatten_values=True),
buffer_max_rows: int = 10000,
buffer_max_bytes: int | None = None,
buffer_max_secs: int = 60,
trigger_max_rows: int | None = None,
trigger_max_bytes: int | None = None,
trigger_max_secs: int = 60,
poll_interval_seconds: int = 1,
start: datetime | None = None,
end: datetime | None = None,
)
Bases: StageTrigger
Categories: trigger
Base StageTrigger for Change Data Capture (CDC) connectors.
This abstract trigger provides the common polling loop, buffering, and staging logic for all CDC connectors. Database-specific subclasses implement the ingestion details for their respective replication protocols.
Initialize a CdcTrigger instance.
Parameters
connThe CDC connection configuration.
tablesQualified table names to track for changes.
cdc_formatThe output format specification for CDC data.
buffer_max_rowsMaximum rows in memory before flushing to the working directory.
buffer_max_bytesMaximum bytes in memory before flushing to the working directory.
buffer_max_secsMaximum seconds before flushing memory data to the working directory.
trigger_max_rowsMaximum total rows before staging from working directory to the stage location.
trigger_max_bytesMaximum total bytes before staging from working directory to the stage location.
trigger_max_secsMaximum seconds before staging from working directory to the stage location.
poll_interval_secondsInterval in seconds between polling for new changes.
startOptional datetime for the trigger to start working.
endOptional datetime for the trigger to stop working.
Properties
buffer_max_bytesint | NoneMaximum bytes in memory before flushing to the working directory.
buffer_max_rowsintMaximum rows in memory before flushing to the working directory.
buffer_max_secsintMaximum seconds before flushing memory data to the working directory.
cdc_formatCdcFormatOutput format specification for CDC data.
connConnCDC connection configuration.
enddatetime | Noneoutdict[int, str] | NoneMapping of table indices to qualified table names.
poll_interval_secondsintInterval in seconds between polling for new changes.
startdatetime | Nonetableslist[str] | NoneQualified table names being tracked for changes.
trigger_max_bytesint | NoneMaximum total bytes before staging from working directory to the stage location.
trigger_max_rowsint | NoneMaximum total rows before staging from working directory to the stage location.
trigger_max_secsintMaximum seconds before staging from working directory to the stage location.
Methods
collect_tablesdef collect_tables(
session: Any,
cdc_format: CdcFormat = CdcFormat(values_format='columns', flatten_values=True),
) -> dict[str, TrackedTable]
harvest_tablesdef harvest_tables(
session: Any,
cdc_format: CdcFormat = CdcFormat(values_format='columns', flatten_values=True),
) -> dict[str, TrackedTable]
rundef run(context: StageTriggerContext)
Continuously runs the stage trigger logic using the received context to trigger and monitor executions. SUCCESS if the StageTrigger.end datetime is reached. exit with 0. ERROR if any issue stops it from working or recovering. exit with 1. STOPPED upon a signal. exit with 0.