Realtime Compute for Apache Flink：建立及使用物化表

使用限制

僅Realtime Compute引擎VVR 8.0.11及以上版本支援。

物化表的更新模式

物化表提供三種更新模式：流更新模式、全量批更新模式、增量批更新模式。

物化表的流批模式由資料新鮮度決定（小於30分鐘為流模式，大於等於30分鐘為批模式）。在批處理模式下，全量更新或累加式更新由引擎自動判斷。累加式更新僅計算自上次更新以來的增量資料，並將其合并至物化表中；而全量更新則會計算整個表或整個分區的資料，並對物化表中的資料進行覆蓋。批模式下引擎優先考慮使用累加式更新，當累加式更新無法支援此物化表時，才會使用全量更新。

累加式更新條件

僅當物化表滿足以下條件時，才會使用累加式更新。

• 在定義物化表時，未配置partition.fields.#.date-formatter參數指定分區欄位的時間格式。

• 源表未定義主鍵。

• 物化表中的查詢語句支援累加式更新情況如下：

  SQL語句	支援情況
  SELECT	支援選擇列以及純量涵式運算式（包括使用者自訂函數），暫不支援彙總函式。
  FROM	支援表名或子查詢。
  WITH	支援通用資料表運算式Common Table Expression（CTE）。
  WHERE	支援過濾條件包括各種純量涵式運算式（包括使用者自訂函數），不支援包含子查詢（如 WHERE [NOT] EXISTS <子查詢>、WHERE <列名> [NOT] IN <子查詢> 等）。
  UNION	僅支援UNION ALL。
  JOIN	支援INNER JOIN。 暫不支援LEFT/RIGHT/FULL [OUTER] JOIN，除了以下LATERAL JOIN及Lookup Join的情況。 支援[LEFT [OUTER]] JOIN LATERAL table函數運算式（包括使用者自訂函數）。 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的增量計算仍然會讀取兩個源表的全量資料。
  GROUP BY	暫不支援。

累加式更新樣本

樣本一：對源表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';

樣本二：對源表orders資料使用Lateral Join以及Lookup Join進行資訊補充。

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;