Hologres:Create dynamic table

Exemple 1 : Créer une table dynamique incrémentielle standard

Avant de commencer, importez le jeu de données public tpch_10g dans Hologres en suivant le guide disponible à la rubrique Importer des jeux de données publics en quelques clics.

-- Create an Incremental Dynamic Table that refreshes every 3 minutes.
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;

Exemple 2 : Créer une table dynamique incrémentielle à partir de jointures flux-flux

Avant de commencer, importez le jeu de données public tpch_10g dans Hologres en suivant le guide disponible à la rubrique Importer des jeux de données publics en quelques clics.

-- Create an Incremental Dynamic Table from stream-stream joins.
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;

Exemple 3 : Créer une table dynamique avec actualisation automatique

Définissez le mode d'actualisation sur auto. Le moteur privilégie l'actualisation incrémentielle et bascule vers une actualisation complète si celle-ci n'est pas prise en charge.

Avant de commencer, importez le jeu de données public tpch_10g dans Hologres en suivant le guide disponible à la rubrique Importer des jeux de données publics en quelques clics.

-- Create an Auto-refresh Dynamic Table that intelligent decides the refresh mode. In this example, incremental refresh is the result.
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;

Exemple 4 : Créer une table dynamique avec partitionnement logique

Pour un tableau de bord transactionnel en temps réel, il est souvent nécessaire de consulter les données actuelles en quasi-temps réel tout en corrigeant les données historiques. Cela requiert une solution intégrée d'analyse en temps réel et hors ligne (Connaissance métier et des données). Nous utilisons généralement des partitions logiques de Dynamic Table pour ce scénario. La démarche est la suivante :

• La table de base est partitionnée par jour. La dernière partition est alimentée par Flink en temps réel ou en quasi-temps réel, tandis que les partitions historiques sont écrites depuis MaxCompute.

• La table dynamique est créée sous forme de table à partitions logiques. Les deux dernières partitions sont actives et actualisées de manière incrémentielle pour l'analyse des données en quasi-temps réel.

• Les partitions historiques sont inactives et utilisent une actualisation complète. Si les partitions historiques de la table de base ont été corrigées ou réinjectées, elles peuvent être actualisées via une actualisation complète.

Cet exemple utilise un jeu de données public provenant de GitHub.

