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

制限事項

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

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

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

例

1. [テーブルの編集] をクリックしてクエリを変更します。コード例は次のとおりです。

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

2. [プレビュー] をクリックすると、変更前後の比較が表示されます。

   [マテリアライズドテーブルの変更詳細] ダイアログボックスには、次の変更が表示されます。[SQL ステートメント] エリアでは、SQL ステートメントが元の SELECT * から SELECT *, orders.price * orders.quantity AS total_price に変更され、新しい WHERE 条件 orders.price * orders.quantity > 1000 が追加されます。また、[テーブル列] エリアでは、新しい total_price (DOUBLE) フィールドが緑色でハイライト表示されます。

3. [OK] をクリックすると、[テーブルスキーマ] タブで新しく追加された列と変更されたクエリロジックを表示できます。

増分更新

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';

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;

データバックフィル

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

1. カタログの下にある対象のマテリアライズドテーブルをクリックします。

2. [データ情報] タブで、データをバックフィルします。

   マテリアライズドテーブルを作成したときにパーティションキーを定義した場合はパーティションテーブルであり、それ以外の場合は非パーティション化テーブルです。

   パーティションテーブル

   [パーティション] セクションで、初めてデータをバックフィルする場合、または必要なパーティションが存在しない場合は、[トリガー更新] をクリックします。 パーティションが既に存在する場合は、特定のパーティションを選択し、「操作」列の [更新] をクリックできます。

   [データ情報] タブをクリックし、ページ下部の [データパーティション] エリアで操作を実行します。

   パラメーター

   • パーティションキー：テーブルのパーティションキーです。たとえば、20241201 と入力すると、パーティション ds=20241201 内のすべてのデータがバックフィルされます。

   • タスク名：データバックフィルタスクの名前です。

   • 更新範囲 (オプション)：ダウンストリームのマテリアライズドテーブルに更新をカスケードするかどうかを指定します。現在のテーブルから始まり、データリネージ内のすべてのマテリアライズドテーブルが更新されます。サポートされるダウンストリームリネージの最大深度は 6 レベルです。

     説明

     • パーティションテーブルを更新する場合、ダウンストリームのマテリアライズドテーブルは、開始テーブルとまったく同じパーティションキーを持つ必要があります。そうでない場合、更新操作は失敗します。

     • リネージ内のいずれかのマテリアライズドテーブルで更新が失敗した場合、後続のすべてのダウンストリームノードも失敗します。

   • デプロイターゲット：キューまたはセッションクラスターを選択できます。デフォルトの選択は default-queue です。

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

   [データ詳細] セクションで、[更新] をクリックします。

   パラメーター

   • タスク名：データバックフィルタスクの名前です。

   • 更新範囲：このオプションは非パーティション化テーブルでは利用できません。

     説明

     • 更新中、ダウンストリームのデータは完全にリフレッシュされます。

     • リネージ内のいずれかのマテリアライズドテーブルで更新が失敗した場合、後続のすべてのダウンストリームノードも失敗します。

     • 開始テーブルがストリーミングジョブによって更新される非パーティション化テーブルである場合、カスケード更新はサポートされません。

   • デプロイターゲット：キューまたはセッションクラスターを選択できます。デフォルトの選択は default-queue です。

3. スケジュールおよびバッチによるバックフィル。

   タスクオーケストレーションを使用して、マテリアライズドテーブルのワークフローを作成し、スケジュールに基づいてバックフィルジョブを実行できます。また、ワークフローのデータバックフィル機能を使用して、指定した時間範囲のデータを一括でバックフィルすることもできます。

データ鮮度の変更

1. 対応するカタログで、[マテリアライズドテーブル] データベースをクリックして、目的の [マテリアライズドテーブル] をクリックします。

2. 右上隅で、[鮮度の変更] をクリックします。

   • マテリアライズドテーブルにプライマリキーがない場合、その更新方法をストリーミングとバッチの間で切り替えることはできません。システムは、データ鮮度の値が 30 分未満の場合はストリーミングジョブを使用し、30 分以上の場合はバッチジョブを使用します。したがって、プライマリキーのないテーブルでは、この 30 分のしきい値を超えて鮮度を変更することは許可されていません。

   • アップストリームテーブルがマテリアライズドテーブルである場合、ダウンストリームテーブルのデータ鮮度がアップストリームテーブルのデータ鮮度の正の整数倍であることを確認してください。

   • データ鮮度の最大値は 1 日です。

データリネージの表示

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

マテリアライズドテーブルのノードをクリックすると、その詳細パネルが展開されます。パネルには、[データ鮮度] の値、[最終更新時刻]、[マテリアライズドテーブルのステータス] が表示され、[更新をトリガー] アクションが利用できます。