Hologres:Buat tabel dinamis

Contoh 1: Membuat Incremental Dynamic Table reguler

Sebelum melanjutkan, impor dataset publik tpch_10g ke Hologres dengan mengikuti panduan di Impor dataset publik dengan beberapa klik.

-- Buat Incremental Dynamic Table yang merefresh setiap 3 menit.
CREATE DYNAMIC TABLE public.tpch_q1_incremental 
WITH (
auto_refresh_mode='incremental',
freshness='3 minutes'
) AS SELECT
        l_returnflag,
        l_linestatus,
        COUNT(*) AS count_order
FROM
        hologres_dataset_tpch_10g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus;

Contoh 2: Buat Incremental Dynamic Table dari join stream-stream

Sebelum melanjutkan, impor dataset publik tpch_10g ke Hologres dengan mengikuti panduan di Impor dataset publik dengan beberapa klik.

-- Buat Incremental Dynamic Table dari join stream-stream.
CREATE DYNAMIC TABLE dt_join
  WITH (
    auto_refresh_mode='incremental',
    freshness='30 minutes'
  ) 
AS 
SELECT
        l_shipmode,
        SUM(CASE
                WHEN o_orderpriority = '1-URGENT'
                        OR o_orderpriority = '2-HIGH'
                        THEN 1
                ELSE 0
        END) AS high_line_count,
        SUM(CASE
                WHEN o_orderpriority <> '1-URGENT'
                        AND o_orderpriority <> '2-HIGH'
                        THEN 1
                ELSE 0
        END) AS low_line_count
FROM
        hologres_dataset_tpch_10g.orders,
        hologres_dataset_tpch_10g.lineitem
WHERE
        o_orderkey = l_orderkey
        AND l_shipmode IN ('FOB', 'AIR')
        AND l_commitdate < l_receiptdate
        AND l_shipdate < l_commitdate
        AND l_receiptdate >= DATE '1997-01-01'
        AND l_receiptdate < DATE '1997-01-01' + INTERVAL '1' YEAR
GROUP BY
        l_shipmode;

Contoh 3: Buat Dynamic Table dengan auto-refresh

Atur mode refresh ke auto. Mesin memprioritaskan incremental refresh dan beralih ke full refresh jika tidak didukung.

Sebelum melanjutkan, impor dataset publik tpch_10g ke Hologres dengan mengikuti panduan di Impor dataset publik dengan beberapa klik.

-- Buat Dynamic Table Auto-refresh yang secara cerdas menentukan mode refresh. Dalam contoh ini, hasilnya adalah incremental refresh.
CREATE DYNAMIC TABLE thch_q6_auto
  WITH (
    auto_refresh_mode='auto',
    freshness='1 hours'
       ) 
AS
SELECT
        SUM(l_extendedprice * l_discount) AS revenue
FROM
        hologres_dataset_tpch_100g.lineitem
WHERE
        l_shipdate >= DATE '1996-01-01'
        AND l_shipdate < DATE '1996-01-01' + INTERVAL '1' YEAR
        AND l_discount BETWEEN 0.02 - 0.01 AND 0.02 + 0.01
        AND l_quantity < 24;

Contoh 4: Buat Dynamic Table partisi logis

Untuk dasbor transaksi real-time, seringkali diperlukan tampilan near-real-time data saat ini dan koreksi data historis, yang memerlukan solusi analisis real-time dan offline terintegrasi (Kognisi bisnis dan data). Pendekatan umum menggunakan partisi logis `Dynamic Table` untuk skenario ini adalah sebagai berikut:

• Tabel dasar dipartisi berdasarkan hari. Partisi terbaru ditulis oleh Flink secara real-time/near-real-time, sedangkan partisi historis ditulis dari MaxCompute.

• Dynamic Table dibuat sebagai tabel partisi logis. Dua partisi terbaru aktif dan direfresh secara inkremental untuk analitik data near-real-time.

• Partisi historis tidak aktif dan menggunakan full refresh. Jika partisi historis tabel dasar telah dikoreksi atau diisi ulang, partisi tersebut dapat direfresh menggunakan full refresh.

Contoh ini menggunakan dataset publik dari GitHub.

