E-MapReduce：使用物化視圖加速資料湖查詢

選擇合適的重新整理策略

目前，StarRocks無法檢測Hudi Catalog中的分區層級資料更改。因此，一旦觸發重新整理任務，將執行全量重新整理。

對於Hive Catalog、Iceberg Catalog（從v3.1.4版本起）、JDBC Catalog（從v3.1.4版本起，且僅支援MySQL Range分區）和Paimon Catalog（從v3.2.1版本起），StarRocks支援檢測分區層級資料更改。從而，StarRocks可以：

• 僅重新整理資料有更改的分區，避免全量重新整理，減少重新整理導致的資源消耗。

• 在查詢改寫期間在一定程度上確保資料一致性。如果資料湖中的基表發生資料更改，查詢將不會被改寫為使用物化視圖。

請注意，如需按照分區重新整理，物化視圖的分區鍵必須包含在基表的分區鍵中。

從v3.2.3版本開始，StarRocks支援在使用Partition Transforms的Iceberg表上建立分區物化視圖，物化視圖將根據變換後的列進行分區。目前，僅支援使用identity、year、month、day或hourTransform的Iceberg表。

以下樣本展示了一個使用dayTransform的Iceberg表的定義，並在該表上建立了一個分區對齊的物化視圖：

-- Iceberg表定義。
CREATE TABLE spark_catalog.test_db.iceberg_sample_datetime_day (
  id         BIGINT,
  data       STRING,
  category   STRING,
  ts         TIMESTAMP)
USING iceberg
PARTITIONED BY (days(ts))

-- 基於以上Iceberg表建立物化視圖。
CREATE MATERIALIZED VIEW `test_iceberg_datetime_day_mv` (`id`, `data`, `category`, `ts`)
PARTITION BY (`ts`)
DISTRIBUTED BY HASH(`id`)
REFRESH MANUAL
AS 
SELECT 
  `iceberg_sample_datetime_day`.`id`, 
  `iceberg_sample_datetime_day`.`data`, 
  `iceberg_sample_datetime_day`.`category`, 
  `iceberg_sample_datetime_day`.`ts`
FROM `iceberg`.`test`.`iceberg_sample_datetime_day`;

對於Hive Catalog，您可以啟用Hive中繼資料快取重新整理功能，允許StarRocks在分區層級檢測資料更改。啟用此功能後，StarRocks定期訪問Hive中繼資料存放區服務（HMS）或AWS Glue，以檢查最近查詢的熱資料的中繼資料資訊。

配置項

要啟用Hive中繼資料快取重新整理功能，您可以使用ADMIN SET FRONTEND CONFIG設定以下FE動態配置項。文法如下。

ADMIN SET FRONTEND CONFIG ("key" = "value")

對於Iceberg Catalog，從v3.1.4版本開始，StarRocks支援檢測分區層級的資料更改，當前只支援Iceberg V1表。

啟用External Catalog物化視圖的查詢改寫

由於不保證資料的強一致性，StarRocks預設禁用Hudi和JDBC Catalog物化視圖的查詢改寫功能。您可以通過在建立物化視圖時將Propertyforce_external_table_query_rewrite設定為true來啟用此功能。對於基於Hive Catalog中的表建立的物化視圖，查詢改寫功能預設開啟。在涉及查詢改寫的情況下，如果您使用非常複雜的查詢語句來構建物化視圖，我們建議您拆分查詢語句並以嵌套方式構建多個簡單的物化視圖。嵌套的物化視圖更加靈活，可以適應更廣泛的查詢模式。

樣本如下。

CREATE MATERIALIZED VIEW ex_mv_par_tbl
PARTITION BY emp_date
DISTRIBUTED BY hash(empid)
PROPERTIES (
"force_external_table_query_rewrite" = "true"
) 
AS
SELECT empid, deptno, emp_date
FROM `hudi_catalog`.`emp_db`.`emps_par_tbl`
WHERE empid < 5;

案例一：加速資料湖中的Join計算

您可以使用物化視圖來加速資料湖中的Join查詢。

假設以下Hive catalog上的查詢為慢查詢。

--Q1
SELECT SUM(lo_extendedprice * lo_discount) AS REVENUE
FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates
WHERE
    lo_orderdate = d_datekey
    AND d_year = 1993
    AND lo_discount BETWEEN 1 AND 3
    AND lo_quantity < 25;

--Q2
SELECT SUM(lo_extendedprice * lo_discount) AS REVENUE
FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates
WHERE
    lo_orderdate = d_datekey
    AND d_yearmonth = 'Jan1994'
    AND lo_discount BETWEEN 4 AND 6
    AND lo_quantity BETWEEN 26 AND 35;

