Create pipeline to CDC data into Databricks

Overview

A Change Data Capture (CDC) pipeline streams real-time changes from a CDC-enabled source database and loads them into Databricks. Etlworks ships a built-in CDC engine based on a customized version of Debezium — nothing to install or run separately, the CDC engine is part of the flow execution.

Supported source databases

• MySQL
• SQL Server
• PostgreSQL
• Oracle
• DB2
• MongoDB
• AS400 (IBMi)

When to use this pipeline

Use this pipeline to replicate changes from a CDC-enabled source database into Databricks in near-real time — typically with seconds to single-digit minutes of lag, depending on configured load frequency.

For loads triggered by a schedule rather than by source-database changes, use Any to Databricks with high-watermark change replication. For streams coming from a message queue rather than a database transaction log, use Stream messages from a queue into Databricks.

Flows optimized for Databricks

How does this flow work?

1. The CDC engine connects to the source database's transaction log and captures inserts, updates, and deletes in real time.
2. Captured events are serialized into staging files (CSV by default) in cloud storage or a Databricks Volume.
3. On a configurable interval (default 60 seconds), the flow loads the staged files into Databricks using COPY INTO.
4. For each destination table, the load uses MERGE semantics to apply inserts, updates, and deletes — honoring the CDC operation flags from the source.
5. The flow is checkpointed throughout. If streaming or loading fails, the pipeline resumes from the last successful position once the issue is resolved.

Single-flow architecture (recommended)

The default and recommended setup is a single flow that handles both streaming and loading. One flow to configure, schedule, and monitor; failures in either streaming or loading pause the same flow, so recovery is one resume action.

Prerequisites

• A CDC-enabled source database. See Database-specific CDC cases for per-database setup.
• A Databricks connection (PAT or OAuth Service Principal). See Get started with Databricks.
• A stage connection — S3, ADLS Gen2, GCS, server storage, or a Databricks Volume.
• An offset-backup connection (typically server storage or cloud storage) for CDC offset and history files.
• (Optional) A signal connection for high-watermark coordination across pipeline stages.

Create and schedule the Databricks CDC flow

Step 1. Create a CDC connection for the source

See Change Data Capture (CDC) from transaction log for the connection setup per source database. The CDC connection points the engine at the source database's transaction log and configures the captured tables.

Step 2. Create the stage connection

The stage holds the CSV files between extraction and load. Use:

• Amazon S3 for AWS-hosted workspaces.
• Azure Storage (ADLS Gen2) for Azure workspaces.
• Google Cloud Storage for GCP workspaces.
• Server storage pointed at a Databricks Volume (/Volumes/<catalog>/<schema>/<volume>/…) or a directory on the Etlworks host.

Step 3. Create the Databricks connection

See Get started with Databricks for connection setup.

Step 4. Create a connection for offsets and history

The CDC engine persists offsets and schema-history files to a backup connection so the pipeline can resume after a restart. Use server storage or cloud storage. The default location is {app.data}/cdc/<flow-name>/.

Step 5. Create the Databricks CDC flow

In Flows, click + and pick Stream CDC events into Databricks.

Step 6. Add the source-to-destination transformation

Set FROM to the CDC connection and the captured table (or a wildcard). Set TO to the Databricks destination — usually catalog.schema.* for multi-table replication.

Step 7. Set the staging connection on the flow

On the flow, set the named connection Stage to the stage connection from step 2. Set the named connection Offset Backup to the connection from step 4.

Step 8. Configure load parameters

Databricks-specific CDC-flow parameters:

Override the auto-generated COPY INTO SQL — optional

By default, Etlworks generates the COPY INTO statement from the destination table, the stage connection, the source file format, and the settings above. Set Override COPY INTO SQL when you need something the generator does not produce — a projection or a SELECT list on the source, COPY_OPTIONS (for example force or mergeSchema), a Databricks feature the generator does not know about, or a hand-tuned statement for a specific pipeline.

When this field is set, Etlworks uses your statement verbatim (after resolving the tokens below) instead of the generated one. The Format options setting in the table above and the auto-injected WITH (CREDENTIAL ...) clause are both bypassed — include everything the load needs in your statement.

Available tokens

Reference these tokens in your statement. The runtime resolves them per file, per batch, per destination table.

Structure of the auto-generated statement

Useful as a starting template for the override. The generator produces this shape:

COPY INTO <target-table>
  FROM '<stage-uri>/{FILE_TO_LOAD}[.{EXT} | *.*]'
  [WITH (CREDENTIAL <name-or-inline>)]
  FILEFORMAT = <CSV | JSON | PARQUET | AVRO>
  [FORMAT_OPTIONS (...)]

