Realtime Compute for Apache Flink:Paimon カタログの管理

• CREATE TABLE ステートメントを使用したテーブルの作成

  [開発] > [スクリプト] の SQL エディターで、以下のコマンドを実行します。

  次の例では、my-catalog の my_db データベースにパーティションテーブルを作成します。

  -- my-catalog をご利用の Paimon カタログの名前に置き換えます。
  -- my_db を使用したいデータベースの名前に置き換えます。
  -- my_tbl を英語のカスタム名に置き換えることもできます。
  CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
    dt STRING,
    shop_id BIGINT,
    user_id BIGINT,
    num_orders INT,
    total_amount INT,
    PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
  ) PARTITIONED BY (dt) WITH (
    'bucket' = '4'
  );

  詳細については、「Apache Paimon コネクタ」および「プライマリキーテーブルと追加専用テーブル」をご参照ください。

• CREATE TABLE AS (CTAS) または CREATE DATABASE AS (CDAS) ステートメントを使用したテーブルの作成

  CTAS および CDAS ステートメントは、データとテーブルスキーマの変更を自動的に同期します。これらのステートメントを使用して、MySQL や Kafka などのソースから Paimon にテーブルを簡単にレプリケーションします。

  説明

  • 'bucket' = '-1' を設定して動的バケットを有効にすることは、プライマリキーテーブルまたは追加専用テーブル (プライマリキーなし) ではサポートされていません。

  • CTAS および CDAS ステートメントでは、ダウンストリームにレプリケーションできるプロパティを設定できます。詳細については、「Paimon コネクタ」をご参照ください。

  • 特定のテーブルのレプリケーション

    次の例では、mysql.tpcds.web_sales テーブルのスキーマに基づいて Paimon テーブル `my-catalog`.`my_db`.`web_sales` を自動的に作成し、データをレプリケーションします。

    CREATE TABLE IF NOT EXISTS `<catalog name>`.`<db name>`.`<table name>`
    WITH (
      'bucket' = '4',
      'changelog-producer' = 'input'
    ) AS TABLE mysql.tpcds.web_sales;

  • データベース全体のレプリケーション

    次の例では、mysql.tpcds データベースの各テーブルのスキーマに基づいて、`my-catalog`.`my_db` データベースに Paimon テーブルを自動的に作成し、データをレプリケーションします。

    CREATE DATABASE IF NOT EXISTS `<catalog name>`.`<db name>`
    WITH (
      'changelog-producer' = 'input'
    ) AS DATABASE mysql.tpcds INCLUDING ALL TABLES;

  • 列の型の変更のレプリケーション

    CTAS または CDAS ステートメントを使用して作成された Paimon テーブルは、特定の列の型の変更をサポートします。

    • デフォルトの型レプリケーション

      デフォルトでは、Paimon テーブルはソースの正確な列の型を保持します。ただし、明示的な設定なしで特定の型昇格がサポートされています。

      • 整数型：TINYINT、SMALLINT、INT、および BIGINT は、同等以上の精度の整数型に昇格できます (TINYINT < SMALLINT < INT < BIGINT)。

      • 浮動小数点型：FLOAT および DOUBLE は、同等以上の精度の浮動小数点型に昇格できます (FLOAT < DOUBLE)。

      • 文字列型：CHAR、VARCHAR、および STRING は、同等以上の精度の文字列型に昇格できます。

    • 型の正規化

      より柔軟な型の変更を有効にし、互換性のあるスキーマ進化によるジョブの失敗を防ぐには、CTAS/CDAS ステートメントの WITH 句で 'enableTypeNormalization' = 'true' を設定します。このモードでは、Flink はスキーマ進化中にデータ型を正規化します。ジョブは、古い型と新しい型が同じ基になる型に正規化できない場合にのみ失敗します。ルール：

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

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

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

      • 他のデータ型は正規化されません。

    • 正規化の影響：

      型の正規化がアクティブな場合、Paimon テーブルに保存されるデータ型は正規化された型になります。たとえば、ソース列 c1 (SMALLINT) と c2 (INT) が両方とも BIGINT に正規化される場合、c1 と c2 の両方が Paimon テーブルに BIGINT として保存されます。

    • 例：

      • 成功：ソースの SMALLINT が INT に変更された場合、両方の型は BIGINT に正規化されます。これは互換性のある変更であり、ジョブは正常に続行されます。

      • 失敗：ソースの FLOAT が BIGINT に変更された場合、それらの正規化された型 (DOUBLE と BIGINT) は互換性がないため、例外が発生します。

[開発] > [スクリプト] の SQL エディターで、次のコマンドを実行します。

ALTER TABLE my_table SET (
  'write-buffer-size' = '256 MB',
  'write-buffer-spillable' = 'true'
);

ALTER TABLE my_table RENAME TO my_table_new;

ALTER TABLE my_table RENAME c0 TO c1;

ALTER TABLE my_table DROP (c1, c2);

ALTER TABLE my_table DROP PARTITION (`dt` = '20240108', `hh` = '08'), PARTITION (`dt` = '20240109', `hh` = '07');

ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';

ALTER TABLE my_table MODIFY col_a DOUBLE;

[開発] > [スクリプト] の SQL エディターで次のコマンドを実行します。

-- my-catalog をご利用の Paimon カタログの名前に置き換えます。
-- my_db を使用したいデータベースの名前に置き換えます。
-- my_tbl を作成した Paimon カタログテーブルの名前に置き換えます。
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;

The following statement has been executed successfully! というメッセージが返された場合、Paimon テーブルは削除されています。