--Q3 
SELECT SUM(lo_revenue), d_year, p_brand
FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates, hive.ssb_1g_csv.part, hive.ssb_1g_csv.supplier
WHERE
    lo_orderdate = d_datekey
    AND lo_partkey = p_partkey
    AND lo_suppkey = s_suppkey
    AND p_brand BETWEEN 'MFGR#2221' AND 'MFGR#2228'
    AND s_region = 'ASIA'
GROUP BY d_year, p_brand
ORDER BY d_year, p_brand;

通過分析查詢概要，您可能會注意到查詢執行時間主要花費在表lineorder與其他維度資料表在列lo_orderdate上的Hash Join上。

此處，Q1和Q2在Joinlineorder和dates後執行彙總，而Q3在Joinlineorder、dates、part和supplier後執行彙總。

因此，您可以利用StarRocks的View Delta Join改寫能力來構建物化視圖，對lineorder、dates、part和supplier進行Join。

CREATE MATERIALIZED VIEW lineorder_flat_mv
DISTRIBUTED BY HASH(LO_ORDERDATE, LO_ORDERKEY) BUCKETS 48
PARTITION BY LO_ORDERDATE
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
PROPERTIES (
    -- 指定唯一約束。
    "unique_constraints" = "
    hive.ssb_1g_csv.supplier.s_suppkey;
    hive.ssb_1g_csv.part.p_partkey;
    hive.ssb_1g_csv.dates.d_datekey",
    -- 指定外鍵約束。
    "foreign_key_constraints" = "
    hive.ssb_1g_csv.lineorder(lo_partkey) REFERENCES hive.ssb_1g_csv.part(p_partkey);
    hive.ssb_1g_csv.lineorder(lo_suppkey) REFERENCES hive.ssb_1g_csv.supplier(s_suppkey);
    hive.ssb_1g_csv.lineorder(lo_orderdate) REFERENCES hive.ssb_1g_csv.dates(d_datekey)",
    -- 啟用查詢改寫。
    "force_external_table_query_rewrite" = "TRUE"
)
AS SELECT
       l.LO_ORDERDATE AS LO_ORDERDATE,
       l.LO_ORDERKEY AS LO_ORDERKEY,
       l.LO_PARTKEY AS LO_PARTKEY,
       l.LO_SUPPKEY AS LO_SUPPKEY,
       l.LO_QUANTITY AS LO_QUANTITY,
       l.LO_EXTENDEDPRICE AS LO_EXTENDEDPRICE,
       l.LO_DISCOUNT AS LO_DISCOUNT,
       l.LO_REVENUE AS LO_REVENUE,
       s.S_REGION AS S_REGION,
       p.P_BRAND AS P_BRAND,
       d.D_YEAR AS D_YEAR,
       d.D_YEARMONTH AS D_YEARMONTH
   FROM hive.ssb_1g_csv.lineorder AS l
            INNER JOIN hive.ssb_1g_csv.supplier AS s ON s.S_SUPPKEY = l.LO_SUPPKEY
            INNER JOIN hive.ssb_1g_csv.part AS p ON p.P_PARTKEY = l.LO_PARTKEY
            INNER JOIN hive.ssb_1g_csv.dates AS d ON l.LO_ORDERDATE = d.D_DATEKEY;

案例二：加速資料湖中的彙總和Join後彙總計算

物化視圖可用於加速彙總查詢，無論是在單個表上還是涉及多個表。

• 單表彙總查詢

  對於典型的單表查詢，如果Query Profile顯示AGGREGATE節點消耗了大量時間，您可以使用常見的彙總運算元構建物化視圖。假設以下查詢速度較慢。

  --Q4
  SELECT
  lo_orderdate, count(distinct lo_orderkey)
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate
  ORDER BY lo_orderdate limit 100;

  Q4是一個計算每日去重訂單數量的查詢，因為count distinct的消耗較大，可以建立下列兩類物化視圖：

  CREATE MATERIALIZED VIEW mv_2_1 
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  AS 
  SELECT
  lo_orderdate, count(distinct lo_orderkey)
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate;
  
  CREATE MATERIALIZED VIEW mv_2_2 
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  AS 
  SELECT
  -- lo_orderkey必須是BIGINT類型，以便可以用於查詢改寫。
  lo_orderdate, bitmap_union(to_bitmap(lo_orderkey))
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate;

  說明

  此處不要建立帶有LIMIT和ORDER BY子句的物化視圖，以避免改寫失敗。

