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

データ同期

スキーマ進化

データベースデータの同期中に、スキーマの変更（列の追加など）をシンクに伝達することもできます。ポリシーは、CTAS 文のポリシーと一致しています。詳細については、「スキーマ進化」をご参照ください。

構文の制限

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

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

  重要

  • CTAS 文または CDAS 文を含む SQL ドラフトを作成する前に、MiniBatch 構成が削除されていることを確認してください。次の手順を実行します。

    1. [O&M] > [構成] に移動します。

    2. [デプロイメントのデフォルト] タブを選択します。

    3. [その他の構成] セクションで、MiniBatch 構成が削除されていることを確認します。

  • SQL ドラフトからデプロイメントを作成したり、デプロイメントを開始したりするときにエラーが報告された場合は、「「現在、CTAS/CDAS 構文で StreamExecMiniBatchAssigner タイプの ExecNode のマージはサポートされていません」エラーを修正する方法」をご参照ください。

サポートされているアップストリームシステムとダウンストリームシステム

次の表に、CDAS 文でサポートされているアップストリームシステムとダウンストリームシステムを示します。

データベースの同期

説明: 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  -- すべてのテーブルを同期します。
/*+ OPTIONS('server-id'='8001-8004') */ ; -- 必要に応じて、MySQL CDC ソーステーブルのオプションを構成します。

データベースシャード間でのデータの同期

説明: MySQL インスタンスには、order_db01 から order_db99 という名前の複数のデータベースシャードがあります。各データベースシャードには、order や order_detail などの複数のテーブルが含まれています。 CDAS 文を使用すると、これらのデータベースシャード内のすべてのテーブルを、データとスキーマの変更と共に Hologres に同期できます。

ソリューション:

データベース名（`order_db[0-9]+`）に正規表現を使用して、同期するすべてのデータベースシャード（order_db01 から order_db99）を照合します。データベース名とテーブル名は、2 つの追加フィールドとして各シンクテーブルに追加されます。

Hologres テーブルのプライマリキーには、一意性を確保するために、データベース名、テーブル名、およびソーステーブルのプライマリキー列が含まれています。

ターゲットテーブルを事前に作成する必要はありません。

サンプルコードと結果:

データベースシャード間で同じ名前のテーブルは、単一の Hologres テーブルに同期される前にマージされます。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order  -- Hologres に holo_order という名前のデータベースを作成します。データベースには、MySQL インスタンスの order_db データベースシャード内のすべてのテーブルが含まれています。
WITH('sink.parallelism'='4')        -- 宛先データベースのパラメーターを指定します。デフォルトでは、各 Hologres シンクの並列度は 4 です。この設定はオプションです。
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES -- MySQL インスタンスの order_db データベースシャード内のすべてのテーブルからデータを同期します。
/*+OPTIONS('server-id'='8001-8004')*/;  -- 必要に応じて、MySQL CDC ソーステーブルの追加パラメーターを指定します。この設定はオプションです。

新しいテーブルの同期

説明: 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 文の実行

説明: tpcds データベース、tpch データベース、および user_db01 から user_db99 までのデータベースシャードのデータを、単一のジョブで Hologres に同期します。

ソリューション: STATEMENT SET 文を使用して、複数の CDAS 文と CTAS 文をグループ化します。このソリューションでは、ソース頂点を再利用して、必要なテーブルからデータを読み取ります。これは MySQL CDC データソースで特に有益であり、サーバー ID、データベース接続の数、および MySQL データベースへの全体的な読み取り負荷を削減します。

サンプルコード:

USE CATALOG holo;

BEGIN STATEMENT SET;

-- データベースシャード全体のユーザテーブルからデータを同期します。
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;

複数の CDAS 文を介してデータベース内のデータを Kafka に同期する

説明: 複数の MySQL データベース（tpcds や tpch など）のテーブル内のデータを Kafka に同期します。

ソリューション: 複数の CDAS 文を使用してデータベース内のデータを Kafka に同期する場合、同じ名前のテーブルが異なるデータベースに存在する可能性があります。トピックの競合を防ぐために、cdas.topic.pattern オプションを構成して、トピック名のパターンを定義します。 {table-name} プレースホルダーを使用できます。たとえば、'cdas.topic.pattern'='dbname-{table-name}' を指定すると、db1 データベースの table1 テーブルからレプリケートされたデータが dbname-table1 Kafka トピックに送信されます。

サンプルコード:

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- tpcds データベースからデータを同期します。
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 データベースからデータを同期します。
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 への負荷を軽減できます。詳細については、「MySQL データベースのすべてのテーブルから Kafka にデータを同期する」をご参照ください。

FAQ

ランタイムエラー

• デプロイメントを開始できない場合はどうすればよいですか？

• デプロイメントの実行後にデプロイメントが再起動される場合はどうすればよいですか？

• エラーメッセージ「akka.pattern.AskTimeoutException」が表示される場合はどうすればよいですか？

ジョブパフォーマンス

• バックプレッシャーの問題をトラブルシューティングするにはどうすればよいですか？

• テーブルから完全データを読み取るときに、データ読み取り効率が低く、バックプレッシャーが存在する場合はどうすればよいですか？

• 高レイテンシの問題をトラブルシューティングするにはどうすればよいですか？

• アップストリームデータの消費が不安定な場合はどうすればよいですか？

データ同期

• Realtime Compute for Apache Flink の開発コンソールで MySQL テーブルを使用してカタログを作成すると、MySQL テーブルのプライマリキーのデータ型が BIGINT UNSIGNED から DECIMAL に変わるのはなぜですか？ CREATE TABLE AS 文を実行して Hologres にデータを同期した後、プライマリキーのデータ型が TEXT に変わるのはなぜですか？

• 「現在、CTAS/CDAS 構文で StreamExecMiniBatchAssigner タイプの ExecNode のマージはサポートされていません」エラーを修正する方法

• ジョブから出力が得られないのはなぜですか？

• フルマネージド Flink がソースデータを読み取れないという問題をトラブルシューティングするにはどうすればよいですか？

• フルマネージド Flink がシンクテーブルにデータを書き込めないという問題をトラブルシューティングするにはどうすればよいですか？

• データの不正確さの問題をトラブルシューティングするにはどうすればよいですか？