Realtime Compute for Apache Flink:CREATE TABLE AS (CTAS)

データ同期

スキーマ進化

データ同期中、CTAS 文は、テーブルの作成やスキーマの変更など、ソーステーブルから結果テーブルへのスキーマ変更のレプリケーションをサポートします。

• サポートされているスキーマ変更

  スキーマ変更	説明
  NULL 値を許容する列の追加	関連する列を結果テーブルのスキーマの末尾に自動的に追加し、追加された列にデータを同期します。新しい列はデフォルトで NULL 値を許容する列として設定され、変更前のこの列のデータは自動的に NULL に設定されます。
  NULL 値を許容しない列の追加	対応する列を結果テーブルのスキーマの末尾に自動的に追加し、データを同期します。
  NULL 値を許容する列の削除	テーブルから列を削除する代わりに、結果テーブルの NULL 値を許容する列に NULL 値を自動的に入力します。
  列名の変更	列名の変更操作には、列の追加と列の削除が含まれます。ソーステーブルで列の名前が変更されると、新しい名前を使用する列が結果テーブルの末尾に追加され、元の名前を使用する列には NULL 値が入力されます。 説明 たとえば、ソーステーブルの col_a 列の名前が col_b に変更された場合、col_b 列が結果テーブルの末尾に追加され、col_a 列には自動的に NULL 値が入力されます。
  列のデータ型の変更	列の型変更をサポートする下流システムの場合：現在、型の変更をサポートしている下流システムは Paimon のみです。CTAS 文は、INT から BIGINT への変更など、通常の列の型変更をサポートしています。 互換性はシステム固有のルールに依存します (コネクタのドキュメントをご参照ください)。 列の型変更をサポートしない下流システムの場合：現在、Hologres のみが型拡張を使用して列の型変更を処理できます。その仕組みは次のとおりです。ジョブの起動時に、より広いデータ型を持つ Hologres テーブルが作成され、結果テーブルの互換性に基づいて列の型の変更がサポートされます。詳細については、「例：型正規化モードでのデータ同期」をご参照ください。 重要 Hologres で型拡張を有効にするには、最初のジョブ起動時に型正規化を有効にする必要があります。既存のジョブの場合は、Hologres テーブルを削除し、ステートなしで再起動して型正規化設定を適用してください。

  重要

  CTAS 文を同期に使用する場合、システムはスキーマの差分のみを比較し、特定の DDL タイプを識別しません。例：

  • ある列が削除され、その後すぐにデータ変更なしで再度追加された場合、システムはスキーマの変更を検出しません。

  • ある列が削除され、その間にデータ変更があった後に再度追加された場合、システムはスキーマの変更を検出して同期します。

• サポートされていないスキーマ変更

  • プライマリキーやインデックスなどの制約の変更。

  • NULL 値を許容しない列の削除。

  • NOT NULL から NULLABLE への変更。

  重要

  サポートされていないスキーマ変更を同期するには、手動で結果テーブルを削除し、ジョブを再起動して既存データを再同期してください。

テーブルの同期

説明：MySQL から Hologres へ web_sales テーブルを同期します。

前提条件：

• holo という名前の Hologres カタログが作成されていること。

• mysql という名前の MySQL カタログが作成されていること。

サンプルコード：

CTAS 文は、多くの場合、ソースカタログと宛先カタログと共に使用され、完全データ同期と増分データ同期をサポートします。ソースカタログは、明示的な DDL なしでソーステーブルのスキーマとプロパティを自動的に解析します。

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales   -- データをデフォルトデータベースの web_sales テーブルに同期します。
WITH ('jdbcWriteBatchSize' = '1024')   -- オプションで、結果テーブルのコネクタオプションを構成します。
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */; -- オプションで、MySQL CDC ソーステーブルの追加オプションを構成します。

テーブルおよびデータベースシャードの統合と同期

説明：シャード化された MySQL のテーブルとデータベースを統合してから、データを Hologres テーブルに同期します。

方法：MySQL カタログと正規表現を使用して、同期したいデータベースとテーブルを照合します。

データベース名とテーブル名は、2 つの追加フィールドの値として結果テーブルに書き込まれます。結果テーブルのプライマリキーは、データベース名、テーブル名、および元のプライマリキーで構成され、プライマリキーが一意であることを保証します。

