CDC ソースAzure SQL Databaseイベントストリームに追加する

この記事では、Azure SQL Database変更データ キャプチャ (CDC) ソースをイベントストリームに追加する方法について説明します。

Microsoft Fabric イベント ストリーム用の Azure SQL Database CDC ソース コネクタを使用すると、Azure SQL database内の現在のデータのスナップショットをキャプチャできます。 コネクタは、このデータに対する将来の行レベルの変更を監視し、記録します。 イベントストリームで変更がキャプチャされたら、この CDC データをリアルタイムで処理し、Fabric内のさまざまな宛先に送信して、さらに処理または分析することができます。

前提条件

• 共同作成者以上のアクセス許可を持つFabric容量ライセンス モードまたは試用版ライセンス モードのワークスペースへのアクセス。
• Azure SQL データベースを持つ実行中のAzure SQL サーバー。
• Azure SQL データベースにはパブリックにアクセスでき、ファイアウォールの内側に置いたり、仮想ネットワークでセキュリティで保護したりしないようにする必要があります。 保護されたネットワークに存在する場合は、 Eventstream コネクタの仮想ネットワークインジェクションを使用して接続します。
• ストアド プロシージャ sys.sp_cdc_enable_db を実行して、Azure SQL データベースで CDC を有効にしました。 詳細については、変更データ キャプチャを有効および無効にするを参照してください。

Azure SQL Databaseで CDC を有効にする

1. Azure ポータルに移動し、Azure SQL データベースを開き、 Query エディターを選択します。 サインインする認証方法を選択します。

   Azure SQL Databaseを開いているスクリーンショット。

2. 次の SQL コマンドを実行して、データベースで CDC を有効にします。

   -- Enable Database for CDC
   EXEC sys.sp_cdc_enable_db;
   
   -- Enable CDC for a table using a gating role option
   EXEC sys.sp_cdc_enable_table
       @source_schema = N'dbo',
       @source_name   = N'MyTable',
       @role_name     = NULL
   GO
   

• イベントストリームがない場合は、イベントストリームを作成します。

データ ソースの選択ウィザードを起動する

まだ eventstream にソースを追加していない場合は、[ データ ソースの接続 ] タイルを選択します。 リボンの [ ソースの追加>データ ソースの接続 ] を選択することもできます。

既に公開されているイベントストリームにソースを追加する場合は、 編集 モードに切り替えます。 リボンで、[ソースの追加] >[データ ソースの接続] を選択します。