Details:

• Target: {TEMP_TABLE} for MERGE strategies, {TABLE} otherwise.
• Stage URI: composed from the stage connection's transport (S3, Azure Data Lake Storage Gen2, GCS, or server storage / Databricks Volume), bucket / container, and folder.
• File pattern: {FILE_TO_LOAD}.{EXT} for a single file, or {FILE_TO_LOAD}*.* when the flow is loading a wildcard match, split files, or archive-wrapped files (ZIP / GZIP).
• Credential: injected automatically when the Storage Credential setting is set or the stage connection has embedded AWS access / secret keys or an Azure SAS token. Skipped when the workspace has a Unity Catalog External Location or a managed Volume covering the stage path.
• FILEFORMAT: derived from the stage file extension.
• FORMAT_OPTIONS: emitted only for CSV, composed from header, delimiter, quote, line separator, date format, and timestamp format. The Format options setting above overrides just this clause without replacing the whole statement — use it in place of Override COPY INTO SQL when only parser options need to change.

Examples

MERGE flow — load into the temp Delta table with a custom escape and date/time formats:

COPY INTO {TEMP_TABLE}
FROM 's3://etlworks-stage/loads/{FILE_TO_LOAD}'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'delimiter' = ',', 'quote' = '"',
                'dateFormat' = 'yyyy-MM-dd',
                'timestampFormat' = 'yyyy-MM-dd HH:mm:ss',
                'escape' = '"')

INSERT flow — load Parquet directly into the destination:

COPY INTO {TABLE}
FROM 's3://etlworks-stage/loads/{FILE_TO_LOAD}.{EXT}'
FILEFORMAT = PARQUET

Advanced — load selected columns with a SELECT list and force re-loads:

COPY INTO {TABLE} (id, name, updated_at)
FROM (SELECT id, name, current_timestamp() AS updated_at
      FROM 's3://etlworks-stage/loads/{FILE_TO_LOAD}.{EXT}')
FILEFORMAT = PARQUET
COPY_OPTIONS ('force' = 'true')

Interaction with other settings

• Format options is bypassed. Include your custom FORMAT_OPTIONS in the override statement itself.
• Storage Credential is bypassed. Include a WITH (CREDENTIAL ...) clause in your statement if the load needs one; the generator's automatic credential injection does not run when the SQL is overridden.
• Before COPY SQL and After COPY SQL still run — they are separate statements executed against the Databricks connection around your COPY INTO, not part of it.
• MERGE strategies (DELETE / INSERT, Native MERGE, Native MERGE with separate DELETE) still run on top of the loaded temp table. Your override handles only the load-into-temp step, not the MERGE step itself. Use {TEMP_TABLE} as the target when the flow uses MERGE, and let Etlworks generate the MERGE statement that follows.

When to use Override COPY INTO SQL vs Format options

• Only parser options need to change (escape, multiLine, quote handling, header, delimiter, date / timestamp format) — use Format options. Safer and narrower; keeps the generated target, stage URI, and credential handling intact.
• The load needs a projection or SELECT list on the source — use Override COPY INTO SQL. The generator emits no SELECT list.
• The load needs COPY_OPTIONS (for example force, mergeSchema, or a custom copy-options map) — use Override COPY INTO SQL. The generator does not emit COPY_OPTIONS.
• A hand-tuned statement is required for a specific pipeline, or a Databricks feature the generator does not know about needs to appear in the SQL — use Override COPY INTO SQL.

Step 9. Schedule the CDC flow

CDC flows run continuously. Schedule the flow with a continuous-execution schedule so it auto-restarts after restarts or failures.

Monitor a running CDC flow

While a CDC flow runs, the streaming progress, current offset, processed events, and pending load batches are all visible from the flow execution view. See CDC Configuration and Monitoring.

Two-flow alternative: separate extract and load

For very high throughput or different scaling requirements on extraction vs. loading, you can split the pipeline into two flows:

1. A CDC Extract flow that streams events from the source database to staged CSV files in the stage.
2. A Bulk load files into Databricks flow scheduled to run periodically against the same stage.

This decouples streaming throughput from Databricks load frequency at the cost of an extra flow to operate. See the equivalent two-flow pattern documented in Create pipeline to CDC data into Snowflake — the Databricks variant follows the same shape.

Troubleshooting

For database-specific CDC setup issues, see Database-specific CDC cases. For common Databricks load issues, see Common issues when loading data into cloud data warehouses. For CDC tips, see Tips and Tricks for CDC Flows.