Hologres:Criar dynamic table

Exemplo 1: Criar uma Incremental Dynamic Table regular

Antes de prosseguir, importe o conjunto de dados público tpch_10g para o Hologres seguindo o guia em 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;

Exemplo 2: Criar uma Incremental Dynamic Table a partir de stream-stream joins

Antes de prosseguir, importe o conjunto de dados público tpch_10g para o Hologres seguindo o guia em 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;

Exemplo 3: Criar uma Dynamic Table com atualização automática

Defina o modo de atualização como auto. O mecanismo prioriza a atualização incremental e reverte para a atualização completa se não houver suporte.

Antes de prosseguir, importe o conjunto de dados público tpch_10g para o Hologres seguindo o guia em 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;

Exemplo 4: Criar uma Dynamic Table com particionamento lógico

Para painéis de transações em tempo real, frequentemente há necessidade tanto de visualização quase em tempo real dos dados atuais quanto de correção de dados históricos, o que exige uma solução integrada de análise em tempo real e offline (Business and data cognition). Normalmente utilizamos partições lógicas de Dynamic Table para esse cenário. A abordagem é a seguinte:

• A tabela base é particionada por dia. A partição mais recente recebe escrita do Flink em tempo real/quase em tempo real, enquanto as partições históricas recebem dados do MaxCompute.

• A Dynamic Table é criada como uma tabela com particionamento lógico. As duas partições mais recentes estão ativas e são atualizadas incrementalmente para análises de dados quase em tempo real.

• As partições históricas estão inativas e usam atualização completa. Se as partições históricas da tabela base tiverem sido corrigidas ou preenchidas (backfill), elas poderão ser atualizadas usando uma atualização completa.

Este exemplo utiliza um conjunto de dados público do GitHub.

1. Prepare a tabela base.

   Use o Flink para gravar os dados mais recentes na tabela base. Para etapas detalhadas, 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. Crie uma Dynamic Table com particionamento lógico.

   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 a Dynamic Table.

   SELECT * FROM ads_dt_github_event ;

4. Faça backfill de uma partição histórica.

   Se os dados históricos na tabela base mudarem (ex.: dados de '2025-04-01') e a Dynamic Table precisar ser atualizada, defina a partição histórica para o modo de atualização completa e acione uma atualização, preferencialmente com recursos de Serverless Computing.

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

Exemplo 5: Calcular UVs com Incremental Dynamic Table

A partir do Hologres V3.1, uma Incremental Dynamic Table suporta a função RB_BUILD_AGG para cálculos como o número de UVs. Comparado à pré-agregação, isso oferece:

• Desempenho mais rápido: Computa apenas dados incrementais.

• Custo menor: Volume de dados e uso de recursos reduzidos, permitindo cálculos em períodos mais longos.

Exemplo:

1. Prepare uma tabela de detalhes de usuário.

   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 UV usando uma 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. Consulte UV para um dia 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;

Exemplo 1: Criar uma dynamic table de atualização completa com início automático

Antes de prosseguir, importe o conjunto de dados público tpch_10g para o Hologres seguindo o guia em 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;

Exemplo 2: Criar uma dynamic table incremental com horário de início

Antes de prosseguir, importe o conjunto de dados público tpch_10g para o Hologres seguindo o guia em Import public datasets with a few clicks.

Exemplo:

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

Exemplo 3: Criar uma dynamic table de atualização completa com múltiplos joins

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

Exemplo 4: Criar uma dynamic table incremental com join de dimensão

Exemplo:

A semântica de um JOIN com tabela de dimensão é que cada registro é unido apenas à versão mais recente dos dados da tabela de dimensão naquele momento, ou seja, o JOIN ocorre no tempo de processamento. Se os dados da tabela de dimensão mudarem (adição, atualização ou exclusão) após o JOIN, os dados já unidos não serão atualizados. Exemplo 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;

Exemplo 5: Criar uma dynamic table particionada

Para painéis de transações em tempo real, frequentemente há necessidade tanto de visualização quase em tempo real dos dados atuais quanto de correção de dados históricos. Isso pode ser alcançado usando uma combinação de atualização incremental e completa da Dynamic Table. A abordagem é a seguinte:

1. Crie uma tabela base particionada onde a partição mais recente recebe escrita em tempo real/quase em tempo real, e as partições históricas são ocasionalmente corrigidas.

2. Crie uma Dynamic Table como tabela pai particionada. Use atualização incremental para a partição mais recente para atender às necessidades de análise quase em tempo real.

3. Alterne as partições históricas para o modo de atualização completa. Se as partições históricas da tabela de origem tiverem sido corrigidas, as partições da Dynamic Table também poderão sofrer backfill usando atualização completa, preferencialmente com Serverless para acelerar o processo.

Exemplo:

1. Prepare a tabela base e os dados.

   A tabela base é uma tabela particionada, com a partição mais recente recebendo dados em tempo 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. Crie uma tabela pai Dynamic Table particionada, definindo apenas a consulta sem um modo de atualização.

   --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. Crie sub-tabelas e defina seus modos de atualização.

   Você pode criar sub-partições da Dynamic Table manualmente ou dinamicamente usando DataWorks. Defina a partição mais recente para atualização incremental e as partições históricas para atualização 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;