Hologres:Create dynamic table

Ejemplo 1: Crear una Dynamic Table incremental estándar

Antes de continuar, importe el conjunto de datos público tpch_10g a Hologres siguiendo la guía de Import public datasets with a few clicks.

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

Ejemplo 2: Crear una Dynamic Table incremental a partir de joins entre flujos

Antes de continuar, importe el conjunto de datos público tpch_10g a Hologres siguiendo la guía de Import public datasets with a few clicks.

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

Ejemplo 3: Crear una Dynamic Table con actualización automática

Establezca el modo de actualización en automático. El motor prioriza la actualización incremental y recurre a la completa si la primera no es compatible.

Antes de continuar, importe el conjunto de datos público tpch_10g a Hologres siguiendo la guía de Import public datasets with a few clicks.

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

Ejemplo 4: Crear una Dynamic Table con particiones lógicas

En un panel de transacciones en tiempo real, suele ser necesario consultar los datos actuales casi en tiempo real y corregir los datos históricos. Esto requiere una solución que integre el análisis en tiempo real y por lotes (Business and data cognition). Para este escenario se utilizan las particiones lógicas de Dynamic Table. El enfoque es el siguiente:

• La tabla base está particionada por día. Flink escribe la partición más reciente en tiempo real o casi en tiempo real, mientras que las particiones históricas se escriben desde MaxCompute.

• La Dynamic Table se crea como una tabla con particiones lógicas. Las dos particiones más recientes están activas y se actualizan de forma incremental para el análisis de datos casi en tiempo real.

• Las particiones históricas están inactivas y usan actualización completa. Si las particiones históricas de la tabla base se corrigieron o rellenaron, se pueden actualizar mediante una actualización completa.

Este ejemplo utiliza un conjunto de datos público de GitHub.

1. Prepare la tabla base.

   Use Flink para escribir los datos más recientes en la tabla base. Para conocer los pasos detallados, consulte Unified batch and real-time analytics on GitHub events.

   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. Cree una Dynamic Table con particiones lógicas.

   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. Consulte la Dynamic Table.

   SELECT * FROM ads_dt_github_event ;

4. Rellene una partición histórica.

   Si los datos históricos de la tabla base cambian (por ejemplo, los datos del '2025-04-01') y necesita actualizar la Dynamic Table, configure la partición histórica en modo de actualización completa y active la actualización. Se recomienda usar recursos de Serverless Computing.

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

Ejemplo 5: Calcular UV con una Dynamic Table incremental

A partir de Hologres V3.1, una Dynamic Table incremental es compatible con la función RB_BUILD_AGG para cálculos como el número de UV. En comparación con la preagregación, esto ofrece lo siguiente:

• Mayor velocidad: solo procesa los datos incrementales.

• Menor costo: reduce el volumen de datos y el uso de recursos, lo que permite ejecutar cálculos sobre periodos más largos.

Ejemplo:

1. Prepare una tabla de detalles de usuarios.

   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. Calcule los UV mediante una Dynamic Table incremental.

   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. Consulte los UV de un día específico.

   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;

Ejemplo 1: Crear una tabla dinámica de actualización completa con inicio automático

Antes de continuar, importe el conjunto de datos público tpch_10g a Hologres. Siga la guía disponible en Import public datasets with a few clicks.

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

Ejemplo 2: Crear una tabla dinámica incremental con hora de inicio

Antes de continuar, importe el conjunto de datos público tpch_10g a Hologres. Siga la guía disponible en Import public datasets with a few clicks.

A continuación se muestra un ejemplo:

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

Ejemplo 3: Crear una tabla dinámica de actualización completa con múltiples uniones

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

Ejemplo 4: Crear una tabla dinámica incremental con unión de dimensiones

A continuación se muestra un ejemplo:

La semántica de un JOIN con una tabla de dimensiones indica que cada registro se une únicamente con la versión más reciente de los datos de dicha tabla en ese momento. Es decir, el JOIN ocurre en el tiempo de procesamiento. Si los datos de la tabla de dimensiones cambian después del JOIN (ya sea por inserción, actualización o eliminación), los datos ya unidos no se actualizan. Ejemplo de 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;

Ejemplo 5: Crear una tabla dinámica particionada

En un panel de transacciones en tiempo real, suele ser necesario consultar los datos actuales casi en tiempo real y corregir los datos históricos. Esto se logra al combinar la actualización incremental y completa de Dynamic Table. El enfoque es el siguiente:

1. Cree una tabla base particionada donde la partición más reciente reciba escrituras en tiempo real o casi en tiempo real, y las particiones históricas se corrijan de manera ocasional.

2. Cree una Dynamic Table como tabla principal particionada. Use la actualización incremental para la partición más reciente y así cubrir las necesidades de análisis casi en tiempo real.

3. Cambie las particiones históricas al modo de actualización completa. Si las particiones históricas de la tabla de origen se corrigieron, también puede rellenar las particiones de la Dynamic Table mediante una actualización completa. Se recomienda usar recursos serverless para acelerar el proceso.

Ejemplo:

1. Prepare la tabla base y los datos.

   La tabla base es una tabla particionada cuya partición más reciente recibe datos en tiempo real.

   -- 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. Cree la tabla principal de Dynamic Table particionada. Defina únicamente la consulta, sin especificar un modo de actualización.

   --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. Cree las subtablas y establezca sus modos de actualización.

   Puede crear las subparticiones de Dynamic Table de forma manual o dinámica mediante DataWorks. Configure la partición más reciente con actualización incremental y las particiones históricas con actualización completa.

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