コードと結果：

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`  
/*+ OPTIONS('server-id'='8001-8004') */;

ALTER TABLE `user02` ADD COLUMN `age` INT;
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30);

カスタム計算列の同期

説明：MySQL から Hologres への統合されたテーブルおよびデータベースシャードの同期中に、カスタム計算列を追加します。

コードと結果：

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */
ADD COLUMN (
  `c_id` AS `id` + 10 AFTER `id`,
  `calss` AS 3  AFTER `id`
);

単一ジョブでの複数の CTAS 文の実行

説明：単一のジョブで、MySQL から Hologres へ web_sales テーブルと user テーブルシャードを同期します。

方法：STATEMENT SET を使用して、複数の CTAS 文をグループとして実行します。このアプローチでは、ソース頂点を再利用して複数のテーブルからデータを読み取るため、サーバー ID の数、データベース接続、および全体的な読み取り負荷が削減されます。

サンプルコード：

USE CATALOG holo;

BEGIN STATEMENT SET;

-- web_sales テーブルからデータを同期します。
CREATE TABLE IF NOT EXISTS web_sales
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;

-- user テーブルシャードからデータを同期します。
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

複数の CTAS 文を使用したソースから複数のシンクへのデータ同期

• 計算列は結果テーブルに追加されません。

  USE CATALOG `holo`;
  
  BEGIN STATEMENT SET;
  
  -- user MySQL テーブルから Hologres の database1 の user テーブルにデータを同期します。
  CREATE TABLE IF NOT EXISTS `database1`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- MySQL データベースの user テーブルから Hologres の database2 の user テーブルにデータを同期します。
  CREATE TABLE IF NOT EXISTS `database2`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

• 計算列は結果テーブルに追加されます

  -- ソーステーブル user に基づいて user_with_changed_id という名前の一時テーブルを作成します。ソーステーブルの id 列に基づいて computed_id 列を定義します。
  CREATE TEMPORARY TABLE `user_with_changed_id` (
    `computed_id` AS `id` + 1000
  ) LIKE `mysql`.`tpcds`.`user`;
  
  -- ソーステーブル user に基づいて user_with_changed_age という名前の一時テーブルを作成します。ソーステーブルの age 列に基づいて computed_age 列を定義します。
  CREATE TEMPORARY TABLE `user_with_changed_age` (
    `computed_age` AS `age` + 1
  ) LIKE `mysql`.`tpcds`.`user`;
  
  BEGIN STATEMENT SET;
  
  -- MySQL データベースの user テーブルから Hologres の user_with_changed_id テーブルにデータを同期します。user_with_changed_id テーブルには、ソーステーブルの id 列に基づく計算から得られた ID が含まれます。得られた ID は computed_id 列にあります。
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
  AS TABLE `user_with_changed_id`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- MySQL データベースの user テーブルから Hologres の user_with_changed_age テーブルにデータを同期します。user_with_changed_age テーブルには、ソーステーブルの age 列に基づく計算から得られた年齢値が含まれます。得られた年齢値は computed_age 列にあります。
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
  AS TABLE `user_with_changed_age`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

複数の CTAS 文を使用した新しいテーブルの同期

シナリオの説明：同期に複数の CTAS 文を使用するジョブが開始された後、新しいテーブルをレプリケーションするために CTAS 文を追加します。

方法：ジョブの新規テーブル検出を有効にし、ジョブの SQL コードに CTAS 文を追加し、セーブポイントから再起動します。新しいテーブルがキャプチャされると、データがレプリケーションされます。

制限事項：

• 新規テーブル検出は VVR 8.0.1 以降でサポートされています。

• CDC ソーステーブルからデータを同期する場合、初期モードで開始されたジョブのみが新しいテーブルを検出できます。

• 新しい CTAS 文を使用して追加されるソーステーブルの構成は、元のソーステーブルの構成と同じでなければなりません。これにより、ソース頂点を再利用できます。

• CTAS 文を追加する前後のジョブ構成パラメーターは同じでなければなりません。たとえば、起動モードは同じである必要があります。

手順：

1. [Deployments] ページで、ターゲットのデプロイメントを見つけ、[Actions] 列の [Cancel] をクリックします。

2. ダイアログボックスで、[More Strategies] セクションを展開し、[Stop With Savepoint] を選択して、[OK] をクリックします。

3. ジョブの SQL ドラフトで、新規テーブル検出を有効にし、CTAS 文を追加します。

   1. 次の文を追加して、新規テーブル検出を有効にします。

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

   2. CTAS 文を追加します。ジョブの完全なコードは次のようになります。

      -- 新規テーブル検出を有効にします
      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';
      
      USE CATALOG holo;
      
      BEGIN STATEMENT SET;
      
      -- web_sales テーブルからデータを同期します。
      CREATE TABLE IF NOT EXISTS web_sales
      AS TABLE mysql.tpcds.web_sales
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      -- user テーブルシャードからデータを同期します。
      CREATE TABLE IF NOT EXISTS user
      AS TABLE mysql.`wp.*`.`user[0-9]+`
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      -- product テーブルからデータを同期します。
      CREATE TABLE IF NOT EXISTS product
      AS TABLE mysql.tpcds.product
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      END;

   3. [Deploy] をクリックします。

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

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

   2. デプロイメントの詳細ページで、[State] タブをクリックします。次に、[History] サブタブをクリックします。

   3. [Savepoints] リストで、ジョブがキャンセルされたときに作成されたセーブポイントを見つけます。

   4. [Actions] 列で [More] > [Start job from this savepoint] を選択します。詳細については、「ジョブデプロイメントの開始」をご参照ください。

Hologres のパーティションテーブルへの同期

シナリオの説明：MySQL から Hologres のパーティションテーブルにデータをレプリケーションします。

使用上の注意：Hologres テーブルにプライマリキーが定義されている場合、パーティション列をプライマリキーに含める必要があります。

サンプルコード：

MySQL テーブルの作成：

CREATE TABLE orders (
    order_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    city VARCHAR(100) NOT NULL
    order_date DATE,
    purchaser INTEGER,
    PRIMARY KEY(order_id, product_id)
);

パーティション列がプライマリキーの一部であるかどうかに応じて、適切な方法を選択します。

• ソースのプライマリキーにパーティション列が含まれている場合：

  • CTAS 文を直接使用します。

    Hologres は、パーティション列がプライマリキーに含まれているかどうかを自動的に検証します。

    CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`
    PARTITIONED BY (product_id)
    AS TABLE `mysql`.`tpcds`.`orders`;

