Realtime Compute for Apache Flink:Hologres リアルタイムデータウェアハウスの構築

RDS for MySQL インスタンスの作成とデータの準備

1. ApsaraDB RDS for MySQL インスタンスを作成します。詳細については、「ApsaraDB RDS for MySQL インスタンスの作成」をご参照ください。

   ApsaraDB RDS for MySQL インスタンスは、Flink ワークスペースおよび Hologres インスタンスと同じ VPC 内にある必要があります。

2. データベースとアカウントを作成します。

   対象のインスタンスに、order_dw という名前のデータベースと、そのデータベースに対する読み取り/書き込み権限を持つ標準アカウントを作成します。詳細については、「データベースの作成」および「アカウントの作成」をご参照ください。

3. MySQL CDC データソースを準備します。

   1. インスタンスの詳細ページで、[Log On to Database] をクリックします。

   2. ログオンページで、作成したデータベースアカウントのユーザー名とパスワードを入力し、[Log On] をクリックします。

   3. ログオン後、order_dw データベースをダブルクリックして切り替えます。

   4. SQL コンソールで、次の DDL ステートメントを入力してビジネス テーブルを作成し、INSERT ステートメントを入力してデータを投入します。

      CREATE TABLE `orders` (
        order_id bigint not null primary key,
        user_id varchar(50) not null,
        shop_id bigint not null,
        product_id bigint not null,
        buy_fee numeric(20,2) not null,   
        create_time timestamp not null,
        update_time timestamp not null default now(),
        state int not null 
      );
      CREATE TABLE `orders_pay` (
        pay_id bigint not null primary key,
        order_id bigint not null,
        pay_platform int not null,
        create_time timestamp not null
      );
      CREATE TABLE `product_catalog` (
        product_id bigint not null primary key,
        catalog_name varchar(50) not null
      );
      -- データを準備します。
      INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
      INSERT INTO orders VALUES
      (100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
      (100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
      (100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
      (100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
      (100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
      (100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
      (100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
      INSERT INTO orders_pay VALUES
      (2001, 100001, 1, '2023-02-15 17:40:56'),
      (2002, 100002, 1, '2023-02-15 17:40:56'),
      (2003, 100003, 0, '2023-02-15 17:40:56'),
      (2004, 100004, 0, '2023-02-15 17:40:56'),
      (2005, 100005, 0, '2023-02-15 18:40:56'),
      (2006, 100006, 0, '2023-02-15 18:40:56'),
      (2007, 100007, 0, '2023-02-15 18:40:56');

4. [Execute] をクリックし、[Direct Execution] をクリックします。

Hologres インスタンスの作成およびコンピューティンググループ

1. Hologres 専用インスタンスを購入します。詳細については、「Hologres インスタンスの購入」をご参照ください。

   Hologres インスタンスは、ApsaraDB RDS for MySQL インスタンスと同じ VPC 内にある必要があります。読み書き分離による強力なリソース分離を体験するため、この例では [Virtual Warehouse] インスタンスタイプを使用し、[予約コンピューティングリソース] を 64 に設定して、追加のコンピューティンググループを作成できるようにします。

2. インスタンスにログオンした後、データベースを作成して権限を付与します。

   order_dw という名前のデータベース (シンプル権限モデルを有効化) を作成し、ユーザーに管理者権限を付与します。データベース管理と権限付与の詳細については、「データベースの管理」をご参照ください。

   説明

   • [User] ドロップダウンリストでアカウントが見つからない場合、そのアカウントはまだインスタンスに追加されていません。[User Management] ページに移動し、ユーザーをスーパーユーザーとして追加してください。

   • Hologres V2.0 以降では、バイナリロギング拡張がデフォルトで有効になっています。手動で有効にする必要はありません。

3. 新しいコンピューティンググループを作成します。

   異なるコンピューティンググループを使用してリソースを分離できます。初期コンピューティンググループ init_warehouse をデータ書き込みに、read_warehouse_1 コンピューティンググループをクエリのサービングに使用します。

   すべての予約コンピューティングリソースは、デフォルトで初期コンピューティンググループ init_warehouse に割り当てられます。新しいコンピューティンググループを作成する前に、まずそのリソースを削減する必要があります。詳細については、「新しいコンピューティンググループインスタンスの作成」をご参照ください。

   1. [Security Center] > [コンピューティンググループ管理] に移動し、インスタンス名を確認します。

   2. init_warehouse コンピューティンググループの行で、[Actions] 列の [Modify Configuration] をクリックします。割り当てられたリソースを削減し、[OK] をクリックします。

   3. [Create Compute Group] をクリックし、read_warehouse_1 という名前の新しいコンピューティンググループを作成して、[OK] をクリックします。

Flink ワークスペースとカタログの作成

1. Flink ワークスペースを作成します。詳細については、「Realtime Compute for Apache Flink のアクティベート」をご参照ください。

   Flink ワークスペースは、ApsaraDB RDS for MySQL インスタンスおよび Hologres インスタンスと同じ VPC 内にある必要があります。

2. Realtime Compute for Apache Flink コンソールにログオンし、ワークスペースの [Actions] 列で [Console] をクリックします。

3. カタログの作成とスクリプト実行用の実行環境として、セッションクラスターを作成します。詳細については、「ステップ 1：セッションクラスターの作成」をご参照ください。

4. Hologres カタログを作成します。

   [Development] > [Scripts] ページの [Scripts] タブで、次のコードをコピーし、プレースホルダーの値を置き換えてから、コードを選択して [Run] をクリックします。これにより、作成したセッションクラスターが実行環境として使用されます。

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = 'BASIC$flinktest',
     'password' = '${secret_values.holosecrect}',
     'dbname' = 'order_dw@init_warehouse', -- データベース名を指定し、init_warehouse コンピューティンググループに接続します。
     'binlog' = 'true', -- カタログ作成時に、ソース、ディメンション、結果テーブルのデフォルトの WITH オプションを設定できます。このカタログ配下で作成されたテーブルは、これらのデフォルト値を継承します。
     'sdkMode' = 'jdbc', -- jdbc モードを推奨します。
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- ワイドテーブルマージでリトラクションを防ぐために必要です。
     'partial-insert.enabled' = 'true', -- ワイドテーブルマージで部分的な列更新を有効にするために必要です。
     'mutateType' = 'insertOrUpdate', -- ワイドテーブルマージで部分的な列更新を有効にするために必要です。
     'table_property.binlog.level' = 'replica', -- カタログ作成時に永続的な Hologres テーブルプロパティを渡すこともできます。以降に作成されるテーブルでは、デフォルトでバイナリロギングが有効になります。
     'table_property.binlog.ttl' = '259200'
   );

   次のパラメータを、実際の Hologres サービスの情報に合わせて変更してください。

   パラメータ	説明	注意事項
   endpoint	Hologres インスタンスのエンドポイント。	Hologres インスタンスの詳細ページで、指定した VPC のドメイン名を取得します。ドメイン名の詳細については、「エンドポイント」をご参照ください。
   username	次のいずれかを選択します。 カスタムアカウントのユーザー名は、BASIC$<user_name> の形式にする必要があります。 Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。	設定するユーザーは、対応する Hologres データベースへのアクセス権限を持っている必要があります。詳細については、「Hologres 権限モデル」および「ユーザーの管理」をご参照ください。 この例では、BASIC$flinktest という名前のカスタムアカウントを使用し、パスワードをプレーンテキストで保存する際のセキュリティリスクを回避するため、holosecrect という名前のプロジェクト変数を使用してパスワードを設定します。詳細については、「プロジェクト変数」をご参照ください。
   password	カスタムアカウントのパスワード。 Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey secret。

   説明

   カタログを作成する際、ソース、ディメンション、結果テーブルのデフォルトの WITH オプションを設定できます。また、table_property で始まるパラメータなど、物理 Hologres テーブルのデフォルトプロパティを設定することもできます。詳細については、「Hologres カタログの管理」および「リアルタイムデータウェアハウス向け Hologres コネクタ (WITH パラメータ)」をご参照ください。

5. MySQL カタログを作成します。

   次のコードを [Scripts] タブにコピーし、パラメータ値を変更してから、コードを選択して [Run] をクリックします。これにより、作成したセッションクラスターが実行環境として使用されます。

   CREATE CATALOG mysqlcatalog WITH(
     'type' = 'mysql',
     'hostname' = '<hostname>',
     'port' = '<port>',
     'username' = '<username>',
     'password' = '${secret_values.mysql_pw}',
     'default-database' = 'order_dw'
   );

   次のパラメータを、実際の MySQL サービスの情報に合わせて変更してください。

   パラメータ	説明
   hostname	MySQL データベースの IP アドレスまたはホスト名。データベースの基本情報ページで、[Network Type] エリアの [View Connection Details] をクリックして、内部エンドポイントを取得します。
   port	MySQL データベースサービスのポート番号。デフォルト値は 3306 です。
   username	MySQL データベースサービスのユーザー名。
   password	MySQL データベースサービスのパスワード。 この例では、パスワードをプレーンテキストで記述するのを避けるため、mysql_pw という名前のプロジェクト変数を使用します。詳細については、「プロジェクト変数」をご参照ください。

ODS 層の構築：ビジネスデータの取り込み

カタログベースの CREATE DATABASE AS (CDAS) ステートメントを使用すると、ODS 層を 1 ステップで作成できます。ODS 層は通常、OLAP クエリやポイントクエリに直接使用されるのではなく、ストリーミングジョブのイベントソースとして機能します。この目的には、バイナリロギングを有効にするだけで十分です。バイナリロギングは Hologres のコア機能です。Hologres コネクタは、フル + インクリメンタルモードもサポートしています。最初にフルスナップショットを読み取り、次にバイナリログをインクリメンタルに消費します。

1. ODS CDAS 同期ジョブを作成します。

   1. [Development] > [ETL] ページで、ODS という名前の新しい SQL ストリームジョブの下書きを作成し、次のコードを SQL エディターにコピーします。

      -- table_property.binlog.level パラメーターはカタログ作成時に設定されたため、CDAS によって作成されたすべてのテーブルでバイナリロギングが有効になります。
      CREATE DATABASE IF NOT EXISTS dw.order_dw   
      AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- 必要に応じて、取り込むアップストリームテーブルを選択できます。
      /*+ OPTIONS('server-id'='8001-8004') */ ;   -- mysql-cdc インスタンスの server-id 範囲を指定します。

      説明

      • デフォルトでは、この例ではデータを order_dw データベースの Public スキーマに同期します。ターゲットの Hologres データベース内の指定したスキーマにデータを同期することもできます。詳細については、「CREATE DATABASE AS... ステートメントの宛先としてHologresカタログを使用する」をご参照ください。スキーマを指定すると、カタログ使用時のテーブル名の形式が変わります。詳細については、「Hologresカタログの使用」をご参照ください。

      • ソーステーブルのスキーマが変更された場合、結果テーブルのスキーマは、ソーステーブルでデータ変更 (削除、挿入、または更新) が発生するまで更新されません。

   2. 右上隅にある [Deploy] をクリックしてジョブをデプロイします。

   3. 左側のメニューで、[O&M] > [Deployments] を選択します。デプロイしたばかりの ODS ジョブの行で、[Actions] 列の [Start] をクリックします。[Initial Mode] を選択し、[Start] をクリックします。

2. コンピューティンググループにデータをロードします。

   テーブルグループは、Hologres におけるデータを格納する単位です。read_warehouse_1 コンピュートグループを使用して order_dw データベースのテーブルグループ、たとえば order_dw_tg_default (テーブルグループの作成については、「テーブルグループ管理」をご参照ください) からデータをクエリすると、order_dw_tg_default テーブルグループが read_warehouse_1 コンピュートグループにロードされます。これにより、init_warehouse コンピュートグループを使用してデータを書き込み、read_warehouse_1 コンピュートグループをサービス クエリに使用できるようになります。

   [HoloWeb] 開発ページで、[SQL Editor] をクリックします。インスタンス名とデータベース名を確認し、次のコマンドを実行します。詳細については、「新しいコンピューティンググループインスタンスの作成」をご参照ください。ロード後、read_warehouse_1 が order_dw_tg_default テーブルグループからデータをロードしたことを確認できます。

   -- 現在のデータベース内のテーブルグループを一覧表示します。
   SELECT tablegroup_name FROM hologres.hg_table_group_properties GROUP BY tablegroup_name;
   -- テーブルグループをコンピューティンググループにロードします。
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   -- コンピューティンググループにロードされたテーブルグループを確認します。
   select * from hologres.hg_warehouse_table_groups;

3. 右上隅で、コンピューティンググループを read_warehouse_1 に切り替えます。以降のクエリと分析では、このコンピューティンググループが使用されます。

   HoloWeb ページの右上隅で、コンピューティンググループのドロップダウンリストから read_warehouse_1 を選択します。

4. [SQL Editor] ページで、次のコマンドを実行して、MySQL から 3 つの Hologres テーブルに同期されたデータを表示します。

   -- ordersテーブルからデータをクエリします。
   SELECT * FROM orders;
   -- orders_payテーブルからデータをクエリします。
   SELECT * FROM orders_pay;
   -- product_catalogテーブルからデータをクエリします。
   SELECT * FROM product_catalog;

   product_catalog テーブルのクエリ結果には、[product_id] (1 から 5) と[catalog_name] (phone_aaa、phone_bbb、phone_ccc、phone_ddd、phone_eee) の 2 つの列と、合計 5 件のレコードが含まれます。これは、データが正常に Hologres に同期されたことを示します。

DWD 層の構築：リアルタイムワイドテーブルの作成

このステップでは、Hologres コネクタの列の部分更新機能を使用します。INSERT DML を使用して部分的な更新を表現できます。このジョブは、Hologres の行ストレージとハイブリッド行列表ストレージによって実現される高性能なポイントクエリを使用して、複数のディメンションテーブルをクエリします。強力なリソース分離により、書き込み、読み取り、分析のワークロードが互いに干渉することはありません。

1. Flink カタログ機能を使用して、Hologres に DWD 層のワイドテーブル dwd_orders を作成します。

   [Development] > [Scripts] ページで、次のコードを [Scripts] タブにコピーし、コードを選択してから [Run] をクリックします。

   -- 異なるストリームが同じ結果テーブルに書き込むため、ワイドテーブルの列はnull許容である必要があります。どの列もnullになる可能性があります。
   CREATE TABLE dw.order_dw.dwd_orders (
     order_id bigint not null,
     order_user_id string,
     order_shop_id bigint,
     order_product_id bigint,
     order_product_catalog_name string,
     order_fee numeric(20,2),
     order_create_time timestamp,
     order_update_time timestamp,
     order_state int,
     pay_id bigint,
     pay_platform int comment 'プラットフォーム 0: phone、1: pc', 
     pay_create_time timestamp,
     PRIMARY KEY(order_id) NOT ENFORCED
   );
   -- カタログを介してHologresの物理テーブルプロパティを変更できます。
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' -- バイナリログの TTL を 1 週間に変更します。
   );

2. ODS 層のテーブル orders と orders_pay からのバイナリロギングの変更をリアルタイムで消費します。

   [Development] > [ETL] ページで、DWD という名前の新しい SQL ストリームジョブの下書きを作成します。次のコードを SQL エディターにコピーし、ジョブを[デプロイ]して[開始]します。この SQL ジョブは、テンポラル結合を使用して orders テーブルと product_catalog テーブルを結合し、結果を dwd_orders テーブルに書き込み、リアルタイムでデータをエンリッチします。

   BEGIN STATEMENT SET;
   INSERT INTO dw.order_dw.dwd_orders 
    (
      order_id,
      order_user_id,
      order_shop_id,
      order_product_id,
      order_fee,
      order_create_time,
      order_update_time,
      order_state,
      order_product_catalog_name
    ) SELECT o.*, dim.catalog_name 
      FROM dw.order_dw.orders as o
      LEFT JOIN dw.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
      ON o.product_id = dim.product_id;
   INSERT INTO dw.order_dw.dwd_orders 
     (pay_id, order_id, pay_platform, pay_create_time)
      SELECT * FROM dw.order_dw.orders_pay;
   END;

3. dwd_orders ワイドテーブルのデータを表示します。

   [HoloWeb] 開発ページで Hologres インスタンスに接続し、対象のデータベースにログインしてから、SQL Editor で次のコマンドを実行します。

   SELECT * FROM dwd_orders;

   正常に実行されると、クエリ結果は [dwd_orders] ワイドテーブルから、[order_id]、[order_user_id]、[order_shop_id]、[order_product_id]、[order_product_catalog_name]、[order_fee]、[order_create_time]、[order_update_time] などのフィールドを含むデータを返します。

DWS 層の構築：リアルタイムメトリクスの計算

1. Flink カタログ機能を使用して、Hologres に DWS 層の集計テーブル dws_users と dws_shops を作成します。

   [Development] > [Scripts] ページで、次のコードを [Scripts] タブにコピーし、コードを選択してから [Run] をクリックします。

   -- ユーザーディメンションの集計テーブル。
   CREATE TABLE dw.order_dw.dws_users (
     user_id string not null,
     ds string not null,
     paied_buy_fee_sum numeric(20,2) not null comment '当日完了した支払いの合計金額',
     primary key(user_id,ds) NOT ENFORCED
   );
   -- 店舗ディメンションの集計テーブル。
   CREATE TABLE dw.order_dw.dws_shops (
     shop_id bigint not null,
     ds string not null,
     paied_buy_fee_sum numeric(20,2) not null comment '当日完了した支払いの合計金額',
     primary key(shop_id,ds) NOT ENFORCED
   );

2. DWD 層のワイドテーブル dw.order_dw.dwd_orders をリアルタイムで消費し、Flink で集計を実行し、最終結果を Hologres の DWS テーブルに書き込みます。

   [Development] > [ETL] ページで、DWS という名前の新しい SQL ストリームジョブの下書きを作成します。次のコードを SQL エディターにコピーし、ジョブを[デプロイ]して[開始]します。

   BEGIN STATEMENT SET;
   INSERT INTO dw.order_dw.dws_users
     SELECT 
       order_user_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       SUM (order_fee)
       FROM dw.order_dw.dwd_orders c
       WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- 注文ストリームと支払いストリームの両方のデータがワイドテーブルに書き込まれています。
       GROUP BY order_user_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   INSERT INTO dw.order_dw.dws_shops
     SELECT 
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       SUM (order_fee)
      FROM dw.order_dw.dwd_orders c
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- 注文ストリームと支払いストリームの両方のデータがワイドテーブルに書き込まれています。
      GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   END;

3. DWS 層の集計結果を表示します。アップストリームのデータが変更されると、結果はリアルタイムで更新されます。

   1. 変更前のデータを Hologres コンソールで表示します。

      dws_users テーブル

      SELECT * FROM dws_users;

      クエリを実行すると、[dws_users] テーブルから、[user_id]、[ds]、[paied_buy_fee_sum] の 3 つの列を含むデータが返されます。結果例では、[user_id] 列には値 user_001、user_002、user_003 が、[ds] 列には値 20230215 が、[paied_buy_fee_sum] 列にはそれぞれ値 8000.08、5000.05、5000.05 が含まれます。[user_id] 列は各ユーザーを一意に識別します。

      dws_shops テーブル

      SELECT * FROM dws_shops;

      クエリ結果から、dws_shops テーブルに[shop_id] (店舗 ID)、[ds] (日付パーティション)、および[paied_buy_fee_sum] (支払額) の 3 つの列が含まれていることがわかります。サンプルデータが 4 行返され、DWS 層のテーブルが正常に構築されたことが確認できます。

   2. RDS コンソールで、order_dw データベースの orders テーブルと orders_pay テーブルに、それぞれ 1 件の新しいレコードを挿入します。

      INSERT INTO orders VALUES
      (100008, 'user_003', 12345, 5, 6000.02, '2023-02-15 09:40:56', '2023-02-15 18:42:56', 1);
      INSERT INTO orders_pay VALUES
      (2008, 100008, 1, '2023-02-15 19:40:56');

   3. 更新されたデータを Hologres コンソールで表示します。

      dwd_orders テーブル

      SELECT * FROM dwd_orders;

      クエリを実行すると、dwd_orders テーブルから 8 件の注文レコードが返されます。これには、[order_id]、[order_user_id]、[order_shop_id]、[order_product_id]、[order_product_catalog_name]、[order_fee]、[order_create_time]、[order_update_time] などのフィールドが含まれます。8 番目のレコード (order_id=100008、user_003、phone_eee、6000.02) が新しく書き込まれたデータです。

      dws_users テーブル

      SELECT * FROM dws_users;

      クエリは、[user_id]、[ds]、および[paied_buy_fee_sum] の 3 つの列を含む 3 行を返します (user_001 / 20230215 / 8000.08、user_002 / 20230215 / 5000.05、user_003 / 20230215 / 11000.07)。合計支払額が最も高いのは user_003 (11000.07) です。

      dws_shops テーブル

      SELECT * FROM dws_shops;

      クエリ実行後、結果には[shop_id]、[ds]、[paied_buy_fee_sum] の 3 つの列と、20230215 の店舗 12345、12346、12347、12348 の料金 (11000.07、4000.04、7000.07、2000.02) を示す 4 行のデータが表示されます。[shop_id] と [paied_buy_fee_sum] が主要なメトリクス列です。

データのプロファイリング

バイナリロギングが有効になっているため、データの変更を直接確認できます。中間結果に対してアドホックなビジネスデータ探索を実行したり、最終的な計算の正しさを検証したりする必要がある場合、このソリューションの各層は永続化されているため、中間プロセスを簡単に調べることができます。

SELECT *
FROM dw.order_dw.dwd_orders /*+ OPTIONS('binlog'='false') */ 
WHERE order_user_id = 'user_001' and order_create_time > '2023-02-15 12:00:00'; -- バッチモードはフィルタープッシュダウンをサポートし、バッチジョブの実行効率を向上させます。

ポイントクエリ

DWS 層の集計指標テーブルをプライマリキーでクエリします。秒間リクエスト数 (RPS) が数百万規模でも処理できます。

HoloWeb の開発ページで、次の SQL を実行して、特定のユーザーの特定の日付における消費金額をクエリします。

-- Hologres SQL
SELECT * FROM dws_users WHERE user_id ='user_001' AND ds = '20230215';

クエリ結果には、user_id、ds、paied_buy_fee_sum (消費金額) の 3 列が返されます。user_001 の 20230215 における消費金額は 18888.88 です。

OLAP クエリ

DWD 層のワイドテーブルに対して OLAP クエリを実行します。

HoloWeb の開発ページで、次の SQL を実行して、2023 年 2 月における特定の顧客の特定の支払いプラットフォームでの注文の詳細をクエリします。

-- Hologres SQL
SELECT * FROM dwd_orders
WHERE order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
AND order_user_id = 'user_001'
AND pay_platform = 0
ORDER BY order_create_time LIMIT 100;

クエリ結果には、order_id、order_user_id、order_shop_id、order_product_id、order_fee、order_create_time、order_update_time などのフィールドを含む 2 件の注文レコードが返されます。この例の注文IDは 100006 と 100004 です。

リアルタイムレポート

DWD 層のワイドテーブルのデータに基づいてリアルタイムレポートを生成します。Hologres のハイブリッド行列表ストレージと列指向テーブルは優れた OLAP クエリ機能を提供するため、数秒で応答できます。

HoloWeb の開発ページで、次の SQL を実行して、2023 年 2 月における各商品カテゴリの注文数合計と注文金額合計をクエリします。

-- Hologres SQL
SELECT
  TO_CHAR(order_create_time, 'YYYYMMDD') AS order_create_date,
  order_product_catalog_name,
  COUNT(*),
  SUM(order_fee)
FROM
  dwd_orders
WHERE
  order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
GROUP BY
  order_create_date, order_product_catalog_name
ORDER BY
  order_create_date, order_product_catalog_name;

SQL ステートメントを実行すると、[Result] タブに、order_create_date、order_product_catalog_name、count、sum の 4 列からなるテーブルが表示されます。例えば、日付 20230215 の結果では、カテゴリ phone_aaa は注文数が 2、合計が 6000.06、カテゴリ phone_bbb は注文数が 1、合計が 4000.04、というように 5 つのカテゴリすべての結果が表示されます。