SQL Server用の統合 CDC パイプラインを作成する

統合 CDC パイプラインは、1 つのパイプラインを使用して変更データをSQL ServerからAzure Databricksに取り込む。 独立したインジェスト ゲートウェイとインジェスト パイプラインを必要とする標準的なゲートウェイ ベースのアーキテクチャとは異なり、統合された CDC パイプラインは、1 つのパイプライン更新で抽出ステージとアプリケーション ステージの両方を実行します。

統合 CDC コネクタを使用する場合

次の表は、統合 CDC パイプラインと標準ゲートウェイ ベースのアーキテクチャを比較したものです。

ソース データベースのセットアップについては、Azure Databricks への取り込み用に Microsoft SQL Server を構成するを参照してください。 同じソース構成が両方のアーキテクチャに適用されます。

統合 CDC パイプラインの実行方法

各パイプライン更新では、次の 2 つのステージが順番に実行されます。

1. 抽出。 パイプラインは、Unity カタログ接続を使用してソース データベースに接続します。 初回実行時または完全更新時には、初期スナップショットがキャプチャされます。 後続の実行では、データベースの組み込みの変更追跡メカニズムを使用して、増分変更 (挿入、更新、および削除) をキャプチャします。 パイプラインは、抽出されたデータを Unity カタログステージング ボリュームに書き込みます。
2. アプリケーション。 パイプラインはステージング ボリュームから読み取り、Unity カタログの宛先ストリーミング テーブルに変更を適用します。 マージ操作では、構成された主キーと SCD の種類が使用されます。 このパイプラインは、各データが厳密に1回だけ処理されることを保証します。

各パイプライン更新では、変更が抽出され、ソースに追い付いた後、最大ランタイムによって制限された状態で自動的に停止されます。 詳細については、 統合 CDC パイプラインのスマート クロージャに関するページを参照してください。 定期的にデータを取り込むには、 Lakeflow ジョブ タスクを使用してパイプラインをスケジュールします。

Requirements

• ワークスペースが Unity Catalog に対して有効になっている。

• 接続を作成する場合: メタストアに対する CREATE CONNECTION 特権があります。 「Unity Catalog の特権の管理」を参照してください。

  コネクタで UI ベースのパイプライン作成がサポートされている場合は、このページの手順を完了することで、接続とパイプラインを同時に作成できます。 ただし、API ベースのパイプライン作成を使用する場合は、このページの手順を完了する前に、カタログ エクスプローラーで接続を作成する必要があります。 「マネージド インジェスト ソースへの接続」を参照してください。

• 既存の接続を使用する場合: 接続に USE CONNECTION 特権または ALL PRIVILEGES があります。

• ターゲット カタログに対する USE CATALOG 権限があります。

• 既存のスキーマに対する USE SCHEMA、 CREATE TABLE、および CREATE VOLUME 特権、またはターゲット カタログに対する CREATE SCHEMA 特権があります。

• ワークスペースで統合 CDC コネクタ機能が有効になっている必要があります。 Azure Databricks アカウント チームにお問い合わせください。
• primary SQL Server インスタンスにアクセスできます。 統合 CDC コネクタは、読み取りレプリカ、スタンバイ インスタンス、またはセカンダリ インスタンスをサポートしていません。
• SQL Server ソースのセットアップが完了しました。 Azure DatabricksへのインジェストのためのMicrosoft SQL Serverの構成を参照してください。
• 次のアクセス許可があります。
  • メタストアで CREATE CONNECTION（新しい Unity Catalog 接続を作成する場合）、または既存の接続で USE CONNECTION。
  • 宛先のカタログ上の USE CATALOG。
  • 宛先スキーマ上の USE SCHEMA と CREATE TABLE。
  • 宛先のスキーマ上の CREATE VOLUME、または data_staging_options で指定されたスキーマ上 ステージング ボリュームは、 data_staging_options が設定されていない場合でも必要です。これは、パイプラインによって宛先スキーマ内のボリュームが自動的に作成されるためです。

