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

制限

VVR 8.0.11 以降を使用する Realtime Compute for Apache Flink でのみ、マテリアライズドテーブルの増分データを更新できます。

マテリアライズドテーブルの更新モード

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

マテリアライズドテーブルのデータ鮮度は、テーブルがストリーミングモードかバッチモードかを定義します。マテリアライズドテーブルのデータ鮮度が 30 分未満の場合、テーブルはストリーミングモードになります。マテリアライズドテーブルのデータ鮮度が 30 分以上の場合、テーブルはバッチモードになります。バッチモードでは、エンジンは自動的にフルアップデートまたは増分アップデートを実行します。増分アップデートは、最後のアップデート以降の増分データのみを計算し、結果をマテリアライズドテーブルにマージします。フルアップデートは、テーブル全体またはパーティション全体のデータを計算し、マテリアライズドテーブルのデータを上書きします。バッチモードでは、エンジンは優先的に増分アップデートを実行します。フルアップデートは、増分アップデートがマテリアライズドテーブルの要件を満たせない場合にのみ実行されます。

増分更新の条件

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

• マテリアライズドテーブルを定義するときに、partition.fields.#.date-formatter パラメーターを設定してパーティション列の時間形式を指定しません。

• ソーステーブルのプライマリキーを指定しません。

• 次のシナリオでは、マテリアライズドテーブルのクエリ文は増分更新をサポートしています。

  SQL 文	サポートされている増分更新
  SELECT	列をクエリでき、ユーザー定義関数を含むスカラー関数式がサポートされています。集計関数はサポートされていません。
  FROM	テーブル名クエリまたはサブクエリがサポートされています。
  WITH	共通テーブル式 (CTE) がサポートされています。
  WHERE	ユーザー定義関数を含むスカラー関数式は、フィルター条件でサポートされています。サブクエリはフィルター条件ではサポートされていません。たとえば、WHERE [NOT] EXISTS <subquery> および WHERE <列名> [NOT] IN <subquery> 形式のサブクエリ句はサポートされていません。
  UNION	UNION ALL のみがサポートされています。
  JOIN	INNER JOIN がサポートされています。 LEFT JOIN、RIGHT JOIN、および FULL [OUTER] JOIN は、Lateal Join と Lookup Join が使用される次のシナリオを除いて、サポートされていません。 [LEFT [OUTER]] JOIN LATERAL と、ユーザー定義関数を含むテーブル関数式がサポートされています。 Lookup Join では、A [LEFT [OUTER]] JOIN B FOR SYSTEM_TIME AS OF PROCTIME() のみがサポートされています。 説明 JOIN キーワードを使用しない JOIN 文 (例：SELECT * FROM a, b WHERE a.id = b.id) はサポートされています。 INNER JOIN を使用して増分更新を実行する場合でも、2 つのソーステーブルの完全データが読み取られます。
  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;