E-MapReduce:Mempercepat kueri data lake menggunakan tampilan yang di-materialisasi

Pilih kebijakan pembaruan yang sesuai

Saat ini, StarRocks tidak dapat mendeteksi perubahan data tingkat partisi di Katalog Hudi. Oleh karena itu, begitu tugas pembaruan dipicu, pembaruan penuh akan dilakukan.

Untuk Katalog Hive, Katalog Iceberg (mulai v3.1.4), Katalog JDBC (mulai v3.1.4, dan hanya partisi Range MySQL yang didukung), serta Katalog Paimon (mulai v3.2.1), StarRocks mendukung deteksi perubahan data tingkat partisi. Akibatnya, StarRocks dapat:

• Hanya memperbarui partisi yang datanya berubah, sehingga menghindari pembaruan penuh dan mengurangi konsumsi sumber daya akibat proses pembaruan.

• Menjamin konsistensi data hingga batas tertentu selama penulisan ulang kueri. Jika tabel dasar di data lake mengalami perubahan data, kueri tidak akan ditulis ulang untuk menggunakan tampilan yang di-materialisasi.

Perhatikan bahwa untuk pembaruan berbasis partisi, kunci partisi tampilan yang di-materialisasi harus termasuk dalam kunci partisi tabel dasar.

Mulai v3.2.3, StarRocks mendukung pembuatan tampilan yang di-materialisasi berpartisi pada tabel Iceberg yang menggunakan Partition Transforms. Tampilan yang di-materialisasi akan dipartisi berdasarkan kolom yang ditransformasi. Saat ini, hanya tabel Iceberg yang menggunakan Transform identity, year, month, day, atau hour yang didukung.

Contoh berikut menunjukkan definisi tabel Iceberg yang menggunakan Transform day dan tampilan yang di-materialisasi selaras partisi yang dibuat pada tabel tersebut:

-- Definisi tabel 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))