1. Siapkan tabel dasar.

   Gunakan Flink untuk menulis data terbaru ke tabel dasar. Untuk langkah-langkah detail, lihat Analitik batch dan real-time terpadu pada event GitHub.

   DROP TABLE IF EXISTS gh_realtime_data;
   
   BEGIN;
   CREATE TABLE gh_realtime_data (
       id BIGINT,
       actor_id BIGINT,
       actor_login TEXT,
       repo_id BIGINT,
       repo_name TEXT,
       org_id BIGINT,
       org_login TEXT,
       type TEXT,
       created_at timestamp with time zone NOT NULL,
       action TEXT,
       iss_or_pr_id BIGINT,
       number BIGINT,
       comment_id BIGINT,
       commit_id TEXT,
       member_id BIGINT,
       rev_or_push_or_rel_id BIGINT,
       ref TEXT,
       ref_type TEXT,
       state TEXT,
       author_association TEXT,
       language TEXT,
       merged BOOLEAN,
       merged_at TIMESTAMP WITH TIME ZONE,
       additions BIGINT,
       deletions BIGINT,
       changed_files BIGINT,
       push_size BIGINT,
       push_distinct_size BIGINT,
       hr TEXT,
       month TEXT,
       year TEXT,
       ds TEXT,
       PRIMARY KEY (id,ds)
   )
   PARTITION BY LIST (ds);
   CALL set_table_property('public.gh_realtime_data', 'distribution_key', 'id');
   CALL set_table_property('public.gh_realtime_data', 'event_time_column', 'created_at');
   CALL set_table_property('public.gh_realtime_data', 'clustering_key', 'created_at');
   
   COMMENT ON COLUMN public.gh_realtime_data.id IS 'Event ID';
   COMMENT ON COLUMN public.gh_realtime_data.actor_id IS 'Event actor ID';
   COMMENT ON COLUMN public.gh_realtime_data.actor_login IS 'Event actor login name';
   COMMENT ON COLUMN public.gh_realtime_data.repo_id IS 'Repo ID';
   COMMENT ON COLUMN public.gh_realtime_data.repo_name IS 'Repo name';
   COMMENT ON COLUMN public.gh_realtime_data.org_id IS 'Repo organization ID';
   COMMENT ON COLUMN public.gh_realtime_data.org_login IS 'Repo organization name';
   COMMENT ON COLUMN public.gh_realtime_data.type IS 'Event type';
   COMMENT ON COLUMN public.gh_realtime_data.created_at IS 'Event time';
   COMMENT ON COLUMN public.gh_realtime_data.action IS 'Event action';
   COMMENT ON COLUMN public.gh_realtime_data.iss_or_pr_id IS 'Issue/pull_request ID';
   COMMENT ON COLUMN public.gh_realtime_data.number IS 'Issue/pull_request number';
   COMMENT ON COLUMN public.gh_realtime_data.comment_id IS 'Comment ID';
   COMMENT ON COLUMN public.gh_realtime_data.commit_id IS 'Commit ID';
   COMMENT ON COLUMN public.gh_realtime_data.member_id IS 'Member ID';
   COMMENT ON COLUMN public.gh_realtime_data.rev_or_push_or_rel_id IS 'Review/push/release ID';
   COMMENT ON COLUMN public.gh_realtime_data.ref IS 'Name of created/deleted resource';
   COMMENT ON COLUMN public.gh_realtime_data.ref_type IS 'Type of created/deleted resource';
   COMMENT ON COLUMN public.gh_realtime_data.state IS 'State of issue/pull_request/pull_request_review';
   COMMENT ON COLUMN public.gh_realtime_data.author_association IS 'Relationship between actor and repo';
   COMMENT ON COLUMN public.gh_realtime_data.language IS 'Programming language';
   COMMENT ON COLUMN public.gh_realtime_data.merged IS 'Whether merged';
   COMMENT ON COLUMN public.gh_realtime_data.merged_at IS 'Merge time';
   COMMENT ON COLUMN public.gh_realtime_data.additions IS 'Number of added lines';
   COMMENT ON COLUMN public.gh_realtime_data.deletions IS 'Number of deleted lines';
   COMMENT ON COLUMN public.gh_realtime_data.changed_files IS 'Number of changed files in pull request';
   COMMENT ON COLUMN public.gh_realtime_data.push_size IS 'Number of pushes';
   COMMENT ON COLUMN public.gh_realtime_data.push_distinct_size IS 'Number of distinct pushes';
   COMMENT ON COLUMN public.gh_realtime_data.hr IS 'Hour of event, e.g., 00 for 00:23';
   COMMENT ON COLUMN public.gh_realtime_data.month IS 'Month of event, e.g., 2015-10 for Oct 2015';
   COMMENT ON COLUMN public.gh_realtime_data.year IS 'Year of event, e.g., 2015';
   COMMENT ON COLUMN public.gh_realtime_data.ds IS 'Date of event, ds=yyyy-mm-dd';
   
   COMMIT;