• ソースのプライマリキーにパーティション列が含まれていない場合：

  • CTAS 文で結果テーブルのプライマリキーを宣言し、プライマリキーの定義にパーティション列を含めます。

    この場合、プライマリキーを再定義しないか、パーティション列を含めないと、ジョブは失敗します。

    -- Hologres パーティションテーブルのプライマリキーとして order_id、product_id、city フィールドを宣言します。
    CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`(
        CONSTRAINT `PK_order_id_city` PRIMARY KEY (`order_id`,`product_id`,`city`) NOT ENFORCED
    )
    PARTITIONED BY (city)
    AS TABLE `mysql`.`tpcds`.`orders`;

レプリケーション中のデータ型の拡張

シナリオの説明：データ同期中に、列の精度を変更する (例：VARCHAR(10) から VARCHAR(20))、または列のデータ型を変更する (例：SMALLINT から INT)。

方法：

• 新規ジョブ：最初の起動時に型正規化モードを有効にします。

• 既存のジョブ：Hologres の結果テーブルを削除し、ステートなしで再起動して型正規化を適用します。

型正規化ルール：

新しいデータ型と元のデータ型が同じデータ型に正規化される場合、データ型は正常に変更され、ジョブは正常に実行されます。それ以外の場合は、例外が報告されます。詳細は次のとおりです。

• TINYINT、SMALLINT、INT、および BIGINT は BIGINT に変換されます。

• CHAR、VARCHAR、および STRING は STRING に変換されます。

• FLOAT と DOUBLE は DOUBLE に変換されます。

• その他のデータ型は、Hologres と Flink フィールド間のデータ型のマッピングに基づいて変換されます。詳細については、「データ型のマッピング」をご参照ください。

サンプルコード：

CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders` 
WITH (
'connector' = 'hologres', 
'enableTypeNormalization' = 'true' -- 型正規化モードを有効にします。
) AS TABLE `mysql`.`tpcds`.`orders`;

MongoDB から Hologres へのデータ同期

制限事項：

• VVR 8.0.6 以降および MongoDB バージョン 6.0 以降でのみサポートされています。

• ソーステーブルのコネクタオプションでは、scan.incremental.snapshot.enabled と scan.full-changelog を true に設定する必要があります。

• MongoDB データベースでプリイメージとポストイメージ機能を有効にする必要があります。詳細については、「Document Preimages」をご参照ください。

• 単一のジョブで複数の MongoDB コレクションからデータを同期するには、すべてのテーブルで次のコネクタオプションの構成が同一であることを確認してください。

  • MongoDB データベース関連のオプション (hosts、scheme、username、password、connectionOptions など)

  • scan.startup.mode

サンプルコード：

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table1`
AS TABLE `mongodb`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table2`
AS TABLE `mongodb`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

END;