Realtime Compute for Apache Flink:CREATE TABLE AS (CTAS) ステートメント (廃止予定)

データ同期

スキーマ変更の同期

CTAS 文は、リアルタイムでデータを同期する際に、ソーステーブルからシンクテーブルにスキーマ変更もレプリケートします。スキーマ変更には、最初のテーブル作成とその後のテーブル変更が含まれます。

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

  スキーマ変更	説明
  NULL 可能列の追加	対応する列がシンクテーブルのスキーマの末尾に自動的に追加され、そのデータが同期されます。新しい列はデフォルトで NULL 可能になり、既存の行では、その値は NULL に設定されます。
  NOT NULL 列の追加	対応する列がシンクテーブルのスキーマの末尾に自動的に追加され、そのデータが同期されます。
  NULL 可能列の削除	列はシンクテーブルから削除されません。代わりに、そのデータは自動的に NULL に設定されます。
  列名の変更	これは、新しい列の追加として扱われます。名前が変更された列はシンクテーブルのスキーマの末尾に追加され、元の列のデータは自動的に NULL に設定されます。 説明 たとえば、col_a の名前が col_b に変更された場合、col_b はシンクテーブルの末尾に追加され、col_a のデータは NULL に設定されます。
  列のデータ型の変更	シンクシステムがデータ型の変更をサポートしている場合：現在、Paimon はデータ型変更の処理をサポートしています。CTAS は、INT から BIGINT への変更など、標準列の型の変更をサポートしています。 サポートされる型の変更は、シンクのルールによって異なります。詳細については、特定のシンクコネクタのドキュメントをご参照ください。 シンクシステムがデータ型の変更をサポートしていない場合：現在、Hologres のみが型正規化モードをサポートしており、データ型の変更を処理します。このモードでは、CTAS ジョブはより広いデータ型を持つダウンストリームテーブルを作成します。このアプローチは、データ型の変更に対するシンクの互換性を活用します。詳細については、「例：型正規化モードでのデータ同期」をご参照ください。 重要 CTAS ジョブの初回起動時に型正規化モードを有効にする必要があります。そうしなかった場合、このモードを有効にするには、シンクテーブルを削除してステートレス再起動を実行する必要があります。

  重要

  CTAS は、特定の DDL 操作を識別するのではなく、連続するレコードのスキーマを比較することによってスキーマ変更を検出します。

  • データ変更がないまま列が削除されてから再追加された場合、CTAS はスキーマ変更を検出しません。

  • CTAS は、新しい列が追加され、データ変更が発生した後にのみスキーマ変更を検出します。その後、スキーマ変更をシンクテーブルに同期します。

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

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

  • NOT NULL 列の削除。

  • 列の NOT NULL から NULL 可能への変更。

  重要

  サポートされないスキーマ変更が発生した場合は、手動でシンクテーブルを削除し、CTAS ジョブを再起動する必要があります。これにより、シンクテーブルが再作成され、履歴データが再同期されます。

単一テーブルの同期

シナリオ：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 ソーステーブルの追加パラメーターを指定します。

シャードテーブルとデータベースのマージ

シナリオ：CTAS を使用して、複数のシャード化された MySQL テーブルを単一の Hologres テーブルにマージします。