1. Préparez la table de base.

   Utilisez Flink pour écrire les dernières données dans la table de base. Pour obtenir la procédure détaillée, consultez la rubrique Analyse unifiée par lots et en temps réel des événements 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. Créez une table dynamique avec partitionnement logique.

   CREATE  DYNAMIC TABLE ads_dt_github_event
   LOGICAL PARTITION BY LIST(ds)
   WITH (
     -- Dynamic table properties
     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. Interrogez la table dynamique.

   SELECT * FROM ads_dt_github_event ;

4. Réinjectez une partition historique.

   Si les données historiques de la table de base changent (par exemple, les données du « 2025-04-01 ») et que la table dynamique doit être mise à jour, définissez la partition historique sur le mode d'actualisation complète et déclenchez une actualisation, de préférence avec des ressources Serverless Computing.

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

Exemple 5 : Calculer les UV avec une table dynamique incrémentielle

À partir de Hologres V3.1, une table dynamique incrémentielle prend en charge la fonction RB_BUILD_AGG pour des calculs tels que le nombre d'utilisateurs uniques (UV). Par rapport à la pré-agrégation, cette approche offre :

• De meilleures performances : seules les données incrémentielles sont traitées.

• Un coût réduit : le volume de données et l'utilisation des ressources sont diminués, permettant des calculs sur des périodes plus longues.

Exemple :

1. Préparez une table de détails utilisateur.

   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');
   -- Set distribution_key based on query needs for optimal sharding effect.
   CALL set_table_property('ods_app_detail', 'distribution_key', 'uid');
   -- For a field with full date-time used in WHERE filters, setting it as clustering_key and event_time_column is recommended.
   CALL set_table_property('ods_app_detail', 'clustering_key', 'ymd');
   CALL set_table_property('ods_app_detail', 'event_time_column', 'ymd');
   COMMIT;

2. Calculez les UV à l'aide d'une table dynamique incrémentielle.

   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. Consultez les UV pour une journée spécifique.

   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;

Exemple 1 : Créer une table dynamique à actualisation complète avec démarrage automatique

Avant de commencer, importez le jeu de données public tpch_10g dans Hologres en suivant le guide disponible à la rubrique Importer des jeux de données publics en quelques clics.

--Create "test" Schema
CREATE SCHEMA test;

--Create a single-table full refresh dynamic table, starting immediately and refreshing every hour.
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;

Exemple 2 : Créer une table dynamique incrémentielle avec une heure de début

Avant de commencer, importez le jeu de données public tpch_10g dans Hologres en suivant le guide disponible à la rubrique Importer des jeux de données publics en quelques clics.

Exemple :

--Enable binlog for the base table:
BEGIN;
CALL set_table_property('hologres_dataset_tpch_10g.lineitem', 'binlog.level', 'replica');
COMMIT;

--Create a single-table incremental refresh dynamic table, specifying a start time and a 3-minute refresh interval.
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
;

Exemple 3 : Créer une table dynamique à actualisation complète avec jointures multiples

--Create a dynamic table with a multi-table join query, using full refresh mode every 3 hours.
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;

Exemple 4 : Créer une table dynamique incrémentielle avec jointure de dimension

Exemple :

La sémantique d'une jointure avec une table de dimension implique que chaque enregistrement est joint uniquement avec la dernière version des données de la table de dimension à cet instant, c'est-à-dire que la jointure s'effectue au moment du traitement. Si les données de la table de dimension changent (ajout, mise à jour ou suppression) après la jointure, les données déjà jointes ne sont pas mises à jour. Exemple SQL :

--Detail table
BEGIN;
CREATE TABLE public.sale_detail(
        app_id TEXT,
        uid TEXT,
        product TEXT,
        gmv BIGINT,
        order_time TIMESTAMPTZ
);
--Enable binlog for the base table; dimension tables do not require it.
CALL set_table_property('public.sale_detail', 'binlog.level', 'replica');
COMMIT;

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

Exemple 5 : Créer une table dynamique partitionnée

Pour un tableau de bord transactionnel en temps réel, il est souvent nécessaire de consulter les données actuelles en quasi-temps réel tout en corrigeant les données historiques. Cela peut être réalisé en combinant l'actualisation incrémentielle et complète de Dynamic Table. La démarche est la suivante :

1. Créez une table de base partitionnée où la dernière partition est écrite en temps réel ou en quasi-temps réel, et les partitions historiques sont occasionnellement corrigées.

2. Créez une Dynamic Table en tant que table parente partitionnée. Utilisez l'actualisation incrémentielle pour la dernière partition afin de répondre aux besoins d'analyse en quasi-temps réel.

3. Basculez les partitions historiques en mode d'actualisation complète. Si les partitions historiques de la table source ont été corrigées, les partitions de la Dynamic Table peuvent également être réinjectées via une actualisation complète, de préférence avec Serverless pour accélérer le processus.

Exemple :

1. Préparez la table de base et les données.

   La table de base est une table partitionnée, la dernière partition recevant les données en temps réel.

   -- Create a partitioned source table
   CREATE TABLE base_sales(
     uid INT,
     opreate_time TIMESTAMPTZ,
     amount FLOAT,
     tt TEXT NOT NULL,
     ds TEXT,
     PRIMARY KEY(ds)
   ) PARTITION BY LIST (ds) ;
   
   --Historical partition
   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');
   
   --Latest partition, typically for real-time writes
   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. Créez la table parente Dynamic Table partitionnée, en définissant uniquement la requête sans mode d'actualisation.

   --Create extension
   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. Créez les sous-tables et définissez leurs modes d'actualisation.

   Vous pouvez créer manuellement les sous-partitions de Dynamic Table ou les générer dynamiquement à l'aide de DataWorks. Définissez la dernière partition sur l'actualisation incrémentielle et les partitions historiques sur l'actualisation complète.

   -- Enable Binlog for the base table
   ALTER TABLE base_sales SET (binlog_level = replica);
   
   -- Assume the historical Dynamic Table sub-partition is as follows:
   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'
          );
   
   -- Create a new Dynamic Table sub-partition, set its refresh mode to incremental, start immediately, refresh every 30 minutes, and use instance resources.
   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'
          );
   
   --Switch the historical partition to full refresh mode
   ALTER DYNAMIC TABLE partition_dt_base_sales_20240615 SET (refresh_mode = 'full');
   --If historical partition data needs correction, execute a refresh, preferably with serverless.
   SET hg_computing_resource = 'serverless';
   REFRESH DYNAMIC TABLE partition_dt_base_sales_20240615;