Realtime Compute for Apache Flink:CREATE DATABASE AS (CDAS) 文

データ同期

スキーマ進化

CDAS がデータベースを同期する際、列の追加などのスキーマ変更が自動的に結果テーブルに反映されます。その動作およびポリシーは CTAS と同様です。詳細については、「スキーマ進化」をご参照ください。

構文に関する制限

• CDAS 文を含む SQL ドラフトのデバッグはサポートされていません。

• MiniBatch はサポートされていません。

  重要

  CTAS または CDAS 文を使用して SQL ドラフトを作成する前に、すべての MiniBatch 構成を削除してください。1. [O&M] > [構成] に移動します。2. [デプロイメントのデフォルト] タブを選択します。3. [その他の構成] セクションで、MiniBatch 構成が削除されていることを確認します。デプロイメントの作成時または開始時に「Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax」というエラーが発生した場合は、このエラーを修正するにはどうすればよいですか。をご参照ください。

サポートされるコネクタ

パラメーター

データベースの同期

tpcds MySQL データベースのすべてのテーブルを Hologres に同期します。

前提条件：

• ワークスペースに holo という名前の Hologres カタログが作成済みであること。

• ワークスペースに mysql という名前の MySQL カタログが作成済みであること。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds          -- Hologres に holo_tpcds を作成。
WITH ('sink.parallelism' = '4')                   -- 結果の並列度を設定（Hologres のデフォルト：4）。
AS DATABASE mysql.tpcds INCLUDING ALL TABLES      -- mysql.tpcds のすべてのテーブルを同期。
/*+ OPTIONS('server-id'='8001-8004') */;          -- MySQL CDC ソースオプションを構成。

WITH 句で設定されたオプションは現在のジョブにのみ適用され、書き込み動作を制御します。これらのオプションは Hologres カタログには永続化されません。サポートされるオプションについては、「Hologres」をご参照ください。

データベースシャードの統合と同期

複数の MySQL データベースシャードにまたがる同一の名前のテーブルをすべてマージし、単一の Hologres テーブルに統合します。

シナリオ： MySQL インスタンスに order_db01 ～ order_db99 という名前のシャードがあります。各シャードには order や order_detail などのテーブルが含まれています。すべてのシャードデータ（スキーマ変更を含む）を Hologres に集約する必要があります。

ソリューション： データベース名に対して正規表現を使用してすべてのシャードにマッチさせます。データベース名とテーブル名が、各結果テーブルに 2 つの追加フィールドとして追加されます。Hologres のプライマリキーには、データベース名、テーブル名、およびソーステーブルのプライマリキーカラムが含まれ、一意性が保証されます。ターゲットテーブルを事前に作成する必要はありません。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order              -- Hologres に holo_order を作成。
WITH('sink.parallelism'='4')                          -- 結果の並列度を設定（任意）。
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES  -- すべての order_db シャードにマッチ。
/*+ OPTIONS('server-id'='8001-8004') */;              -- MySQL CDC ソースオプションを構成（任意）。

シャード間で同一の名前を持つテーブルは、単一の Hologres テーブルにマージされます。

新規テーブルの同期

実行中の CDAS ジョブが開始された後、新規テーブル検出を有効にしてセーブポイントから再起動し、新たに追加されたテーブルをキャプチャします。

新規テーブル検出には VVR 8.0.6 以降が必要です。ソーステーブルの起動モードは initial である必要があります。

1. デプロイメント ページで、対象のデプロイメントを見つけ、操作 列の キャンセル をクリックします。

2. ダイアログで、詳細戦略 を展開し、セーブポイント付きで停止 を選択して、OK をクリックします。

3. SQL ドラフトに次の文を追加します。

   SET 'table.cdas.scan.newly-added-table.enabled' = 'true';

4. デプロイ をクリックします。

5. セーブポイントからジョブを復元します。

   1. デプロイメント ページで、デプロイメント名をクリックします。

   2. デプロイメント詳細ページで、ステート タブ、次に 履歴 サブタブをクリックします。

   3. セーブポイント リストで、ジョブを停止した際に作成されたセーブポイントを見つけます。

   4. 操作 列で、その他 > このセーブポイントからジョブを開始 を選択します。「ジョブの開始」をご参照ください。

複数の CDAS および CTAS 文を 1 つのジョブとして実行

単一のジョブで複数のデータベースを同期し、ソースの再利用を最大化します。

シナリオ： tpcds、tpch、および user_db01～user_db99 のシャードを 1 つのジョブで Hologres に同期します。

ソリューション： すべての文を STATEMENT SET ブロックで囲みます。Flink はすべての文で単一のソース頂点を再利用するため、サーバー ID 数、データベース接続数、および全体的な読み取り負荷（特に MySQL CDC ソースの場合）が削減されます。

USE CATALOG holo;

BEGIN STATEMENT SET;

-- すべての user_db シャードからユーザーテーブルを同期。
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- tpcds データベースを同期。
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- tpch データベースを同期。
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

複数の MySQL データベースを Kafka に同期

複数の MySQL データベースのテーブルを Kafka に同期し、トピック名の競合を回避します。

シナリオ： tpcds および tpch の両方に同じ名前のテーブルが存在します。これらを同じ Kafka トピックに送信すると、データが上書きされてしまいます。

ソリューション： cdas.topic.pattern オプションを使用して、データベースごとに一意のトピック名を生成します。{table-name} プレースホルダーは実行時に実際のテーブル名に置き換えられます。たとえば、'cdas.topic.pattern'='tpcds-{table-name}' は tpcds の table1 を tpcds-table1 トピックにルートします。

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- tpcds データベースを同期。トピック名は "tpcds-<table_name>" パターンに従う。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpcds-{table-name}')
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- tpch データベースを同期。トピック名は "tpch-<table_name>" パターンに従う。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpch-{table-name}')
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

MySQL と Flink の間に Kafka を中間レイヤーとして導入することで、MySQL の負荷も軽減されます。「Flink CDC による MySQL データベースの Kafka への同期」をご参照ください。