ソリューション：MySQL カタログと組み合わせて、正規表現を使用して同期するテーブルのデータベース名とテーブル名を照合します。データベース名とテーブル名は、シンクテーブルに追加の 2 列として書き込まれます。プライマリキーの一意性を保証するために、データベース名、テーブル名、および元のプライマリキーを組み合わせて、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') */;

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`,
  `class` AS 3  AFTER `id`
);

複数の CTAS ステートメントを単一ジョブとして実行

シナリオ：MySQL の web_sales テーブルとシャード化された user テーブルを、単一ジョブとして Hologres に同期します。

ソリューション：STATEMENT SET 構文を使用して、複数の CTAS ステートメントを単一ジョブとしてコミットします。このアプローチにより、単一のソースノードを複数の業務テーブルからのデータ読み取りに再利用でき、MySQL CDC ソースのサーバー 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;

1 つのソースから複数のシンクテーブルへの同期

• シンクテーブルに計算列が不要な場合

  USE CATALOG `holo`;
  
  BEGIN STATEMENT SET;
  
  -- CTAS ステートメントを使用して、MySQL の user テーブルを Hologres データウェアハウスの database1 内の user テーブルに同期します。
  CREATE TABLE IF NOT EXISTS `database1`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- CTAS ステートメントを使用して、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;
  
  -- CTAS ステートメントを使用して、MySQL の user テーブルを Hologres データウェアハウスの user_with_changed_id テーブルに同期します。テーブルには computed_id 列に計算された ID が含まれます。
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
  AS TABLE `user_with_changed_id`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- CTAS ステートメントを使用して、MySQL の user テーブルを Hologres データウェアハウスの user_with_changed_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 ステートメントを追加する必要があります。

ソリューション：SQL ジョブで新しいテーブルの検出機能を有効にし、新しい CTAS ステートメントを追加してから、セーブポイントからジョブを再起動します。新しいテーブルがキャプチャされると、そのデータが同期されます。

制限事項：

• 新しいテーブルの検出機能は、Realtime Compute for Apache Flink (VVR) 8.0.1 以降でサポートされています。

• CDC ソーステーブルから同期する場合、新規テーブル検出機能は、ソーステーブルの起動モードが 初期 に設定されているジョブでのみサポートされます。

• ソースを再利用するためには、追加された CTAS ステートメント内の新しいソーステーブルの設定が、既存のソーステーブルの設定と同一である必要があります。

• 新しい CTAS ステートメントを追加する前後で、起動モードなどのジョブ設定パラメーターは変更できません。

手順：

1. 新しい CTAS ステートメントを追加する必要がある場合は、[運用保守] ページに移動し、ジョブを停止して [Stop With Savepoint] を選択します。

2. SQL ジョブで、新しいテーブルの検出を有効にし、新しい CTAS ステートメントを追加してから、ジョブを再度 [Deploy] します。

   1. 次のステートメントを SQL ジョブに追加して、新しいテーブルの検出を有効にします。

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

   2. 新しい CTAS ステートメントを SQL ジョブに追加します。最終的なコード全体は次のとおりです。

      -- 新しいテーブルの検出を有効にします。
      SET 'table.ctas.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] をクリックします。

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

   1. [運用保守] ページで、対象のジョブの名前をクリックし、[State Management] タブに移動して [History] をクリックします。

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

   3. 対象のセーブポイントの [Actions] 列で、[More] > [Start Job from This Savepoint] を選択してジョブを開始します。詳細については、「ジョブの開始」をご参照ください。

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

シナリオ：CTAS ステートメントを使用して、MySQL ソーステーブルを Hologres パーティションテーブルに同期します。

Hologres のパーティションルール：Hologres では、シンクテーブルにプライマリキーが定義されている場合、パーティション列はプライマリキーに含まれている必要があります。

コード例：

MySQL ソーステーブルの DDL ステートメントは次のとおりです。

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 ステートメントでシンクテーブルのプライマリキーを再宣言します。

  パーティション列 (例: city) がソーステーブルのプライマリキーの一部ではない場合、ジョブは失敗します。パーティション列がプライマリキーの一部となるように、CTAS 文でシンクテーブルのプライマリキーを再宣言する必要があります。

  -- 次の SQL を使用して、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`;

型の正規化モードでのデータ同期

シナリオ: CTAS ステートメントを使用して Hologres テーブルにデータを同期する際に、既存のデータ型の精度の調整 (たとえば、VARCHAR(10) から VARCHAR(20) へ) や、データ型の変更 (たとえば、SMALLINT から INT へ) が必要なシナリオをサポートします。

ソリューション：Hologres の型の正規化モードを使用してデータを同期します。このモードは、CTAS ジョブの初回起動時に有効にする必要があります。起動時に有効にしなかった場合、有効にするには、ダウンストリームテーブルを削除した上で、ジョブをステートレスで再起動する必要があります。

型の正規化ルール：

アップストリームのデータ型が変更された後、新しい型と元の型が同じターゲット型に正規化される場合、ジョブは正常に実行されます。そうでない場合、変更は互換性がないと見なされ、CTAS ジョブは例外をスローします。具体的なルールは次のとおりです。

• TINYINT、SMALLINT、INT、および BIGINT は BIGINT に正規化されます。

• CHAR、VARCHAR、および STRING は STRING に正規化されます。

• FLOAT と DOUBLE は DOUBLE に正規化されます。

• その他のデータ型は、元の型マッピングルールに従って作成されます。詳細については、「型マッピング」をご参照ください。

コード例：

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 以降が必要です。

• SQL ヒントで、scan.incremental.snapshot.enabled と scan.full-changelog を true に設定します。

• MongoDB データベースで pre-image と post-image 機能を有効にする必要があります。有効にする方法については、「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;