Hologres:動的テーブルの作成

例 1：通常の増分動的テーブルの作成

進む前に、「数回のクリックでパブリックデータセットをインポート」のガイドに従って、tpch_10g パブリックデータセットを Hologres にインポートしてください。

-- 3 分ごとに更新する増分動的テーブルを作成します。
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;

例 2：ストリーム-ストリーム JOIN から増分動的テーブルを作成する

進む前に、「数回のクリックでパブリックデータセットをインポート」のガイドに従って、tpch_10g パブリックデータセットを Hologres にインポートしてください。

-- ストリーム-ストリーム JOIN から増分動的テーブルを作成します。
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;

例 3：自動更新動的テーブルの作成

更新モードを auto に設定します。エンジンは増分更新を優先し、サポートされていない場合は完全更新にフォールバックします。

進む前に、「数回のクリックでパブリックデータセットをインポート」のガイドに従って、tpch_10g パブリックデータセットを Hologres にインポートしてください。

-- 更新モードをインテリジェントに決定する自動更新動的テーブルを作成します。この例では、増分更新が結果となります。
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;

例 4：論理パーティション動的テーブルの作成

リアルタイムのトランザクションダッシュボードでは、現在のデータのニアリアルタイム表示と既存データの修正の両方が必要になることが多く、統合されたリアルタイムおよびオフライン分析ソリューションが必要です (ビジネスとデータの認識)。通常、このシナリオでは `動的テーブル` の論理パーティションを使用します。アプローチは次のとおりです：

• ベーステーブルは日次でパーティション化されます。最新のパーティションは Flink によってリアルタイム/ニアリアルタイムで書き込まれ、既存のパーティションは MaxCompute から書き込まれます。

• 動的テーブルは論理パーティションテーブルとして作成されます。最新の 2 つのパーティションはアクティブであり、ニアリアルタイムのデータ分析のために増分更新されます。

• 既存のパーティションは非アクティブで、完全更新を使用します。ベーステーブルの既存のパーティションが修正またはバックフィルされた場合、完全更新を使用して更新できます。

この例では、GitHub のパブリックデータセットを使用します。

1. ベーステーブルを準備します。

   Flink を使用して最新のデータをベーステーブルに書き込みます。詳細な手順については、「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 'イベント ID';
   COMMENT ON COLUMN public.gh_realtime_data.actor_id IS 'イベントアクター ID';
   COMMENT ON COLUMN public.gh_realtime_data.actor_login IS 'イベントアクターログイン名';
   COMMENT ON COLUMN public.gh_realtime_data.repo_id IS 'リポジトリ ID';
   COMMENT ON COLUMN public.gh_realtime_data.repo_name IS 'リポジトリ名';
   COMMENT ON COLUMN public.gh_realtime_data.org_id IS 'リポジトリ組織 ID';
   COMMENT ON COLUMN public.gh_realtime_data.org_login IS 'リポジトリ組織名';
   COMMENT ON COLUMN public.gh_realtime_data.type IS 'イベントタイプ';
   COMMENT ON COLUMN public.gh_realtime_data.created_at IS 'イベント時間';
   COMMENT ON COLUMN public.gh_realtime_data.action IS 'イベントアクション';
   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 番号';
   COMMENT ON COLUMN public.gh_realtime_data.comment_id IS 'コメント ID';
   COMMENT ON COLUMN public.gh_realtime_data.commit_id IS 'コミット ID';
   COMMENT ON COLUMN public.gh_realtime_data.member_id IS 'メンバー ID';
   COMMENT ON COLUMN public.gh_realtime_data.rev_or_push_or_rel_id IS 'レビュー/プッシュ/リリース ID';
   COMMENT ON COLUMN public.gh_realtime_data.ref IS '作成/削除されたリソースの名前';
   COMMENT ON COLUMN public.gh_realtime_data.ref_type IS '作成/削除されたリソースのタイプ';
   COMMENT ON COLUMN public.gh_realtime_data.state IS 'Issue/pull_request/pull_request_review の状態';
   COMMENT ON COLUMN public.gh_realtime_data.author_association IS 'アクターとリポジトリの関係';
   COMMENT ON COLUMN public.gh_realtime_data.language IS 'プログラミング言語';
   COMMENT ON COLUMN public.gh_realtime_data.merged IS 'マージされたかどうか';
   COMMENT ON COLUMN public.gh_realtime_data.merged_at IS 'マージ時間';
   COMMENT ON COLUMN public.gh_realtime_data.additions IS '追加された行数';
   COMMENT ON COLUMN public.gh_realtime_data.deletions IS '削除された行数';
   COMMENT ON COLUMN public.gh_realtime_data.changed_files IS 'プルリクエストで変更されたファイル数';
   COMMENT ON COLUMN public.gh_realtime_data.push_size IS 'プッシュ数';
   COMMENT ON COLUMN public.gh_realtime_data.push_distinct_size IS '個別プッシュ数';
   COMMENT ON COLUMN public.gh_realtime_data.hr IS 'イベントの時間、例：00:23 の場合は 00';
   COMMENT ON COLUMN public.gh_realtime_data.month IS 'イベントの月、例：2015年10月の場合は 2015-10';
   COMMENT ON COLUMN public.gh_realtime_data.year IS 'イベントの年、例：2015';
   COMMENT ON COLUMN public.gh_realtime_data.ds IS 'イベントの日付、ds=yyyy-mm-dd';
   
   COMMIT;