• 多表彙總查詢

  在涉及Join結果彙總的情境中，您可以在現有Join多表的物化視圖上建立嵌套物化視圖，進一步彙總串連結果。例如，根據案例一中的樣本，您可以建立以下物化視圖以加速Q1和Q2，因為它們的彙總模式相似。

  CREATE MATERIALIZED VIEW mv_2_3
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  AS 
  SELECT
  lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth, SUM(lo_extendedprice * lo_discount) AS REVENUE
  FROM lineorder_flat_mv
  GROUP BY lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth;

  當然，您也可以在單個物化視圖中同時執行Join和彙總計算。儘管這類的物化視圖改寫查詢的機會更少（因為涉及的計算更加具體），但在彙總後，其佔用儲存空間更少。您可以基於您的真實情境選擇。

  CREATE MATERIALIZED VIEW mv_2_4
  DISTRIBUTED BY HASH(lo_orderdate)
  PARTITION BY LO_ORDERDATE
  REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
  PROPERTIES (
      "force_external_table_query_rewrite" = "TRUE"
  )
  AS
  SELECT lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth, SUM(lo_extendedprice * lo_discount) AS REVENUE
  FROM hive.ssb_1g_csv.lineorder, hive.ssb_1g_csv.dates
  WHERE lo_orderdate = d_datekey
  GROUP BY lo_orderdate, lo_discount, lo_quantity, d_year, d_yearmonth;

案例三：加速資料湖中的彙總後Join計算

在某些情況下，您可能需要首先對一個表執行彙總計算，然後再與其他表執行Join查詢。為了充分利用StarRocks的查詢改寫功能，我們建議您構建嵌套的物化視圖。樣本如下。

--Q5
SELECT * FROM  (
    SELECT 
      l.lo_orderkey, l.lo_orderdate, c.c_custkey, c_region, sum(l.lo_revenue)
    FROM 
      hive.ssb_1g_csv.lineorder l 
      INNER JOIN (
        SELECT distinct c_custkey, c_region 
        from 
          hive.ssb_1g_csv.customer 
        WHERE 
          c_region IN ('ASIA', 'AMERICA') 
      ) c ON l.lo_custkey = c.c_custkey
      GROUP BY  l.lo_orderkey, l.lo_orderdate, c.c_custkey, c_region
  ) c1 
WHERE 
  lo_orderdate = '19970503'

Q5首先在customer表上執行彙總，然後在lineorder表上執行Join和彙總。類似的查詢可能涉及對c_region和lo_orderdate的不同過濾條件。為了利用查詢改寫功能，您可以建立兩個物化視圖，一個用於彙總，另一個用於串連。

--mv_3_1
CREATE MATERIALIZED VIEW mv_3_1
DISTRIBUTED BY HASH(c_custkey)
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
PROPERTIES (
    "force_external_table_query_rewrite" = "TRUE"
)
AS
SELECT distinct c_custkey, c_region from hive.ssb_1g_csv.customer; 

--mv_3_2
CREATE MATERIALIZED VIEW mv_3_2
DISTRIBUTED BY HASH(lo_orderdate)
PARTITION BY LO_ORDERDATE
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
PROPERTIES (
    "force_external_table_query_rewrite" = "TRUE"
)
AS
SELECT l.lo_orderdate, l.lo_orderkey, mv.c_custkey, mv.c_region, sum(l.lo_revenue)
FROM hive.ssb_1g_csv.lineorder l 
INNER JOIN mv_3_1 mv
ON l.lo_custkey = mv.c_custkey
GROUP BY l.lo_orderkey, l.lo_orderdate, mv.c_custkey, mv.c_region;

案例四：對即時資料和資料湖中的歷史資料進行冷熱分離

例如以下情景：過去三天內的新資料直接寫入StarRocks，三天前的舊資料經過校對後批量寫入Hive。但是查詢仍然可能有涉及過去七天資料。在這種情況下，您可以使用物化視圖建立一個簡單的模型來自動到期資料。

CREATE MATERIALIZED VIEW mv_4_1 
DISTRIBUTED BY HASH(lo_orderdate)
PARTITION BY LO_ORDERDATE
REFRESH ASYNC EVERY(INTERVAL 1 DAY) 
AS 
SELECT lo_orderkey, lo_orderdate, lo_revenue
FROM hive.ssb_1g_csv.lineorder
WHERE lo_orderdate<=current_date()
AND lo_orderdate>=date_add(current_date(), INTERVAL -4 DAY);

您可以根據上層商務邏輯進一步構建視圖或物化視圖，以封裝計算。