コンピューティング要件

統合 CDC パイプラインは、クラシック コンピューティングまたはサーバーレス コンピューティングで実行されます。

• クラシックコンピューティング: クラシックコンピューティングプレーンは、Azure Databricksワークスペース VPC または VNet で実行され、ネットワーク経由でSQL Serverインスタンスに到達できる必要があります。 VPC または VNet ピアリング、パブリック エンドポイント、オンプレミスのSQL Server、AWS Direct Connect、Azure ExpressRoute、VPN など、コンピューティング プレーンがデータベースに到達できるようにするネットワーク パスがサポートされます。
• サーバーレス コンピューティング: Azure Databricks サーバーレス コンピューティングとソース データベース間のサーバーレス ネットワーク接続を構成します。 オンプレミスのソースには、構成されたサーバーレス エグレス (ExpressRoute または VPN を使用したトランジット ゲートウェイやピアリングされた VNet など) を経由するネットワーク パスが必要です。

クラシック コンピューティングの場合は、無制限のクラスター作成アクセス許可またはカスタム クラスター ポリシー を使用できます。 cluster_type は dltに固定され、 runtime_engine は STANDARDに固定され、効率的な抽出に推奨されるコアは少なくとも 8 個です。

SQL Server への Unity カタログ接続を作成する

パイプラインを作成する前に、SQL Serverへの Unity カタログ接続を作成します。 「SQL Server接続の作成を参照してください。

統合 CDC パイプラインを作成する

API、Databricks CLI、ノートブック、または宣言型オートメーション バンドルを使用して、統合 CDC パイプラインを作成します。 UI の作成機能はまだ利用できません。

宣言型オートメーション バンドル

バンドル ファイルでパイプライン リソースを定義します (例: resources/integrated_cdc_pipeline.yml)。

variables:
  pipeline_name:
    description: 'Name for the integrated CDC pipeline'
  connection_name:
    description: 'Unity Catalog connection name'
  dest_catalog:
    description: 'Destination catalog for ingested data'
  dest_schema:
    description: 'Destination schema for ingested data'

resources:
  pipelines:
    integrated_cdc_pipeline:
      name: ${var.pipeline_name}
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        connector_type: CDC
        objects:
          - table:
              source_catalog: 'my_database'
              source_schema: 'dbo'
              source_table: 'customers'
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: 'customers'
              table_configuration:
                scd_type: 'SCD_TYPE_1'

スケジュールに従ってパイプラインを実行するには、パイプラインをトリガーするジョブ ( resources/integrated_cdc_job.ymlなど) を定義します。 各抽出ステージは少なくとも 10 分間実行されるため、60 分以上の間隔が適しています。

resources:
  jobs:
    integrated_cdc_job:
      name: '${var.pipeline_name}-job'
      tasks:
        - task_key: 'cdc_ingestion'
          pipeline_task:
            pipeline_id: ${resources.pipelines.integrated_cdc_pipeline.id}
      schedule:
        quartz_cron_expression: '0 0 * * * ?'
        timezone_id: 'UTC'

Databricks CLI を使用してバンドルをデプロイします。

databricks bundle deploy
databricks bundle run integrated_cdc_job

詳細については、「 宣言型オートメーション バンドルとは」を参照してください。

Databricks ノートブック

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

w = WorkspaceClient()

pipeline = w.pipelines.create(
    name="<pipeline-name>",
    channel="PREVIEW",
    catalog="<destination-catalog>",
    schema="<destination-schema>",
    ingestion_definition=IngestionPipelineDefinition(
        connection_name="<unity-catalog-connection-name>",
        connector_type=ConnectorType.CDC,
        objects=[
            IngestionConfig(
                table=TableSpec(
                    source_catalog="<source-database>",
                    source_schema="<source-schema>",
                    source_table="<source-table>",
                    destination_catalog="<destination-catalog>",
                    destination_schema="<destination-schema>",
                )
            )
        ],
    ),
)

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

Databricks コマンドラインインターフェース (CLI)