2. Buat Dynamic Table partisi logis.

   CREATE  DYNAMIC TABLE ads_dt_github_event
   LOGICAL PARTITION BY LIST(ds)
   WITH (
     -- Properti dynamic table
     freshness = '5 minutes', 
     auto_refresh_mode = 'auto', 
     auto_refresh_partition_active_time = '2 days' ,
     partition_key_time_format = 'YYYY-MM-DD'
   )
   AS
   SELECT
       repo_name,
       COUNT(*) AS events,
       ds
   FROM
       gh_realtime_data
   GROUP BY repo_name,ds

3. Kueri Dynamic Table.

   SELECT * FROM ads_dt_github_event ;

4. Isi ulang partisi historis.

   Jika data historis di tabel dasar berubah (misalnya, data untuk '2025-04-01'), dan Dynamic Table perlu diperbarui, atur partisi historis ke mode full refresh dan picu refresh, sebaiknya dengan sumber daya Serverless Computing.

   REFRESH OVERWRITE DYNAMIC TABLE ads_dt_github_event
   PARTITION (ds = '2025-04-01') 
   WITH (
     refresh_mode = 'full'
   );

Contoh 5: Hitung UV dengan Incremental Dynamic Table

Mulai dari Hologres V3.1, Incremental Dynamic Table mendukung fungsi RB_BUILD_AGG untuk perhitungan seperti jumlah UV. Dibandingkan dengan pre-agregasi, pendekatan ini menawarkan:

• Performa lebih cepat: Hanya menghitung data inkremental.

• Biaya lebih rendah: Volume data dan penggunaan sumber daya berkurang, memungkinkan perhitungan periode lebih panjang.

Contoh:

1. Siapkan tabel detail pengguna.

   BEGIN;
   CREATE TABLE IF NOT EXISTS ods_app_detail (
        uid INT,
        country TEXT,
        prov TEXT,
        city TEXT,
        channel TEXT,
        operator TEXT,
        brand TEXT,
        ip TEXT,
        click_time TEXT,
        year TEXT,
        month TEXT,
        day TEXT,
        ymd TEXT NOT NULL
   );
   CALL set_table_property('ods_app_detail', 'orientation', 'column');
   CALL set_table_property('ods_app_detail', 'bitmap_columns', 'country,prov,city,channel,operator,brand,ip,click_time, year, month, day, ymd');
   -- Atur distribution_key berdasarkan kebutuhan kueri untuk efek sharding optimal.
   CALL set_table_property('ods_app_detail', 'distribution_key', 'uid');
   -- Untuk bidang dengan tanggal-waktu lengkap yang digunakan dalam filter WHERE, disarankan mengaturnya sebagai clustering_key dan event_time_column.
   CALL set_table_property('ods_app_detail', 'clustering_key', 'ymd');
   CALL set_table_property('ods_app_detail', 'event_time_column', 'ymd');
   COMMIT;

2. Hitung UV menggunakan Incremental Dynamic Table.

   CREATE DYNAMIC TABLE ads_uv_dt
     WITH (
       freshness = '5 minutes', 
       auto_refresh_mode = 'incremental') 
     AS 
   SELECT  
    RB_BUILD_AGG(uid),
    country,
    prov,
    city,
    ymd,
   COUNT(1)
   FROM    ods_app_detail
   WHERE ymd >= '20231201' AND ymd <='20240502'
   GROUP BY country,prov,city,ymd;

3. Kueri UV untuk hari tertentu.

   SELECT  
   RB_CARDINALITY(RB_OR_AGG(rb_uid)) AS uv,
     country,
     prov,
     city,
     SUM(pv) AS pv
   FROM    ads_uv_dt
   WHERE   ymd = '20240329'
   GROUP BY country,prov,city;

Contoh 1: Buat dynamic table full-refresh yang mulai otomatis

Sebelum melanjutkan, impor dataset publik tpch_10g ke Hologres dengan mengikuti panduan di Impor dataset publik dengan beberapa klik.

--Buat Skema "test"
CREATE SCHEMA test;

--Buat dynamic table single-table full refresh, mulai segera dan merefresh setiap jam.
CREATE DYNAMIC TABLE test.thch_q1_full
  WITH (
    refresh_mode='full',
    auto_refresh_enable='true',
    full_auto_refresh_interval='1 hours',
    full_guc_hg_computing_resource='serverless',
    full_guc_hg_experimental_serverless_computing_required_cores='32'
       ) 
