Run an integrated CDC pipeline in continuous mode

Applies to: Red X icon SaaS connectors Green check icon Database connectors Red X icon Query-based connectors

Important

This feature is in Beta. Workspace admins can control access to this feature from the Previews page. See Manage Azure Databricks previews.

Continuous mode runs an integrated CDC pipeline as an always-on stream instead of on a schedule. By default, an integrated CDC pipeline runs in triggered mode, where each update extracts and applies change data, then stops. Use continuous mode for:

  • Low-latency ingestion. Change data is applied to destination streaming tables as it arrives, typically within a few minutes, instead of waiting for the next scheduled update.
  • Sources with limited change-log retention. Some databases buffer changes in transaction logs that can grow large or be purged between updates. Running continuously keeps the pipeline caught up with the source, reducing the risk of falling behind the available log window.

Enable continuous mode

To run an integrated CDC pipeline in continuous mode, set continuous to true in the pipeline settings. The pipeline uses scale-optimized mode by default. For full pipeline creation steps, see the integrated pipeline page for your connector: Create an integrated CDC pipeline for SQL Server, Create an integrated CDC pipeline for MySQL, or Create an integrated CDC pipeline for Oracle.

Declarative Automation Bundles

resources:
  pipelines:
    continuous_cdc_pipeline:
      name: my-continuous-cdc-pipeline
      channel: PREVIEW
      continuous: true
      catalog: main
      schema: ingestion
      ingestion_definition:
        connection_name: my-sqlserver-connection
        connector_type: CDC
        objects:
          - table:
              source_catalog: my_database
              source_schema: dbo
              source_table: customers

Databricks notebook

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import (
    ConnectorType,
    IngestionConfig,
    IngestionPipelineDefinition,
    TableSpec,
)

w = WorkspaceClient()

pipeline = w.pipelines.create(
    name="my-continuous-cdc-pipeline",
    channel="PREVIEW",
    continuous=True,
    catalog="main",
    schema="ingestion",
    ingestion_definition=IngestionPipelineDefinition(
        connection_name="my-sqlserver-connection",
        connector_type=ConnectorType.CDC,
        objects=[
            IngestionConfig(
                table=TableSpec(
                    source_catalog="my_database",
                    source_schema="dbo",
                    source_table="customers",
                )
            )
        ],
    ),
)

print(f"Pipeline created: {pipeline.pipeline_id}")

Databricks CLI

databricks pipelines create --json '{
  "name": "my-continuous-cdc-pipeline",
  "channel": "PREVIEW",
  "continuous": true,
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers"
        }
      }
    ]
  }
}'

REST API

POST /api/2.0/pipelines

{
  "name": "my-continuous-cdc-pipeline",
  "channel": "PREVIEW",
  "continuous": true,
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers"
        }
      }
    ]
  }
}

Convert a triggered pipeline to continuous

To switch an existing triggered pipeline to continuous mode:

  1. Stop the current update, if one is running. See Stop the current update.
  2. Update the pipeline and set continuous to true.

Note

The update operation replaces the entire pipeline spec, so include the complete pipeline definition, not just the changed field.

Declarative Automation Bundles

Set continuous: true in the pipeline resource, then redeploy the bundle:

databricks bundle deploy

Databricks notebook

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

w.pipelines.update(
    pipeline_id="<pipeline-id>",
    name="my-continuous-cdc-pipeline",
    channel="PREVIEW",
    continuous=True,
    catalog="main",
    schema="ingestion",
    ingestion_definition=existing_ingestion_definition,
)

Databricks CLI

databricks pipelines update --json '{
  "pipeline_id": "<pipeline-id>",
  "name": "my-continuous-cdc-pipeline",
  "channel": "PREVIEW",
  "continuous": true,
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers"
        }
      }
    ]
  }
}'

REST API

PUT /api/2.0/pipelines/<pipeline-id>

{
  "pipeline_id": "<pipeline-id>",
  "name": "my-continuous-cdc-pipeline",
  "channel": "PREVIEW",
  "continuous": true,
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers"
        }
      }
    ]
  }
}

Run modes

Continuous CDC pipelines support two run modes:

Mode Description
Scale-optimized (default) Ingests up to 500 tables by internally rotating ingestion across them, without aggressive autoscaling. Use this mode to ingest a large number of tables in a single pipeline.
Speed-optimized (Beta) Runs the ingestion stream for all tables continuously to minimize ingestion latency, typically to within a few minutes. Speed-optimized mode supports up to 50 tables and uses aggressive autoscaling. Use speed-optimized mode when latency is the top priority.

The run mode is set by the pipelines.managedIngestion.continuous.runMode Spark configuration on the pipeline. Scale-optimized mode is the default. To enable speed-optimized mode, set runMode to SPEED.

Enable speed-optimized mode

To enable speed-optimized mode, set the pipelines.managedIngestion.continuous.runMode Spark configuration to SPEED when you create the pipeline, in addition to continuous: true:

Declarative Automation Bundles

resources:
  pipelines:
    continuous_cdc_pipeline:
      name: my-continuous-cdc-pipeline
      channel: PREVIEW
      continuous: true
      catalog: main
      schema: ingestion
      configuration:
        pipelines.managedIngestion.continuous.runMode: SPEED
      ingestion_definition:
        connection_name: my-sqlserver-connection
        connector_type: CDC
        objects:
          - table:
              source_catalog: my_database
              source_schema: dbo
              source_table: customers

Databricks notebook

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import (
    ConnectorType,
    IngestionConfig,
    IngestionPipelineDefinition,
    TableSpec,
)

w = WorkspaceClient()

pipeline = w.pipelines.create(
    name="my-continuous-cdc-pipeline",
    channel="PREVIEW",
    continuous=True,
    catalog="main",
    schema="ingestion",
    configuration={"pipelines.managedIngestion.continuous.runMode": "SPEED"},
    ingestion_definition=IngestionPipelineDefinition(
        connection_name="my-sqlserver-connection",
        connector_type=ConnectorType.CDC,
        objects=[
            IngestionConfig(
                table=TableSpec(
                    source_catalog="my_database",
                    source_schema="dbo",
                    source_table="customers",
                )
            )
        ],
    ),
)

print(f"Pipeline created: {pipeline.pipeline_id}")

Databricks CLI

databricks pipelines create --json '{
  "name": "my-continuous-cdc-pipeline",
  "channel": "PREVIEW",
  "continuous": true,
  "catalog": "main",
  "schema": "ingestion",
  "configuration": {
    "pipelines.managedIngestion.continuous.runMode": "SPEED"
  },
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers"
        }
      }
    ]
  }
}'

REST API

POST /api/2.0/pipelines

{
  "name": "my-continuous-cdc-pipeline",
  "channel": "PREVIEW",
  "continuous": true,
  "catalog": "main",
  "schema": "ingestion",
  "configuration": {
    "pipelines.managedIngestion.continuous.runMode": "SPEED"
  },
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers"
        }
      }
    ]
  }
}

Note

To use scale-optimized mode, remove the pipelines.managedIngestion.continuous.runMode configuration.

Select the right mode

Use the following comparison to choose between triggered mode and the two continuous run modes:

Capability Triggered Continuous (scale-optimized) Continuous (speed-optimized)
Maximum tables supported 300 500 50
Compute required Low (runs on a schedule) Medium (always on) High (always on with aggressive autoscaling)
Source load Low (runs on a schedule) High High (continuous queries)
Consistency Low (risk of change-log rollover) High High

Note

The limits in this table apply to continuous CDC pipelines. Individual connectors might have lower limits. See the documentation for your connector.

Stop the current update

Stop the running pipeline update before you convert a pipeline to continuous mode or run a selective full refresh. Replace <pipeline-id> with your pipeline's ID, which you can find in the pipelines UI.

Databricks notebook

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
w.pipelines.stop(pipeline_id="<pipeline-id>")

Databricks CLI

databricks pipelines stop <pipeline-id>

REST API

POST /api/2.0/pipelines/<pipeline-id>/stop

Fully refresh a subset of tables

In a continuous pipeline, you can fully refresh a subset of tables while all other tables continue to ingest within the same update. This is useful when a single table needs a full refresh (for example, after an incompatible schema change) without disrupting the rest of the pipeline.

To perform a selective full refresh:

  1. Stop the current update. See Stop the current update.

  2. Start a new update that lists the tables to fully refresh in full_refresh_selection and sets refresh_selection to the wildcard ["*"].

    Databricks notebook

    from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
w.pipelines.start_update(
    pipeline_id="<pipeline-id>",
    full_refresh_selection=["customers", "orders"],
    refresh_selection=["*"],
)

    Databricks CLI

    databricks pipelines start-update <pipeline-id> --json '{
  "full_refresh_selection": ["customers", "orders"],
  "refresh_selection": ["*"]
}'

    REST API

    POST /api/2.0/pipelines/<pipeline-id>/updates

{
  "full_refresh_selection": ["customers", "orders"],
  "refresh_selection": ["*"]
}

The tables in full_refresh_selection are fully refreshed, while all other tables continue refreshing within the same update. After the full refresh completes, the pipeline automatically resumes normal continuous ingestion for all tables. There is no need to stop the update and start a new one without full_refresh_selection.

Note

In continuous mode, one of the refresh selections must include the * wildcard so that all other tables keep ingesting while the selected tables are fully refreshed. Refreshing only a subset of tables without * (a partial refresh) is not supported in continuous mode.

Limitations

Continuous mode has the following limitations:

  • Updates restart to apply state changes. The pipeline uses a cancel-and-restart mechanism to reload the pipeline graph or apply state changes, such as schema changes.
  • Full refresh might require multiple restarts. A full refresh might require multiple pipeline restarts to complete, because the source snapshot is staged asynchronously.

Additional resources

  • Last updated on