2. 論理パーティション動的テーブルを作成します。

   CREATE  DYNAMIC TABLE ads_dt_github_event
   LOGICAL PARTITION BY LIST(ds)
   WITH (
     -- 動的テーブルのプロパティ
     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. 動的テーブルをクエリします。

   SELECT * FROM ads_dt_github_event ;

4. 既存のパーティションをバックフィルします。

   ベーステーブルの既存データが変更された場合 (例：'2025-04-01' のデータ)、動的テーブルを更新する必要がある場合は、既存のパーティションを完全更新モードに設定し、更新をトリガーします。サーバーレスコンピューティングリソースを使用することが望ましいです。

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

例 5：増分動的テーブルで UV を計算する

Hologres V3.1 から、増分動的テーブルは RB_BUILD_AGG 関数をサポートし、UV 数などの計算が可能です。事前集計と比較して、以下の利点があります：

• より高速なパフォーマンス：増分データのみを計算します。

• 低コスト：データ量とリソース使用量が削減され、より長期間にわたる計算が可能になります。

例：

1. ユーザー詳細テーブルを準備します。

   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');
   -- 最適なシャーディング効果を得るために、クエリのニーズに基づいて distribution_key を設定します。
   CALL set_table_property('ods_app_detail', 'distribution_key', 'uid');
   -- WHERE フィルターで使用される完全な日時を持つフィールドについては、clustering_key と 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. 増分動的テーブルを使用して UV を計算します。

   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. 特定の日付の UV をクエリします。

   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;

例 1：自動開始の完全更新動的テーブルの作成

進む前に、「数回のクリックでパブリックデータセットをインポート」のガイドに従って、tpch_10g パブリックデータセットを Hologres にインポートしてください。

--「test」スキーマを作成
CREATE SCHEMA test;

--単一テーブルの完全更新動的テーブルを作成し、即時開始して 1 時間ごとに更新します。
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;

例 2：開始時刻を指定した増分動的テーブルの作成

進む前に、「数回のクリックでパブリックデータセットをインポート」のガイドに従って、tpch_10g パブリックデータセットを Hologres にインポートしてください。

例：

--ベーステーブルの binlog を有効にする：
BEGIN;
CALL set_table_property('hologres_dataset_tpch_10g.lineitem', 'binlog.level', 'replica');
COMMIT;

--単一テーブルの増分更新動的テーブルを作成し、開始時刻と 3 分の更新間隔を指定します。
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
;

例 3：複数 JOIN の完全更新動的テーブルの作成

--複数テーブル JOIN クエリを持つ動的テーブルを作成し、3 時間ごとに完全更新モードを使用します。
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;

例 4：ディメンション JOIN の増分動的テーブルの作成

例：

ディメンションテーブル JOIN のセマンティクスは、各レコードがその時点でのディメンションテーブルデータの最新バージョンとのみ結合される、つまり JOIN が処理時に行われるということです。JOIN 後にディメンションテーブルのデータが変更 (追加、更新、または削除) されても、すでに結合されたデータは更新されません。SQL の例：

--詳細テーブル
BEGIN;
CREATE TABLE public.sale_detail(
        app_id TEXT,
        uid TEXT,
        product TEXT,
        gmv BIGINT,
        order_time TIMESTAMPTZ
);
--ベーステーブルの binlog を有効にします。ディメンションテーブルには不要です。
CALL set_table_property('public.sale_detail', 'binlog.level', 'replica');
COMMIT;

--プロパティテーブル
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;

例 5：パーティション動的テーブルの作成

リアルタイムのトランザクションダッシュボードでは、現在のデータのニアリアルタイム表示と既存データの修正の両方が必要になることがよくあります。これは、`動的テーブル` の増分更新と完全更新を組み合わせて実現できます。アプローチは次のとおりです：

1. パーティション化されたベーステーブルを作成し、最新のパーティションはリアルタイム/ニアリアルタイムで書き込まれ、既存のパーティションは時々修正されます。

2. `動的テーブル` をパーティション化された親テーブルとして作成します。最新のパーティションには増分更新を使用して、ニアリアルタイムの分析ニーズに対応します。

3. 既存のパーティションを完全更新モードに切り替えます。ソーステーブルの既存のパーティションが修正された場合、`動的テーブル` のパーティションも完全更新を使用してバックフィルできます。高速化のためにサーバーレスを使用することが望ましいです。

例：

1. ベーステーブルとデータを準備します。

   ベーステーブルはパーティションテーブルで、最新のパーティションがリアルタイムデータを受け取ります。

   -- パーティション化されたソーステーブルを作成
   CREATE TABLE base_sales(
     uid INT,
     opreate_time TIMESTAMPTZ,
     amount FLOAT,
     tt TEXT NOT NULL,
     ds TEXT,
     PRIMARY KEY(ds)
   ) PARTITION BY LIST (ds) ;
   
   --既存のパーティション
   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');
   
   --最新のパーティション、通常はリアルタイム書き込み用
   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. パーティション化された `動的テーブル` の親テーブルを作成し、更新モードなしでクエリのみを定義します。

   --拡張機能を作成
   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. サブテーブルを作成し、その更新モードを設定します。

   `動的テーブル` のサブパーティションは手動で作成するか、DataWorks を使用して動的に作成できます。最新のパーティションを増分更新に設定し、既存のパーティションを完全更新に設定します。

   -- ベーステーブルの Binlog を有効にする
   ALTER TABLE base_sales SET (binlog_level = replica);
   
   -- 既存の動的テーブルのサブパーティションは次のようになると仮定します：
   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'
          );
   
   -- 新しい動的テーブルのサブパーティションを作成し、その更新モードを増分に設定し、即時開始し、30 分ごとに更新し、インスタンスリソースを使用します。
   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'
          );
   
   --既存のパーティションを完全更新モードに切り替える
   ALTER DYNAMIC TABLE partition_dt_base_sales_20240615 SET (refresh_mode = 'full');
   --既存のパーティションデータの修正が必要な場合は、更新を実行します。サーバーレスを使用することが望ましいです。
   SET hg_computing_resource = 'serverless';
   REFRESH DYNAMIC TABLE partition_dt_base_sales_20240615;