AS
  SELECT
        l_returnflag,
        l_linestatus,
        SUM(l_quantity) AS sum_qty,
        SUM(l_extendedprice) AS sum_base_price,
        SUM(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
        SUM(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
        AVG(l_quantity) AS avg_qty,
        AVG(l_extendedprice) AS avg_price,
        AVG(l_discount) AS avg_disc,
        COUNT(*) AS count_order
FROM
        hologres_dataset_tpch_10g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus;

Contoh 2: Buat dynamic table inkremental dengan waktu mulai

Sebelum melanjutkan, impor dataset publik tpch_10g ke Hologres dengan mengikuti panduan di Impor dataset publik dengan beberapa klik.

Contoh:

--Aktifkan binlog untuk tabel dasar:
BEGIN;
CALL set_table_property('hologres_dataset_tpch_10g.lineitem', 'binlog.level', 'replica');
COMMIT;

--Buat dynamic table single-table incremental refresh, menentukan waktu mulai dan interval refresh 3 menit.
CREATE DYNAMIC TABLE public.tpch_q1_incremental 
WITH (
refresh_mode='incremental',
auto_refresh_enable='true',
incremental_auto_refresh_schd_start_time='2024-09-15 23:50:0',
incremental_auto_refresh_interval='3 minutes',
incremental_guc_hg_computing_resource='serverless',
incremental_guc_hg_experimental_serverless_computing_required_cores='30'
) AS SELECT
        l_returnflag,
        l_linestatus,
        COUNT(*) AS count_order
FROM
        hologres_dataset_tpch_10g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus
;

Contoh 3: Buat dynamic table full-refresh multi-join

--Buat dynamic table dengan kueri join multi-tabel, menggunakan mode full refresh setiap 3 jam.
CREATE DYNAMIC TABLE dt_q_full
  WITH (
    refresh_mode='full',
    auto_refresh_enable='true',
    full_auto_refresh_schd_start_time='immediate',
    full_auto_refresh_interval='3 hours',
    full_guc_hg_computing_resource='serverless',
    full_guc_hg_experimental_serverless_computing_required_cores='64'
  ) 
AS 
SELECT
        o_orderpriority,
        COUNT(*) AS order_count
FROM
        hologres_dataset_tpch_10g.orders
WHERE
        o_orderdate >= DATE '1996-07-01'
        AND o_orderdate < DATE '1996-07-01' + INTERVAL '3' MONTH
        AND EXISTS (
                SELECT
                        *
                FROM
                        hologres_dataset_tpch_10g.lineitem
                WHERE
                        l_orderkey = o_orderkey
                        AND l_commitdate < l_receiptdate
        )
GROUP BY
        o_orderpriority;

Contoh 4: Buat dynamic table inkremental join dimensi

Contoh:

Semantik JOIN tabel dimensi adalah bahwa setiap catatan hanya digabung dengan versi terbaru data tabel dimensi pada waktu tersebut, yaitu JOIN terjadi pada waktu pemrosesan. Jika data tabel dimensi berubah (tambah, perbarui, atau hapus) setelah JOIN, data yang sudah digabung tidak diperbarui. Contoh SQL:

--Tabel detail
BEGIN;
CREATE TABLE public.sale_detail(
        app_id TEXT,
        uid TEXT,
        product TEXT,
        gmv BIGINT,
        order_time TIMESTAMPTZ
);
--Aktifkan binlog untuk tabel dasar; tabel dimensi tidak memerlukannya.
CALL set_table_property('public.sale_detail', 'binlog.level', 'replica');
COMMIT;

--Tabel properti
CREATE TABLE public.user_info(
        uid TEXT,
        province TEXT,
        city TEXT
);

CREATE DYNAMIC TABLE public.dt_sales_incremental
  WITH (
    refresh_mode='incremental',
    auto_refresh_enable='true',
    incremental_auto_refresh_schd_start_time='2024-09-15 00:00:00',
    incremental_auto_refresh_interval='5 minutes',
    incremental_guc_hg_computing_resource='serverless',
    incremental_guc_hg_experimental_serverless_computing_required_cores='128') 
AS 
SELECT 
    sale_detail.app_id,
    sale_detail.uid,
    product,
    SUM(sale_detail.gmv) AS sum_gmv,
    sale_detail.order_time,
    user_info.province,
    user_info.city 
FROM public.sale_detail 
INNER JOIN public.user_info  FOR SYSTEM_TIME AS OF PROCTIME()
ON sale_detail.uid =user_info.uid
GROUP BY sale_detail.app_id,sale_detail.uid,sale_detail.product,sale_detail.order_time,user_info.province,user_info.city;

Contoh 5: Buat dynamic table partisi

Untuk dasbor transaksi real-time, seringkali diperlukan tampilan near-real-time data saat ini dan koreksi data historis. Hal ini dapat dicapai dengan kombinasi `Dynamic Table` incremental dan full refresh. Pendekatannya sebagai berikut:

1. Buat tabel dasar partisi di mana partisi terbaru ditulis secara real-time/near-real-time, dan partisi historis kadang-kadang dikoreksi.

2. Buat `Dynamic Table` sebagai tabel induk partisi. Gunakan incremental refresh untuk partisi terbaru untuk memenuhi kebutuhan analisis near-real-time.

3. Alihkan partisi historis ke mode full refresh. Jika partisi historis tabel sumber telah dikoreksi, partisi `Dynamic Table` juga dapat diisi ulang menggunakan full refresh, sebaiknya dengan Serverless untuk mempercepatnya.

Contoh:

1. Siapkan tabel dasar dan data.

   Tabel dasar adalah tabel partisi, dengan partisi terbaru menerima data real-time.

   -- Buat tabel sumber partisi
   CREATE TABLE base_sales(
     uid INT,
     opreate_time TIMESTAMPTZ,
     amount FLOAT,
     tt TEXT NOT NULL,
     ds TEXT,
     PRIMARY KEY(ds)
   ) PARTITION BY LIST (ds) ;
   
   --Partisi historis
   CREATE TABLE base_sales_20240615 PARTITION OF base_sales FOR VALUES IN ('20240615');
   INSERT INTO base_sales_20240615 VALUES (2,'2024-06-15 16:18:25.387466+08','111','2','20240615');
   
   --Partisi terbaru, biasanya untuk penulisan real-time
   CREATE TABLE base_sales_20240616 PARTITION OF base_sales FOR VALUES IN ('20240616');
   INSERT INTO base_sales_20240616 VALUES (1,'2024-06-16 16:08:25.387466+08','2','1','20240616');

2. Buat tabel induk `Dynamic Table` partisi, hanya mendefinisikan kueri tanpa mode refresh.

   --Buat ekstensi
   CREATE EXTENSION roaringbitmap;
   
   CREATE DYNAMIC TABLE partition_dt_base_sales
   PARTITION BY LIST (ds)
   as
   SELECT  
    public.RB_BUILD_AGG(uid),
    opreate_time,
    amount,
    tt,
    ds,
   COUNT(1)
   FROM    base_sales
   GROUP BY opreate_time ,amount,tt,ds;

3. Buat sub-tabel dan atur mode refresh-nya.

   Anda dapat membuat sub-partisi `Dynamic Table` secara manual atau dinamis menggunakan DataWorks. Atur partisi terbaru ke incremental refresh dan partisi historis ke full refresh.

   -- Aktifkan Binlog untuk tabel dasar
   ALTER TABLE base_sales SET (binlog_level = replica);
   
   -- Asumsikan sub-partisi Dynamic Table historis adalah sebagai berikut:
   CREATE DYNAMIC TABLE partition_dt_base_sales_20240615 PARTITION OF partition_dt_base_sales FOR VALUES IN ('20240615')
     WITH (
       refresh_mode='incremental',
       auto_refresh_enable='true',
       incremental_auto_refresh_schd_start_time='immediate',
       incremental_auto_refresh_interval='30 minutes'
          );
   
   -- Buat sub-partisi Dynamic Table baru, atur mode refresh-nya ke incremental, mulai segera, refresh setiap 30 menit, dan gunakan sumber daya instans.
   CREATE DYNAMIC TABLE partition_dt_base_sales_20240616 PARTITION OF partition_dt_base_sales FOR VALUES IN ('20240616')
     WITH (
       refresh_mode='incremental',
       auto_refresh_enable='true',
       incremental_auto_refresh_schd_start_time='immediate',
       incremental_auto_refresh_interval='30 minutes'
          );
   
   --Alihkan partisi historis ke mode full refresh
   ALTER DYNAMIC TABLE partition_dt_base_sales_20240615 SET (refresh_mode = 'full');
   --Jika data partisi historis perlu dikoreksi, jalankan refresh, sebaiknya dengan serverless.
   SET hg_computing_resource = 'serverless';
   REFRESH DYNAMIC TABLE partition_dt_base_sales_20240615;