databricks pipelines create --json '{
  "name": "<pipeline-name>",
  "channel": "PREVIEW",
  "catalog": "<destination-catalog>",
  "schema": "<destination-schema>",
  "ingestion_definition": {
    "connection_name": "<unity-catalog-connection-name>",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "<source-database>",
          "source_schema": "<source-schema>",
          "source_table": "<source-table>"
        }
      }
    ]
  }
}'

REST API

次の例では、SQL Server データベースから 2 つのテーブルをレプリケートします。 customers テーブルでは SCD Type 1 が使用され、orders テーブルでは SCD Type 2 が使用されます (ソースに CDC SQL Server必要)。 どちらも最上位レベルの宛先 main.ingestionを継承します。 この例では、 serverlessを省略します。既定値は false (クラシック コンピューティング) です。 代わりに、サーバーレス コンピューティングで実行する "serverless": true を追加します。

POST /api/2.0/pipelines

{
  "name": "my-integrated-cdc-pipeline",
  "channel": "PREVIEW",
  "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",
          "table_configuration": {
            "scd_type": "SCD_TYPE_1"
          }
        }
      },
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "orders",
          "table_configuration": {
            "scd_type": "SCD_TYPE_2"
          }
        }
      }
    ],
    "data_staging_options": {
      "catalog_name": "main",
      "schema_name": "ingestion_staging"
    }
  }
}

ソース スキーマ内のすべてのテーブルをレプリケートするには、個々のschema オブジェクトではなく、table オブジェクトを使用します。 パイプラインは、ソースで CDC または変更追跡が有効になっていないテーブルをスキップします。

POST /api/2.0/pipelines

{
  "name": "my-integrated-cdc-schema-pipeline",
  "channel": "PREVIEW",
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "schema": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "destination_catalog": "main",
          "destination_schema": "ingestion"
        }
      }
    ]
  }
}

パイプラインの更新を開始するには:

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

{
  "full_refresh": false
}

定期的な更新をスケジュールする

デフォルトでは、統合されたCDCパイプラインはトリガーモードで動作します。 常時オンの実行については、「 連続モードで統合されたCDCパイプラインを実行する」を参照してください。 定期的なスケジュールでデータを取り込むには、パイプラインを実行する Lakeflow ジョブ タスクを作成します。 更新期間は、ソースに含まれる変更データの量によって異なります。また、大規模なバックログが 1 回の更新で完了しない可能性があります ( 統合 CDC パイプラインのスマート クロージャを参照)。 後続の更新が追いつくために十分な頻度でパイプラインをスケジュールします。 60 分の開始点は、ほとんどのワークロードに適しています。 以前の更新プログラムの実行中にトリガーが起動した場合、新しい更新プログラムはキューに登録されます。

構成のリファレンス

パイプライン パラメーター

テーブルの仕様

テーブル構成

データ型マッピングSQL Serverについては、SQL Server コネクタリファレンスを参照してください。 統合 CDC パイプラインでは、自動型拡大がサポートされます。ソース列の型が拡大された場合 (たとえば、 INT から BIGINT)、変換先テーブルは自動的に調整されます。

パイプラインを監視する

統合 CDC パイプラインを作成して開始したら、次を使用してその状態を監視します。

• AZURE DATABRICKS UI。 [パイプライン] セクションでパイプラインを開き、更新の状態、テーブルごとのインジェスト メトリック、および系列を表示します。

• REST API。

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

• Events API。

  GET /api/2.0/pipelines/<pipeline-id>/events
  

最初のパイプライン更新では、選択したすべてのテーブルの完全なスナップショットが実行されます。増分更新よりも時間がかかる場合があります。 大規模なテーブルの場合、初期スナップショットの完了には複数のスケジュールされた更新が必要になる場合があります。 後続の各更新プログラムは、前の更新プログラムが終了したところから続行されます。

インジェストを確認するには:

-- Check row counts in the destination table
SELECT COUNT(*) FROM <destination_catalog>.<destination_schema>.<destination_table>;

