統合CDCパイプラインを継続的モードでランする

適用対象 : SaaS コネクタ データベース コネクタ クエリーベースのコネクタ

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

継続的モードでは、統合されたCDCパイプラインがスケジュールではなく、常時稼働のストリームとして実行されます。By default、統合されたCDCパイプラインはTriggerモードで実行され、各更新で変更データを抽出して適用した後、停止します。継続的モードを使用する対象：

• 低レイテンシの取り込み。 変更データは、次のスケジュールされた更新を待つのではなく、到着次第（通常は数分以内）宛先のストリーミングテーブルに適用されます。
• 変更Logsの保持期間が制限されているソース。 一部のデータベースは、トランザクションLogsに変更をバッファリングしますが、これらは肥大化したり、更新の合間にパージされたりする可能性があります。継続的に実行することでパイプラインはソースに追従し続け、利用可能なLogsウィンドウから遅延するリスクを低減します。

連続モードを有効にする

統合CDCパイプラインを継続的モードで実行するには、パイプライン設定で continuous を true に設定します。パイプラインはdefaultでスケール最適化モードを使用します。パイプライン作成の全ステップについては、コネクターの統合パイプラインページを参照してください：SQL Server用の統合CDCパイプラインを作成する、MySQL用の統合CDCパイプラインを作成する、またはOracle用の統合CDCパイプラインを作成する。

Declarative Automation Bundles

YAML

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

Python

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

Shell

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

Text

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"
        }
      }
    ]
  }
}

Triggerパイプラインを連続パイプラインに変換する

既存のTriggerパイプラインを連続モードに切り替えるには：

1. 更新が実行されている場合は、現在の更新を停止します。現在の更新を停止を参照してください。
2. パイプラインを更新し、continuous を true に設定します。

注記

更新操作はパイプライン仕様全体を置き換えるため、変更されたフィールドだけでなく、パイプライン定義全体を含めてください。

Declarative Automation Bundles

パイプラインリソースで continuous: true を設定してから、バンドルを再デプロイします：

Shell

databricks bundle deploy

Databricks notebook

Python

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

Shell

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

Text

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"
        }
      }
    ]
  }
}

ランモード

継続的CDCパイプラインは、2つのランモードをサポートしています：

モード	説明
スケール最適化 (default)	積極的なオートスケールを行わずに、内部的に取り込みをローテーションさせることで、最大500個のテーブルを取り込みます。単一のパイプラインで多数のテーブルを取り込むには、このモードを使用します。
速度最適化 （ベータ）	すべてのテーブルの取り込みストリームを継続的に実行し、取り込みレイテンシを最小限（通常は数分以内）に抑えます。速度最適化モードは最大50個のテーブルをサポートし、積極的なオートスケールを使用します。レイテンシが最優先される場合は、速度最適化モードを使用してください。

モード	説明
スケール最適化 (default)	積極的なオートスケールを行わずに、内部的に取り込みをローテーションさせることで、最大500個のテーブルを取り込みます。単一のパイプラインで多数のテーブルを取り込むには、このモードを使用します。
速度最適化 （ベータ）	すべてのテーブルの取り込みストリームを継続的に実行し、取り込みレイテンシを最小限（通常は数分以内）に抑えます。速度最適化モードは最大50個のテーブルをサポートし、積極的なオートスケールを使用します。レイテンシが最優先される場合は、速度最適化モードを使用してください。

ランモードは、パイプラインの pipelines.managedIngestion.continuous.runMode Spark 構成によって設定されます。スケール最適化モードがdefaultです。速度最適化モードを有効にするには、runMode を SPEED に設定します。

速度最適化モードの有効化

速度最適化モードを有効にするには、パイプラインの作成時に continuous: true に加えて、pipelines.managedIngestion.continuous.runMode Spark 設定を SPEED に設定します：

Declarative Automation Bundles

YAML

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

Python

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

Shell

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

Text

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"
        }
      }
    ]
  }
}

注記

スケール最適化モードを使用するには、pipelines.managedIngestion.continuous.runMode構成を削除してください。

適切なモードを選択する

Triggerモードと2つの継続的ランモードのどちらかを選択するには、以下の比較を使用してください：

機能	トリガー	継続的（スケール最適化済み）	連続（速度最適化）
サポートされている最大テーブル数	300	500	50
コンピュートが必要です。	低（スケジュールで実行）	Medium（常時オン）	高（常にオン、積極的なオートスケールあり）
ソースのロード	低（スケジュールで実行）	高	高（継続的クエリー）
整合性	低 (変更 Logs のロールオーバーのリスク)	高	高

機能	トリガー	継続的（スケール最適化済み）	連続（速度最適化）
サポートされている最大テーブル数	300	500	50
コンピュートが必要です。	低（スケジュールで実行）	Medium（常時オン）	高（常にオン、積極的なオートスケールあり）
ソースのロード	低（スケジュールで実行）	高	高（継続的クエリー）
整合性	低 (変更 Logs のロールオーバーのリスク)	高	高

注記

この表の制限は、継続的なCDCパイプラインに適用されます。個々のコネクタには、より低い制限が適用される場合があります。コネクタのドキュメントを参照してください。

現在の更新を停止

パイプラインを連続モードに変換したり、選択的な完全更新を実行したりする前に、実行中のパイプライン更新を停止してください。<pipeline-id> をパイプラインの ID に置き換えます。ID は パイプライン UI で確認できます。

Databricks notebook

Python

from databricks.sdk import WorkspaceClient

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

Databricks CLI

Shell

databricks pipelines stop <pipeline-id>

REST API

Text

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

テーブルのサブセットを完全更新する

連続パイプラインでは、他のすべてのテーブルが同じ更新内で取り込みを継続している間に、テーブルのサブセットを完全更新できます。これは、パイプラインの残りの部分を中断することなく、単一のテーブルの完全更新が必要な場合（互換性のないスキーマ変更後など）に役立ちます。

選択的なフル更新を実行するには：

1. 現在の更新を停止します。現在の更新を停止するを参照してください。
2. full_refresh_selection で完全に更新するテーブルをリストし、 refresh_selection をワイルドカード ["*"] に設定する新しい更新を起動します。

Databricks notebook

Python

from databricks.sdk import WorkspaceClient

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

Databricks CLI

Shell

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

REST API

Text

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

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

full_refresh_selection 内のテーブルは完全に更新されますが、他のすべてのテーブルは同じ更新内で更新を継続します。完全更新が完了すると、パイプラインはすべてのテーブルに対して通常の継続的な取り込みを自動的に再開します。full_refresh_selection なしで更新を停止して新しく開始する必要はありません。

注記

継続的モードでは、選択したテーブルが完全更新されている間も他のすべてのテーブルが取り込みを継続できるように、更新の選択肢のいずれかに * ワイルドカードを含める必要があります。継続的モードでは、 * を使用せずにテーブルのサブセットのみを更新する（部分更新）ことはサポートされていません。

制限事項

継続的モードには以下の制限があります：

• 状態の変更を適用するために更新が再起動します。 パイプラインは、パイプライングラフの再読み込みや、スキーマ変更などの状態変更を適用するために、キャンセルおよび再起動のメカニズムを使用します。
• 完全更新には、複数回の再起動が必要になる場合があります。 ソースのスナップショットが非同期でステージングされるため、完全更新の完了にはパイプラインの再起動が複数回必要になる場合があります。

その他のリソース

• SQL Server 用の統合CDCパイプラインを作成する
• マネージド データベース コネクター
• 統合CDCパイプラインのスマートクロージャ
• ターゲットテーブルを完全に更新する