MySQL用の統合CDCパイプラインを作成する
統合CDCパイプラインは、単一のパイプラインを使用してMySQLからDatabricksに変更データを取り込みます。標準のゲートウェイベースのアーキテクチャとは異なり、統合CDCパイプラインは抽出ステージと適用ステージの両方を1回のパイプライン更新で実行します。標準アーキテクチャには、個別の取り込みゲートウェイと取り込みパイプラインが必要です。
統合CDCコネクタを使用するタイミング
次の場合は統合CDCパイプラインを選択してください:
- 個別の取り込みゲートウェイと取り込みパイプラインの代わりに、1つのパイプラインによる、よりシンプルな設定をご希望です。
- Trigger（スケジュール済み）実行がニーズを満たします。By default, 統合されたCDCパイプラインはスケジュールに基づいて実行されます。継続的モードはベータ版で利用可能です。統合されたCDCパイプラインを継続的モードで実行するを参照してください。
- 既存のMySQLゲートウェイベースのフローでは利用できない自動フル更新のサポートが必要です。
以下の表では、2つのアーキテクチャを詳細に比較しています：
機能 | 標準CDC（ゲートウェイベース） | 統合 CDC |
|---|---|---|
パイプラインの数 | 2つ（インジェストゲートウェイとインジェストパイプライン） | 1つの（統合されたパイプライン） |
設定 | ゲートウェイを作成してから、ゲートウェイIDを参照する取り込みパイプラインを作成します。 | Unity Catalog接続を参照する単一のパイプラインを作成する |
ゲートウェイモード： | ゲートウェイは、独立した長時間実行プロセスとして継続的に実行されます | 抽出は、スケジュールされた各パイプラインの更新に埋め込まれています。 |
接続リファレンス |
|
|
コネクタタイプ | 暗黙的なデフォルトCDCの動作 | 明示的： |
ステージングボリューム | ゲートウェイによって内部的に管理されます。 | 宛先スキーマに自動作成される、または次の方法で構成される |
パイプラインモード | 連続 | Triggerパイプラインと連続パイプライン |
コンピュート | ゲートウェイにはクラシック、マネージド取り込みパイプラインにはサーバレス | クラシックまたはServerless |
自動フルリフレッシュ | 既存のMySQLゲートウェイベースのフローではサポートされていません | サポートされています |
ソースデータベースのセットアップについては、「Databricks への取り込み用に MySQL を構成する」を参照してください。同じソース構成が両方のアーキテクチャに適用されます。
統合された CDC パイプラインの実行方法
各パイプラインの更新は、2つのステージを順次実行します：
- 抽出 : パイプラインは、Unity Catalog 接続を使用してソースデータベースに接続します。初回ラン時、または完全更新後に、初期スナップショットをキャプチャします。後続のランでは、バイナリログ（binlog）を使用して増分変更（挿入、更新、削除）をキャプチャします。パイプラインは、抽出されたデータを Unity Catalog ステージングボリュームに書き込みます。
- アプリケーション ：パイプラインはステージングボリュームから読み取り、Unity Catalogのターゲットストリーミングテーブルに変更を適用します。Merge操作は、コネクタがソーステーブルから導出するキー（主キー、またはキーがないテーブルの場合は推論されたキー）と構成されたSCDタイプを使用します。パイプラインは、各変更を正確に1回適用するように設計されています。
各パイプラインの更新は、変更を抽出し、ソースに追いついた後、最大ランタイムによって制限され、自動的に停止します。詳細については、統合されたCDCパイプラインのスマートクローズを参照してください。定期的にデータを取り込むには、Lakeflow Jobsタスクを使用してパイプラインをスケジュールします。
要件
-
ワークスペースでUnity Catalogが有効になっています。
-
接続を作成する予定の場合：メタストアに対する
CREATE CONNECTION権限があります。Unity Catalog での権限の管理を参照してください。コネクタがUIベースのパイプラインオーサリングをサポートしている場合は、このページのステップを完了することで、接続とパイプラインを同時に作成できます。ただし、APIベースのパイプラインオーサリングを使用する場合は、このページのステップを完了する前に、カタログエクスプローラーで接続を作成する必要があります。管理された取り込みソースに接続するを参照してください。
-
既存の接続を使用する予定の場合: 接続に対して
USE CONNECTION権限またはALL PRIVILEGES権限を持っている必要があります。 -
ターゲットカタログには
USE CATALOG権限があります。 -
既存のスキーマに対する
USE SCHEMA、CREATE TABLE、およびCREATE VOLUME権限、またはターゲットカタログに対するCREATE SCHEMA権限を持っています。 -
MySQL ソースのセットアップが完了しました。Databricks への取り込み用に MySQL を構成するを参照してください。
-
次の権限が必要です。
CREATE CONNECTIONメタストア上 (新しいUnity Catalog接続を作成する場合)、または既存の接続でUSE CONNECTION。USE CATALOG送信先カタログ上。USE SCHEMAおよび宛先スキーマでのCREATE TABLE。CREATE VOLUME宛先スキーマ、またはdata_staging_optionsで指定されたスキーマについて。data_staging_optionsが設定されていなくても、パイプラインが宛先スキーマに自動作成するため、ステージングボリュームが必要です。
コンピュートの要件
統合CDCパイプラインは、クラシックまたはServerlessコンピュートで実行されます。
- クラシックコンピュート ：コンピュートプレーンは、DatabricksワークスペースのVPCまたはVNetで実行され、ネットワーク経由でMySQLインスタンスに到達する必要があります。サポートされているネットワークパスには、VPCまたはVNetピアリング、パブリックエンドポイント、およびオンプレミスMySQLの場合、AWS Direct Connect、Azure ExpressRoute、またはVPNが含まれます。
- Serverless コンピュート : Databricks Serverless コンピュートとソース データベースの間のServerless ネットワーク接続を設定します。オンプレミスのソースには、設定されたServerlessエグレス（例えば、トランジットゲートウェイ、またはExpressRouteまたはVPNとピアリングされたVNetなど）を経由するネットワークパスが必要です。
クラシックコンピュートの場合は、無制限のクラスター作成権限を使用するか、cluster_type が dlt に固定されたカスタム クラスターポリシー を使用してください。Databricks では、効率的な抽出のために少なくとも 8 コアを推奨しています。
MySQLへのUnity Catalog接続を作成します。
パイプラインを作成する前に、MySQLへのUnity Catalog接続を作成します。MySQL接続を作成するを参照してください。
統合された CDC パイプラインを作成します
API、Databricks CLI、ノートブック、または宣言型自動化バンドルを使用して、統合されたCDCパイプラインを作成します。
- Declarative Automation Bundles
- Databricks notebook
- Databricks CLI
- REST API
バンドルファイル（例：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}
catalog: ${var.dest_catalog}
schema: ${var.dest_schema}
ingestion_definition:
connection_name: ${var.connection_name}
connector_type: CDC
objects:
- table:
source_schema: 'my_database'
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
詳細情報については、宣言型オートメーションバンドルとは何ですか？をご覧ください。
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import (
ConnectorType,
IngestionConfig,
IngestionPipelineDefinition,
TableSpec,
)
w = WorkspaceClient()
pipeline = w.pipelines.create(
name="<pipeline-name>",
catalog="<destination-catalog>",
schema="<destination-schema>",
ingestion_definition=IngestionPipelineDefinition(
connection_name="<unity-catalog-connection-name>",
connector_type=ConnectorType.CDC,
objects=[
IngestionConfig(
table=TableSpec(
source_schema="<source-database>",
source_table="<source-table>",
destination_catalog="<destination-catalog>",
destination_schema="<destination-schema>",
)
)
],
),
)
print(f"Pipeline created: {pipeline.pipeline_id}")
databricks pipelines create --json '{
"name": "<pipeline-name>",
"catalog": "<destination-catalog>",
"schema": "<destination-schema>",
"ingestion_definition": {
"connection_name": "<unity-catalog-connection-name>",
"connector_type": "CDC",
"objects": [
{
"table": {
"source_schema": "<source-database>",
"source_table": "<source-table>"
}
}
]
}
}'
次の例は、MySQLデータベースから2つのテーブルをレプリケートします。両方ともトップレベルの宛先main.ingestionを継承します。この例ではserverlessが省略されており、default値はfalse（クラシックコンピュート）です。代わりにServerlessコンピュートで実行するには、"serverless": trueを追加します。
POST /api/2.0/pipelines
{
"name": "my-integrated-cdc-pipeline",
"catalog": "main",
"schema": "ingestion",
"ingestion_definition": {
"connection_name": "my-mysql-connection",
"connector_type": "CDC",
"objects": [
{
"table": {
"source_schema": "my_database",
"source_table": "customers",
"table_configuration": {
"scd_type": "SCD_TYPE_1"
}
}
},
{
"table": {
"source_schema": "my_database",
"source_table": "orders",
"table_configuration": {
"scd_type": "SCD_TYPE_1"
}
}
}
],
"data_staging_options": {
"catalog_name": "main",
"schema_name": "ingestion_staging"
}
}
}
ソースデータベース内のすべてのテーブルをレプリケートするには、個々のtableオブジェクトの代わりにschemaオブジェクトを使用します。
POST /api/2.0/pipelines
{
"name": "my-integrated-cdc-schema-pipeline",
"catalog": "main",
"schema": "ingestion",
"ingestion_definition": {
"connection_name": "my-mysql-connection",
"connector_type": "CDC",
"objects": [
{
"schema": {
"source_schema": "my_database",
"destination_catalog": "main",
"destination_schema": "ingestion"
}
}
]
}
}
パイプラインの更新を開始するには:
POST /api/2.0/pipelines/<pipeline-id>/updates
{
"full_refresh": false
}
定期的な更新をスケジュール
By default、統合CDCパイプラインはTrigger モードで実行されます。常時実行については、統合CDCパイプラインを継続的モードで実行するを参照してください。定期的なスケジュールでデータを取り込むには、パイプラインを実行する Lakeflow Jobs タスクを作成します。更新時間はソースが持つ変更データの量によって異なり、大規模なバックログは1回の更新で完了しない場合があります（統合CDCパイプラインのスマートクロージャを参照してください）。後続の更新が追いつけるよう、十分な頻度でパイプラインをスケジュールしてください。ほとんどのワークロードでは、60分を開始点とすると適切に機能します。前の更新が実行されている間にTriggerが起動した場合、新しい更新はキューに入れられます。
構成リファレンス
パイプラインパラメーター
パラメーター | Type | 説明 |
|---|---|---|
| string | パイプラインの名前。 |
| string | オプション。default は |
| Boolean | オプション。defaultは |
| string | デフォルトの宛先カタログ。テーブルごとの |
| string | デフォルトの宛先スキーマ。テーブルごとの |
| string | ソースデータベースへの Unity Catalog 接続。 |
| string |
|
| array | 取り込むテーブルまたはスキーマのリスト。 |
| オブジェクト | オプション。パイプラインがステージングボリュームを作成するカタログとスキーマ。パイプラインの宛先スキーマにデフォルト設定されます。 |
テーブルの仕様
パラメーター | 必須 | 説明 |
|---|---|---|
| はい | ソースのMySQLデータベース名。 |
| はい | ソーステーブル名。 |
| No | 宛先カタログです。デフォルトでは、パイプラインの |
| No | 送信先スキーマ。パイプラインの |
| No | 宛先テーブル名。 |
テーブル構成
パラメーター | デフォルト | 説明 |
|---|---|---|
|
|
|
| 自動検出されました。 | CDCイベントの順序付けに使用される列。指定されていない場合、ソースCDCメカニズムに基づいて自動検出されます。 |
| 無効 | サポートされていないDDL操作が検出された場合に、自動フル更新を構成します。自動フル更新ポリシーを参照してください。 |
MySQLのデータ型マッピングについては、MySQLコネクタの参照を参照してください。統合CDCパイプラインは自動型拡張をサポートしています。ソース列の型が拡張される（例：INTからBIGINT）と、宛先テーブルは自動的に適応します。
パイプラインを監視します
統合CDCパイプラインを作成して開始したら、次を使用してそのステータスを監視してください。
-
Databricks UI。 更新ステータス、テーブルごとの取り込みメトリクス、およびリネージを表示するには、 パイプライン セクションでパイプラインを開きます。
-
REST API。
TextGET /api/2.0/pipelines/<pipeline-id> -
イベントAPI。
TextGET /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 1 tables)
SELECT * FROM <destination_catalog>.<destination_schema>.<destination_table>
ORDER BY __START_AT DESC
LIMIT 10;
フル更新と自動フル更新の動作については、ターゲットテーブルを完全に更新するを参照してください。
統合CDCパイプラインは、デフォルトで垂直オートスケールを有効にします。メモリ不足の状態が原因でパイプラインの更新が失敗した場合、次回の更新で、より大きなドライバーが自動的にプロビジョニングされます。この動作をオーバーライドするには、カスタムクラスターポリシーを使用してください。
制限事項
- defaultでTrigger : defaultでは、統合された CDC パイプラインは Trigger モードで実行されます。Lakeflow Jobs タスクを使用してスケジュールを設定してください。連続モードはベータ版で利用できます。連続モードで統合 CDC パイプラインをランするを参照してください。
- 接続およびコネクタタイプは不変です ：パイプライン作成後に
connection_nameおよびconnector_typeを変更することはできません。ソースを変更するには、新しいパイプラインを作成してください。 - 主キーのないテーブル : テーブルに
NOT NULLのすべての列を持つ一意のキーがある場合、パイプラインはそれをキーとして使用します。それ以外の場合は、大きなオブジェクトを除くすべての列が複合キーとして扱われます。重複する行が単一の行に折りたたまれる可能性があります。 - 最初のスナップショットは1回の更新で完了します ：統合されたCDCコネクタは、大きなテーブルであっても、単一のパイプライン更新で最初のスナップショットを完了します。
- 更新のランタイムは自動的に管理されます ：スマートクロージャは、各更新を停止するタイミングを決定します。更新は、ソースに追いついた後に完了し、最大ランタイムによって制限されます。統合CDCパイプラインのスマートクロージャを参照してください。最小または最大ランタイムを構成することはできません。大きな変更バックログは、複数の更新にまたがる可能性があります。後続のスケジュールされた更新は、前回の更新が終了した場所から再開されます。
- バイナリログのパージには完全な更新が必要 ：パイプラインが変更を処理する前にMySQLバイナリログがパージされた場合、影響を受けるテーブルで完全な更新を実行してください。パイプラインはこの状態を検出し、イベントログにエラーを表示します。
トラブルシューティング
一部のエラーコードでは、INGESTION_GATEWAY_ のプレフィックスが使用されます。これは従来の命名規則であり、個別のインジェストゲートウェイが必要であることを示すものではありません。
エラー | 原因 | 解決方法 |
|---|---|---|
| パイプラインは直接発行モードではありません。 | 統合CDCパイプラインでは、直接公開Modeが自動的に設定されます。このエラーが表示された場合は、パイプラインを再作成してください。 |
| バイナリログが有効になっていないか、 |
|
| ソースにはサポートされていない |
|
| 指定されたソーステーブルが存在しないか、または削除されています。 | テーブルが存在し、接続ユーザーがアクセス権限を持っていることを確認してください。 |
| ソーススキーマは存在しません。 | ソースデータベースにスキーマが存在することを確認してください。 |
| ソースデータベースタイプはサポートされていません。 | 統合されたCDCコネクタは、MySQL、SQL Server、およびOracleをサポートしています。 |
| テーブル仕様に |
|
| ワークスペースの機能フラグは有効になっていません。 | 統合CDCコネクタをワークスペースで有効にするには、Databricksアカウントチームにお問い合わせください。 |
ここに記載されていない問題が発生した場合は、次の点にご注意ください：
- Databricks UI または
GET /api/2.0/pipelines/<pipeline-id>/eventsを通じてパイプラインイベントログを確認してください。 - Catalog Explorer から Unity Catalog 接続をテストして、ソースに到達可能であることを確認します。
- ソースデータベースでバイナリロギングが
binlog_format=ROWおよびbinlog_row_image=FULLで有効になっており、binlog_row_value_optionsが空であることを確認します。 - データベースユーザーにMySQLユーザー権限の付与に記載されているMySQL権限があることを確認してください。