マテリアライズドテーブルの作成と使用 - Realtime Compute for Apache Flink

このトピックでは、マテリアライズドテーブルの作成、データバックフィル、データの鮮度の変更、データリネージの表示方法について説明します。

制限事項

この機能は、Ververica Runtime (VVR) 8.0.10 以降のバージョンでのみ利用可能です。
マテリアライズドテーブルは、メタデータストレージとしてファイルシステムまたは DLF を使用する Apache Paimon カタログでのみ作成できます。カスタムの Apache Paimon カタログはサポートされていません。
ジョブを開発およびデプロイする権限が必要です。詳細については、「開発コンソールの権限付与」をご参照ください。
一時テーブル、一時ユーザー定義関数、一時ビューなど、一時オブジェクトはサポートされていません。

マテリアライズドテーブルの作成

構文

CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
-- Primary key constraint
[([CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED)]

[COMMENT table_comment]
-- Partition key
[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
-- With options
[WITH (key1=val1, key2=val2, ...)]
-- Data freshness
FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
-- Refresh mode
[REFRESH_MODE = { CONTINUOUS | FULL }]
AS  <select_statement>

パラメーター

パラメーター	必須	説明
FRESHNESS	はい	マテリアライズドテーブルのデータの鮮度。ソーステーブルからのデータ更新の最大許容レイテンシを定義します。説明アップストリームテーブルもマテリアライズドテーブルである場合、ダウンストリームテーブルのデータの鮮度は、アップストリームテーブルのデータの鮮度の正の整数倍である必要があります。データの鮮度の最大値は 1 日です。
AS <select_statement>	はい	マテリアライズドテーブルにデータを投入するクエリを定義します。アップストリームテーブルには、マテリアライズドテーブル、通常テーブル、またはビューを指定できます。`SELECT` 文は、すべての Flink SQL クエリをサポートします。
PRIMARY KEY	いいえ	テーブル内の各行を一意に識別するオプションの列。これらの列に NULL 値を含めることはできません。
PARTITIONED BY	いいえ	マテリアライズドテーブルをパーティション分割するために使用されるオプションの列。
WITH Options	いいえ	テーブルプロパティとパーティション列の時刻フォーマットパラメーターを定義します。たとえば、`WITH ('partition.fields.#.date-formatter' = 'yyyyMMdd')` を使用して、パーティション列の時刻フォーマットパラメーターを設定できます。パラメーターの使用方法の詳細については、手順の例をご参照ください。
REFRESH_MODE	いいえ	マテリアライズドテーブルのリフレッシュモードを指定します。指定されたリフレッシュモードは、データの鮮度に基づいてフレームワークが自動的に推論するモードよりも優先されます。これにより、特定のシナリオに対応できます。 CONTINUOUS：ストリーミングジョブがマテリアライズドテーブルを増分更新します。ダウンストリームのデータは、即時またはチェックポイント完了後に可視化されます。 FULL：ワークフローが定期的にマテリアライズドテーブルのバッチ更新をトリガーします。このモードでは、エンジンがフル更新と増分更新のどちらを実行するかを自動的に決定します。詳細については、「マテリアライズドテーブルの増分更新」をご参照ください。データリフレッシュサイクルは、データの鮮度の設定と一致します。デフォルトでは、データはテーブルレベルで上書きされます。パーティション列が存在する場合、最新のパーティションのみをリフレッシュするか、すべてのパーティションを更新するかを選択できます。

操作手順

Realtime Compute for Apache Flink コンソールにログインします。
対象のワークスペースの操作列で、コンソール をクリックします。
左側のナビゲーションウィンドウで カタログ をクリックし、対象の Apache Paimon カタログをクリックします。
対象のデータベースをクリックし、マテリアライズドテーブルの作成 をクリックします。
プライマリキーが order_id、カテゴリ名が order_name、日付フィールドが ds の orders という名前のソーステーブルがあるとします。以下の例は、このテーブルに基づいてマテリアライズドテーブルを作成する方法を示しています。
- orders テーブルに基づいてマテリアライズドテーブル mt_order を作成します。クエリはすべての列を選択し、データの鮮度は 5 秒に設定されます。
```
CREATE MATERIALIZED TABLE mt_order
FRESHNESS = INTERVAL '5' SECOND
AS
SELECT * FROM `paimon`.`db`.`orders`
;
```
- マテリアライズドテーブル mt_order に基づいてマテリアライズドテーブル mt_id を作成します。クエリは order_id と ds をテーブル列として選択し、order_id をプライマリキー、ds をパーティション列として設定し、データの鮮度を 30 分に設定します。
```
CREATE MATERIALIZED TABLE mt_id (
 PRIMARY KEY (order_id) NOT ENFORCED
)
PARTITIONED BY(ds)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT order_id,ds FROM mt_order
;
```
- マテリアライズドテーブル mt_order に基づいてマテリアライズドテーブル mt_ds を作成し、ds パーティション列に date-formatter (時刻フォーマット) を指定します。スケジュールが実行されるたびに、スケジュール時刻から新鮮さを引いた値が、対応する ds パーティション値に変換されます。たとえば、データの新鮮さが 1 時間に設定され、スケジュール時刻が 2024-01-01 00:00:00 の場合、ds の計算値は 20231231 となり、パーティション ds = '20231231' のデータのみがリフレッシュされます。スケジュール時刻が 2024-01-01 01:00:00 の場合、ds の計算値は 20240101 となり、パーティション ds = '20240101' のデータがリフレッシュされます。
```
CREATE MATERIALIZED TABLE mt_ds
PARTITIONED BY(ds)
WITH (
    'partition.fields.ds.date-formatter' = 'yyyyMMdd'
)
FRESHNESS = INTERVAL '1' HOUR
AS
SELECT order_id,order_name,ds FROM mt_order
;
```
  説明
  - partition.fields.#.date-formatter では、# プレースホルダーは STRING 型の有効なパーティション列でなければなりません。
  - partition.fields.#.date-formatter オプションは、マテリアライズドテーブルの時間パーティションフォーマットを指定します。# プレースホルダーは、文字列型のパーティション列の名前を表します。この情報により、システムはスケジュールされた更新中にどのパーティションをリフレッシュするかを識別できます。
マテリアライズドテーブルの更新を開始または停止します。
1. カタログの下にある対象のマテリアライズドテーブルをクリックします。
2. 右上隅で、更新の開始 または 更新の停止 をクリックします。
  説明
  進行中の更新を停止した場合、ジョブは現在の更新サイクルを完了してから停止します。
マテリアライズドテーブルジョブの詳細を表示します。
テーブルスキーマ タブの 基本情報 セクションで、最新ジョブ または ワークフロー の横にあるジョブ ID をクリックして詳細を表示します。

マテリアライズドテーブルクエリの変更

制限事項

クエリを変更できるのは、VVR 11.1 以降で作成されたマテリアライズドテーブルのみです。

クエリを変更する際、列の追加と計算ロジックの変更のみが可能です。既存の列の順序を変更したり、その定義を変更したりすることはできません。

操作	サポート状況	説明
新しい列の追加	はい	既存の列の順序を維持したまま、スキーマに新しい列を追加できます。
既存の列の計算ロジックの変更 (名前や型は変更しない)	はい	計算ロジックは変更できますが、列名とデータ型は同じでなければなりません。
既存の列の順序の変更	いいえ	列の順序は固定です。変更するには、マテリアライズドテーブルを削除して再作成する必要があります。
既存の列の名前またはデータ型の変更	いいえ	マテリアライズドテーブルを削除して再作成する必要があります。

例

テーブルの編集 をクリックしてクエリを変更します。以下のコードは例です。

ALTER MATERIALIZED TABLE `paimon`.`default`.`mt-orders`
    AS
    SELECT
      *,
      price * quantity AS total_price
    FROM orders
    WHERE price * quantity > 1000
;

プレビュー をクリックして、変更前後の比較を確認します。
OK をクリックします。テーブルスキーマ タブで、新しく追加された列と変更されたクエリロジックを確認できます。

重要

列の追加は、通常、ダウンストリームコンシューマーに影響しません。しかし、ダウンストリームジョブがアップストリームのマテリアライズドテーブルからデータを消費する際に動的解析 (SELECT * や自動フィールドマッピングなど) に依存している場合、ジョブが失敗したり、データ形式の不一致エラーが報告されたりする可能性があります。動的解析を避け、固定の列リストを使用し、アップストリームのスキーマが変更されるたびにダウンストリームのテーブルスキーマを速やかに更新することをお勧めします。

増分更新

制限事項

この機能は、VVR 8.0.11 以降のバージョンでのみ利用可能です。

更新モード

マテリアライズドテーブルは、ストリーミング、フルバッチ、増分バッチの 3 つの更新モードをサポートしています。

モードはデータの鮮度の設定によって決まります。鮮度が 30 分未満の場合はストリーミングモードが有効になり、30 分以上の場合はバッチモードが有効になります。バッチモードでは、エンジンがフル更新と増分更新のどちらを実行するかを自動的に決定します。増分更新は、最後の更新以降に変更されたデータのみを計算し、それをマテリアライズドテーブルにマージします。フル更新は、テーブル全体またはパーティション全体のデータを計算し、マテリアライズドテーブル内の既存のデータを上書きします。バッチモードでは、エンジンは増分更新を優先し、増分更新が不可能な場合にのみフル更新にフォールバックします。

増分更新の条件

増分更新は、マテリアライズドテーブルが以下のすべての条件を満たす場合にのみ実行されます。

テーブル定義で partition.fields.#.date-formatter パラメーターが設定されていない。
ソーステーブルにプライマリキーが定義されていない。

マテリアライズドテーブル定義のクエリが、以下の表で説明されているように増分更新をサポートしている。

SQL 句	サポート状況
SELECT	列選択およびユーザー定義関数を含むスカラー関数式でサポートされます。集計関数はサポートされていません。
FROM	テーブル名とサブクエリでサポートされます。
WITH	共通テーブル式 (CTE) でサポートされます。
WHERE	ユーザー定義関数を含むさまざまなスカラー関数式を含むフィルター条件でサポートされます。`WHERE [NOT] EXISTS <subquery>` や `WHERE <column> [NOT] IN <subquery>` などのサブクエリはサポートされていません。
UNION	`UNION ALL` のみがサポートされます。
JOIN	`INNER JOIN` はサポートされます。 `LEFT/RIGHT/FULL [OUTER] JOIN` は、以下で説明する特定の `LATERAL JOIN` およびルックアップ結合の場合を除き、サポートされていません。テーブル関数式 (ユーザー定義関数を含む) を伴う `[LEFT [OUTER]] JOIN LATERAL` はサポートされます。ルックアップ結合については、`A [LEFT [OUTER]] JOIN B FOR SYSTEM_TIME AS OF PROCTIME()` のみがサポートされます。説明 `JOIN` キーワードを使用しない暗黙的な結合 (例：`SELECT * FROM a, b WHERE a.id = b.id`) はサポートされます。 `INNER JOIN` の増分計算では、両方のソーステーブルから完全なデータを読み取ります。
GROUP BY	サポートされていません。

増分更新の例

例 1：スカラー関数を使用して orders ソーステーブルのデータを処理します。

CREATE MATERIALIZED TABLE mt_shipped_orders (
    PRIMARY KEY (order_id) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT 
    order_id,
    COALESCE(customer_id, 'Unknown') AS customer_id,
    CAST(order_amount AS DECIMAL(10, 2)) AS order_amount,
    CASE 
        WHEN status = 'shipped' THEN 'Completed'
        WHEN status = 'pending' THEN 'In Progress'
        ELSE 'Unknown'
    END AS order_status,
    DATE_FORMAT(order_ts, 'yyyyMMdd') AS order_date,
    UDSF_ProcessFunction(notes) AS notes
FROM 
    orders
WHERE
    status = 'shipped';

例 2：LATERAL JOIN と lookup join を使用して orders ソーステーブルのデータをエンリッチします。

CREATE MATERIALIZED TABLE mt_enriched_orders (
    PRIMARY KEY (order_id, order_tag) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
WITH o AS (
    SELECT
        order_id,
        product_id,
        quantity,
        proc_time,
        e.tag AS order_tag
    FROM 
        orders,
        LATERAL TABLE(UDTF_StringSplitFunction(tags, ',')) AS e(tag))
SELECT 
    o.order_id,
    o.product_id,
    p.product_name,
    p.category,
    o.quantity,
    p.price,
    o.quantity * p.price AS total_amount,
    order_tag
FROM o 
LEFT JOIN 
    product_info FOR SYSTEM_TIME AS OF PROCTIME() AS p
ON 
    o.product_id = p.product_id;

データバックフィル

以前は、履歴データのストリーム処理結果を修正するには、別のバッチジョブを開発する必要がありました。マテリアライズドテーブルを使用すると、履歴データパーティションを直接バックフィルできます。このアプローチにより、バッチ処理とストリーミング処理が統合され、開発および運用保守コストが削減されます。

カタログの下にある対象のマテリアライズドテーブルをクリックします。
データ情報 タブで、データをバックフィルします。
マテリアライズドテーブルを作成したときにパーティション列を定義した場合、それはパーティションテーブルです。そうでない場合は、非パーティションテーブルです。
パーティションテーブル
データパーティション セクションで、初めてデータをバックフィルする場合や必要なパーティションが存在しない場合は 更新のトリガー をクリックします。パーティションが既に存在する場合は、特定のパーティションを選択し、操作列の更新をクリックします。
パラメーター
- パーティション列：テーブルのパーティション列。たとえば、20241201 と入力すると、パーティション ds=20241201 のすべてのデータがバックフィルされます。
- タスク名：データバックフィルタスクの名前。
- 更新範囲 (オプション)：ダウンストリームのマテリアライズドテーブルに更新をカスケードするかどうかを指定します。現在のテーブルから開始して、データリネージ内のすべてのマテリアライズドテーブルが更新されます。サポートされるダウンストリームリネージの最大深度は 6 レベルです。
  説明
  パーティションテーブルを更新する場合、ダウンストリームのマテリアライズドテーブルは、開始テーブルとまったく同じパーティション列を持つ必要があります。そうでない場合、更新操作は失敗します。
  リネージ内のいずれかのマテリアライズドテーブルの更新が失敗した場合、後続のすべてのダウンストリームノードも失敗します。
- デプロイ先：キューまたはセッションクラスターを選択できます。デフォルトの選択は default-queue です。
非パーティションテーブル
データステータス セクションで、更新をクリックします。
パラメーター
- タスク名：データバックフィルタスクの名前。
- 更新範囲：このオプションは非パーティションテーブルでは利用できません。
  説明
  更新中、ダウンストリームのデータは完全にリフレッシュされます。
  リネージ内のいずれかのマテリアライズドテーブルの更新が失敗した場合、後続のすべてのダウンストリームノードも失敗します。
  開始テーブルがストリーミングジョブによって更新される非パーティションテーブルである場合、カスケード更新はサポートされません。
- デプロイ先：キューまたはセッションクラスターを選択できます。デフォルトの選択は default-queue です。
スケジュールされたバッチバックフィル。
タスクオーケストレーションを使用して、マテリアライズドテーブルのワークフローを作成し、スケジュールに基づいてバックフィルジョブを実行できます。また、ワークフローのデータバックフィル機能を使用して、指定した時間範囲のデータを一括でバックフィルすることもできます。

データの鮮度の変更

対応するカタログの下で、マテリアライズドテーブル データベースをクリックし、次に対象の マテリアライズドテーブル をクリックします。
右上隅で、鮮度の編集 をクリックします。
- マテリアライズドテーブルにプライマリキーがない場合、その更新方法をストリーミングとバッチの間で切り替えることはできません。システムは、データの鮮度が 30 分未満の場合はストリーミングジョブを使用し、30 分以上の場合はバッチジョブを使用します。したがって、プライマリキーのないテーブルでは、この 30 分のしきい値を超えて鮮度を変更することは許可されていません。
- アップストリームテーブルがマテリアライズドテーブルである場合は、ダウンストリームテーブルのデータの鮮度がアップストリームテーブルのデータの鮮度の正の整数倍であることを確認してください。
- データの鮮度の最大値は 1 日です。

データリネージの表示

左側のナビゲーションウィンドウで、オペレーションセンター > データリネージ を選択して、マテリアライズドテーブルのデータリネージページに移動します。このページでは、すべてのマテリアライズドテーブル間のリネージ関係を表示できます。また、マテリアライズドテーブルに対して 更新の開始/停止 や 鮮度の編集 などの操作を直接実行することもできます。詳細をクリックすると、対応するマテリアライズドテーブルの詳細ページに移動します。

Realtime Compute for Apache Flink:マテリアライズドテーブルの作成と使用

制限事項

マテリアライズドテーブルの作成

構文

パラメーター

操作手順

マテリアライズドテーブルクエリの変更

制限事項

例

増分更新

制限事項

更新モード

増分更新の条件

増分更新の例

データバックフィル

パーティションテーブル

非パーティションテーブル

データの鮮度の変更

データリネージの表示

関連ドキュメント