使用物化视图加速数据湖查询

选择合适的刷新策略

目前，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后聚合计算

物化视图可用于加速聚合查询，无论是在单个表上还是涉及多个表。使用OSS Foreign Table进行数据湖分析

• 单表聚合查询

  对于典型的单表查询，如果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);

您可以根据上层业务逻辑进一步构建视图或物化视图，以封装计算。