-- View recent changes (SCD Type 2 tables)
SELECT * FROM <destination_catalog>.<destination_schema>.<destination_table>
ORDER BY __START_AT DESC
LIMIT 10;

完全更新と自動更新の動作については、「 ターゲット テーブルを完全に更新する」を参照してください。

統合 CDC パイプラインでは、垂直自動スケールが既定で有効になっています。 メモリ不足の状態が原因でパイプラインの更新が失敗した場合、次の更新では、より大きなドライバーが自動的にプロビジョニングされます。 この動作をオーバーライドするには、カスタム クラスター ポリシーを使用します。

制限事項

• ベータ版。 統合 CDC コネクタには、ワークスペース レベルの有効化が必要です。 Azure Databricks アカウント チームにお問い合わせください。
• デフォルトでトリガーされます。 デフォルトでは、統合されたCDCパイプラインはトリガーモードで動作します。Lakeflow Jobsタスクを使ってスケジュールを組みます。 Beta版では連続モードが利用可能です。 「 継続モードで統合されたCDCパイプラインを実行」を参照してください。
• API のみで作成。 パイプラインの作成は、REST API、Databricks CLI、ノートブック、および宣言型オートメーション バンドルを使用して使用できます。 UI の作成はまだサポートされていません。
• チャネルは PREVIEWする必要があります。 パイプライン スペックには、 "channel": "PREVIEW"を含める必要があります。
• 接続とコネクタの種類は変更できません。 connection_name および connector_type は、パイプラインの作成後に変更することはできません。 ソースを変更するには、新しいパイプラインを作成します。
• パイプラインあたり最大 300 個のテーブルを推奨します。
• プライマリ インスタンスのみ。 統合 CDC コネクタは、読み取りレプリカ、スタンバイ インスタンス、またはセカンダリ インスタンスをサポートしていません。
• 主キーのないテーブル。 パイプラインは、LOB 以外のすべての列を複合キーとして扱います。 SCD タイプ 2 を有効にしない限り、重複する行が 1 つの行に折りたたまれる可能性があります。
• 初期スナップショットは、複数の更新プログラムにまたがる場合があります。 大規模なテーブルの場合、初期スナップショットが 1 回の更新で完了しない可能性があります。 後続のスケジュールされた更新は、前回の更新が中断された場所で再開されます。
• 更新ランタイムは自動的に管理されます。 スマート クロージャは、各更新がいつ停止するかを決定します。 更新は、最大ランタイムによって制限されたソースに追い付いた後に完了します。 統合 CDC パイプラインのスマート クロージャを参照してください。 最小ランタイムまたは最大ランタイムを構成することはできません。 大規模な変更バックログは、複数の更新プログラムにまたがることがあります。 後続のスケジュールされた更新は、前回の更新が中断された場所で再開されます。
• ログの消去には、完全な更新が必要です。 パイプラインがそれらを処理する前に SQL Server が変更追跡ログまたは CDC ログをパージした場合は、影響を受けるテーブルでフル リフレッシュを実行します。 パイプラインはこの状態を検出し、イベント ログにエラーを表示します。

トラブルシューティング

ここで取り上げられない問題が発生した場合:

1. Azure Databricks UI または GET /api/2.0/pipelines/<pipeline-id>/events を使用して、パイプライン イベント ログを確認します。
2. カタログ エクスプローラーから Unity カタログ接続をテストして、ソースに到達可能であることを確認します。
3. ソース データベースとテーブルで変更の追跡または CDC が有効になっていることを確認します。
4. データベース ユーザーが、Microsoft SQL Server データベース ユーザー要件に記載されているSQL Serverアクセス許可を持っていることを確認します。
5. パイプライン スペックに "channel": "PREVIEW"が含まれていることを確認します。

その他のリソース

• Microsoft SQL Server インジェスト コネクタ
• Azure Databricks への取り込み用に Microsoft SQL Server を構成する
• SQL Server コネクタ リファレンス
• SQL Server インジェストのトラブルシューティング
• ターゲット テーブルを完全に更新する