-- Buat tampilan yang di-materialisasi berdasarkan tabel Iceberg di atas.
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`;

Untuk Katalog Hive, Anda dapat mengaktifkan fitur pembaruan cache metadata Hive agar StarRocks dapat mendeteksi perubahan data tingkat partisi. Setelah fitur ini diaktifkan, StarRocks secara berkala mengakses layanan Hive Metastore (HMS) atau AWS Glue untuk memeriksa informasi metadata data panas yang sering dikueri.

Item konfigurasi

Untuk mengaktifkan fitur pembaruan cache metadata Hive, Anda dapat menggunakan ADMIN SET FRONTEND CONFIG untuk mengatur item konfigurasi dinamis FE berikut. Sintaksnya sebagai berikut.

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

Untuk Katalog Iceberg, mulai v3.1.4, StarRocks mendukung deteksi perubahan data tingkat partisi. Saat ini, hanya tabel Iceberg V1 yang didukung.

Aktifkan penulisan ulang kueri untuk tampilan yang di-materialisasi Katalog Eksternal

Karena konsistensi data yang kuat tidak dapat dijamin, StarRocks secara default menonaktifkan fitur penulisan ulang kueri untuk tampilan yang di-materialisasi Katalog Hudi dan JDBC. Anda dapat mengaktifkan fitur ini dengan mengatur properti force_external_table_query_rewrite menjadi true saat membuat tampilan yang di-materialisasi. Untuk tampilan yang di-materialisasi yang dibuat pada tabel di Katalog Hive, fitur penulisan ulang kueri diaktifkan secara default. Dalam skenario yang melibatkan penulisan ulang kueri, jika Anda menggunakan pernyataan kueri yang sangat kompleks untuk membangun tampilan yang di-materialisasi, kami merekomendasikan agar Anda membagi pernyataan kueri tersebut dan membangun beberapa tampilan yang di-materialisasi sederhana secara bersarang. Tampilan yang di-materialisasi bersarang lebih fleksibel dan dapat beradaptasi dengan pola kueri yang lebih luas.

Berikut adalah contohnya.

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;

Kasus 1: Mempercepat komputasi Join di data lake

Anda dapat menggunakan tampilan yang di-materialisasi untuk mempercepat kueri Join di data lake.

Asumsikan kueri berikut pada katalog Hive merupakan kueri lambat.

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

Dengan menganalisis profil kueri, Anda mungkin menyadari bahwa sebagian besar waktu eksekusi kueri dihabiskan pada Hash Join antara tabel lineorder dan tabel dimensi lainnya pada kolom lo_orderdate.

Di sini, Q1 dan Q2 melakukan agregasi setelah menggabungkan lineorder dan dates, sedangkan Q3 melakukan agregasi setelah menggabungkan lineorder, dates, part, dan supplier.

Oleh karena itu, Anda dapat memanfaatkan kemampuan penulisan ulang View Delta Join StarRocks untuk membangun tampilan yang di-materialisasi yang menggabungkan lineorder, dates, part, dan supplier.

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 (
    -- Tentukan kendala UNIK.
    "unique_constraints" = "
    hive.ssb_1g_csv.supplier.s_suppkey;
    hive.ssb_1g_csv.part.p_partkey;
    hive.ssb_1g_csv.dates.d_datekey",
    -- Tentukan kendala kunci asing.
    "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)",
    -- Aktifkan penulisan ulang kueri.
    "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;

Kasus 2: Mempercepat agregasi dan agregasi pasca-join di data lake

Tampilan yang di-materialisasi dapat digunakan untuk mempercepat kueri agregasi, baik pada satu tabel maupun melibatkan beberapa tabel.

• Kueri agregasi satu tabel

  Untuk kueri satu tabel yang khas, jika Query Profile menunjukkan bahwa node AGGREGATE menghabiskan banyak waktu, Anda dapat membangun tampilan yang di-materialisasi yang menggunakan operator agregasi umum. Asumsikan kueri berikut lambat.

  --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 adalah kueri yang menghitung jumlah pesanan harian yang unik. Karena count distinct sangat boros sumber daya, Anda dapat membuat dua jenis tampilan yang di-materialisasi berikut:

  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 harus bertipe BIGINT agar dapat digunakan untuk penulisan ulang kueri.
  lo_orderdate, bitmap_union(to_bitmap(lo_orderkey))
  FROM hive.ssb_1g_csv.lineorder
  GROUP BY lo_orderdate;

  Catatan

  Jangan buat tampilan yang di-materialisasi dengan klausa LIMIT dan ORDER BY di sini untuk menghindari kegagalan penulisan ulang.

• Kueri agregasi multi-tabel

  Dalam skenario yang melibatkan agregasi hasil join, Anda dapat membuat tampilan yang di-materialisasi bersarang pada tampilan yang di-materialisasi yang sudah ada yang menggabungkan beberapa tabel untuk lebih lanjut mengagregasi hasilnya. Misalnya, berdasarkan contoh pada Kasus 1, Anda dapat membuat tampilan yang di-materialisasi berikut untuk mempercepat Q1 dan Q2 karena pola agregasinya serupa.

  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;

  Anda juga dapat melakukan join dan agregasi dalam satu tampilan yang di-materialisasi. Jenis tampilan yang di-materialisasi ini memiliki peluang lebih sedikit untuk penulisan ulang kueri karena perhitungannya lebih spesifik. Namun, mereka menggunakan ruang penyimpanan lebih sedikit setelah agregasi. Anda dapat memilih pendekatan terbaik untuk skenario spesifik Anda.

  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;

Kasus 3: Mempercepat komputasi Join pasca-agregasi di data lake

Dalam beberapa kasus, Anda mungkin perlu terlebih dahulu melakukan komputasi agregasi pada satu tabel, lalu melakukan kueri Join dengan tabel lain. Untuk memanfaatkan sepenuhnya fitur penulisan ulang kueri StarRocks, kami merekomendasikan agar Anda membangun tampilan yang di-materialisasi bersarang. Berikut adalah contohnya.

--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 pertama-tama melakukan agregasi pada tabel customer, lalu melakukan Join dan agregasi pada tabel lineorder. Kueri serupa mungkin melibatkan kondisi filter berbeda pada c_region dan lo_orderdate. Untuk memanfaatkan fitur penulisan ulang kueri, Anda dapat membuat dua tampilan yang di-materialisasi, satu untuk agregasi dan satu lagi untuk join.

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

Kasus 4: Pemisahan data panas dan dingin untuk data real-time dan historis di data lake

Pertimbangkan skenario di mana data baru dari tiga hari terakhir ditulis langsung ke StarRocks, sedangkan data yang lebih tua dari tiga hari ditulis secara batch ke Hive. Namun, kueri mungkin tetap perlu mengakses data dari tujuh hari terakhir. Dalam kasus ini, Anda dapat menggunakan tampilan yang di-materialisasi untuk membuat model kedaluwarsa data sederhana.

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

Anda dapat lebih lanjut membangun tampilan atau tampilan yang di-materialisasi berdasarkan logika bisnis lapisan atas untuk mengenkapsulasi komputasi.