[データ ソースの選択 ページで、Azure SQL DB (CDC) タイルで Connect を検索して選択します。

Azure SQL Database CDC の構成と接続

CDC を介したテーブル スキーマの自動登録を使用して、Azure SQL データベースから Eventstream に変更データを取り込みます。

1. Connect ページで、新しい接続を選択します。

   

2. Connection 設定 セクションで、Azure SQL データベースの次の値を入力します。

   • Server: Azure ポータルからAzure SQL サーバー名を入力します。 これは、mysqlservername.database.windows.net という形式です。

   • Database: Azure ポータルからAzure SQLデータベース名を入力します。

     

3. 下にスクロールし、[接続資格情報] セクションで、次の 手順に従います。

   • 接続名 には、接続の名前を入力します。

   • [認証の種類] で、[Basic] を選択します。

     注

     現在、Fabric Eventstream では、Basic 認証のみがサポートされています。

   • データベースで、[ユーザー名] と [パスワード] を入力します。

4. 接続 を選択します。

   

5. [ 接続 ] ページで、[ すべてのテーブル] または [ テーブル名の入力] を選択します。 後者を選択する場合は、完全なテーブル識別子 (schemaName.tableName) または有効な正規表現のコンマ区切りの一覧を使用してテーブルを指定します。 例えば次が挙げられます。

   • dbo.test.*を使用して、名前が dbo.test で始まるすべてのテーブルを選択します。
   • dbo\.(test1|test2)を使用して、dbo.test1とdbo.test2を選択します。

   両方の形式をコンマで混在させることができます。 エントリ全体の文字制限の合計は 102,400 文字です。

6. Advanced 設定 を展開して、Azure SQL Database CDC ソースのその他の構成オプションにアクセスできます。

   • 10 進処理モード: コネクタが DECIMAL と NUMERIC 列の値を処理する方法を定義します。
     • Precise: 正確な 10 進型 (Java BigDecimal など) を使用して値を表し、データ表現の完全な精度と精度を確保します。
     • Double: 値を倍精度浮動小数点数に変換します。 この設定により、使いやすさとパフォーマンスが向上しますが、精度が低下する可能性があります。
     • String: 値を書式設定された文字列としてエンコードします。 この設定により、ダウンストリーム システムで簡単に使用できますが、元の数値型に関するセマンティック情報は失われます。
   • スナップショット モード: コネクタの起動時にスナップショットを実行するための条件を指定します。
     • Initial: コネクタは、論理サーバー名のオフセットが記録されていない場合、または以前のスナップショットが完了しなかったことを検出した場合にのみ、スナップショットを実行します。 スナップショットが完了すると、コネクタは後続のデータベース変更のイベント レコードのストリーミングを開始します。
     • InitialOnly: コネクタは、論理サーバー名のオフセットが記録されていない場合にのみスナップショットを実行します。 スナップショットが完了すると、コネクタは停止します。 binlog から変更イベントを読み取るためにストリーミングに移行することはありません。
     • NoData: コネクタは、スキーマのみをキャプチャするスナップショットを実行しますが、テーブル データはキャプチャしません。 データの一貫性のあるスナップショットは必要ないが、コネクタの起動後に発生する変更のみが必要な場合は、このオプションを設定します。
   • 列除外リスト: 完全修飾名 (schemaName.tableName.columnName) を使用して変更イベント値から除外する列を指定します。
   • Database applicationIntent: SQL Server Always On 可用性グループでのルーティング動作を決定します。
     • ReadWrite: プライマリ レプリカに接続します。 接続で読み取り操作と書き込み操作の両方を実行する必要がある場合は、このオプションを使用します。
     • ReadOnly: 読み取り専用操作のために読み取り可能なセカンダリ レプリカへのルーティングを許可します。 これを使用して、レプリカで CDC を直接有効にします。 snapshot.isolation.mode をスナップショットに設定する必要があります。これは、読み取り専用レプリカでサポートされる唯一のトランザクション分離モードです。
   • スナップショット選択ステートメントのオーバーライド: スナップショットにテーブル内の行のサブセットのみを含める場合は、このプロパティを使用します。 このプロパティはスナップショットにのみ影響します。 コネクタがログから読み取るイベントには適用されません。

ストリームまたはソースの詳細

1. [ 接続 ] ページで、Eventstream または Real-Time ハブのどちらを使用しているかに基づいて、次のいずれかの手順に従います。

   • Eventstream:

     右側の [ ソースの詳細 ] ウィンドウで、次の手順に従います。

     1. [ソース名] で、[鉛筆] ボタンを選択して名前を変更します。

     2. Eventstream 名と Stream 名は読み取り専用であることに注意してください。

   • リアルタイムハブ:

     右側の [ストリームの詳細] セクションで、次の手順に従います。

     1. イベントストリームを作成するFabricワークスペースを選択します。

     2. "イベントストリーム名" の鉛筆ボタンを選択し、イベントストリームの名前を入力します。

     3. Stream 名の値は、eventstream の名前に -stream を追加することによって自動的に生成されます。 このストリームは、ウィザードの終了時にリアルタイム ハブの [すべてのデータ ストリーム ] ページに表示されます。

2. [構成] ページの下部にある [ 次へ ] を選択 します。

確認して接続する

[ 確認と接続 ] 画面で概要を確認し、[ 追加] (Eventstream) または [接続 ] (Real-Time ハブ) を選択します。

スキーマ処理ページ

1. [ スキーマ処理 ] ページで、次のいずれかのオプションを選択します。

   • 分析対応イベントと自動更新スキーマ (DeltaFlow プレビュー):コネクタは、ソース テーブル構造をミラーリングする分析対応ストリームに未加工の CDC イベントを変換します。 DeltaFlow は、変更の種類 (挿入、更新、削除) やタイムスタンプなどのメタデータを使用してイベントを強化し、変換先テーブルとスキーマの進化を自動的に管理します。
   • 未加工の CDC イベント: コネクタは、生の CDC イベントを取り込んで使用できるようにします。 必要に応じて、コネクタはテーブル スキーマを自動検出し、スキーマ レジストリに登録できます。 DeltaFlow 変換を使用せずにスキーマ認識を行う場合は、このオプションを使用します。

   注

   次のスクリーンショットは、CDC Azure SQL Database示しています。 スキーマ処理オプションは、サポートされているすべての CDC ソース コネクタで同じです。

   

2. イベント スキーマの関連付けを有効にします。

3. Workspace で、スキーマ セットのFabric ワークスペースを選択します。

4. [ スキーマ セット] では、[ + 作成 ] が既定で選択され、新しいスキーマ セットが作成されます。 既存のイベント スキーマ セットを選択するように変更できます。

5. 前の手順で [+ 作成 ] オプションを選択した場合は、スキーマ セットの名前を入力します。

6. [ 確認と接続 ] ページで、[ 追加] (Eventstream) または [接続 ] (Real-Time ハブ) を選択します。

   

   Azure SQL データベース内のすべてのテーブルまたは選択したテーブルに対して、コネクタは自動検出を行い、スキーマを作成し、スキーマ レジストリに登録します。

7. 中央の イベントストリーム ノードを選択し、下部ウィンドウの [ 関連付けられたスキーマ ] タブに切り替えます。

   

スキーマ セット

1. 前の手順で選択したワークスペースに移動します。 次の例では、 個人用ワークスペースです。

2. Azure SQL Database (CDC) コネクタが作成したスキーマ セットを選択します。

   

3. 次の図に示すように、スキーマ セットにスキーマが表示されます。

   

4. スキーマの JSON バージョンを表示するには、 JSON スキーマ ビューに切り替えます。

   

   このエディターを使用して、検出されたこれらのスキーマを変更しないでください。これは、Azure SQL データベース ソース内のテーブルのスキーマに対して不確定になるためです。

DeltaFlow: Analytics 対応イベント変換 (プレビュー)

Analytics 対応イベントと自動更新スキーマ (DeltaFlow) を有効にすると、コネクタは次の機能を提供します。

• 分析対応のイベント形状: 生の Debezium CDC イベントは、ソース テーブル構造を反映する表形式に変換されます。 イベントは、変更の種類 (insert、 update、または delete) やイベントタイムスタンプなどのメタデータ列でエンリッチされます。

• 自動宛先テーブル管理: DeltaFlow 対応ストリームをイベントハウスなどのサポート対象の宛先にルーティングすると、変換先テーブルがソース テーブル スキーマと一致するように自動的に作成されます。 コピー先テーブルを手動で作成または構成する必要はありません。

• スキーマの進化の処理: ソース データベース テーブルが変更されると (たとえば、新しい列が追加されたり、テーブルが作成されたりすると)、DeltaFlow は変更を自動的に検出し、登録されたスキーマを更新し、それに応じて変換先テーブルを調整します。 このオプションにより、スキーマの変更による手動による介入が最小限に抑えられます。

  

DeltaFlow が未加工の CDC イベントを分析対応の出力 (操作の種類やメタデータ列など) に変換する方法の詳細については、「 DeltaFlow 出力変換」を参照してください。

更新されたイベントストリームを表示する

1. eventstream に追加された Azure SQL Database (CDC) ソースは、Edit モードで確認できます。

   

2. この新しく追加されたAzure SQL Database CDC ソースを実装するには、Publish を選択します。 これらの手順を完了すると、Azure SQL Database CDC ソースを Live ビューで視覚化できるようになります。

   

スキーマを使用するように Eventstream の宛先を構成する

現在、関連付けられたスキーマを持つ Eventstream では、Eventhouse、カスタム エンドポイント、および派生ストリーム変換先のみがサポートされています。 このセクションでは、イベントストリームに対して拡張機能 (スキーマ サポートなど) が有効になっている場合に Eventhouse 変換先を追加および構成する方法について説明します。

カスタム エンドポイント変換先のスキーマを構成する

1. [ イベントの変換] または [宛先の追加] を選択し、[ CustomEndpoint] を選択します。

2. [ カスタム エンドポイント ] ウィンドウで、宛先の名前を指定します。

3. [入力スキーマ] で、イベントのスキーマを選択します。 このボックスでは、イベントストリームのスキーマサポートを有効にするときに選択します。

カスタム エンドポイントの送信先を構成する詳細な手順については、「 カスタム エンドポイントまたはカスタム アプリの宛先をイベントストリームに追加する」を参照してください。

イベントハウスの宛先に対するスキーマを設定する

1. [ イベントの変換] または [宛先の追加] を選択し、[ Eventhouse] を選択します。

2. [Eventhouse] ウィンドウで、次のスキーマ関連の設定を構成します。

   1. [入力スキーマ] で、ドロップダウン リストから 1 つ以上のスキーマを選択します。

      

      注

      Event Hubs ソースを構成するときに [ヘッダーによる動的スキーマ ] オプションを選択した場合は、ソースに対して複数のスキーマを構成し、それらをさまざまなプロパティとその値にマップしている可能性があります。

   2. [テーブルの作成方法] では、要件に応じて、すべてのスキーマを結合した 1 つのテーブルを選択するか、スキーマごとに個別のテーブルを選択します。

      

   3. [ データの書き込み方法] で、次のいずれかのオプションを選択します。

      • ペイロードのみ: 抽出されたペイロード データをテーブルに書き込みます。 複数の入力スキーマがある場合、データは複数のテーブルに送信されます。
      • メタデータとペイロード: メタデータとペイロード データを 1 つのテーブルに書き込みます。 列の例には、 source 、 subject、 type、および dataがあります。

      

イベントハウスの宛先を構成する詳細な手順については、「 イベントストリームにイベントハウスの宛先を追加する」を参照してください。

DeltaFlow 分析対応の出力を表示する (プレビュー)

Analytics 対応イベントと自動更新スキーマ (DeltaFlow) を有効にした場合、変換先テーブルはソース データベース テーブルを反映する図形に自動的に作成されます。 各テーブルには、元の列と、変更の種類とタイムスタンプのメタデータ列が含まれます。

未加工の Debezium CDC ペイロードを解析しなくても、Kusto クエリ言語 (KQL) またはその他の分析ツールを使用してこれらのテーブルに対してクエリを実行できます。

::: ゾーン終了

関連するコンテンツ

その他のコネクタは次のとおりです。

• Amazon Kinesis データ ストリーム
• Azure Cosmos DB
• Azure Event Hubs
• Azure Service Bus
• Azure IoT Hub
• Confluent Kafka
• カスタム エンドポイント
• Google Cloud Pub/Sub
• MySQL データベース CDC
• PostgreSQL データベースの CDC
• サンプル データ
• Azure Blob Storage イベント
• Fabric ワークスペース イベント