Hologres:Hologres と Flink による事前集計を利用したリアルタイム UV カウント

リアルタイム UV カウントのワークフロー

このソリューションは、単純なデータパイプラインを特徴とし、任意のディメンションにわたる柔軟な計算を可能にし、単一のビットマップをストレージに使用するため、ストレージ爆発の問題を回避します。また、リアルタイムの更新を保証し、より応答性が高く、柔軟で、強力な多次元分析データウェアハウスを作成します。

操作手順

1. Hologres でのベーステーブルの作成

   1. ユーザーマッピングテーブルの作成

      Hologres で、次のステートメントを実行して、uid_mapping という名前のユーザーマッピングテーブルを作成します。このテーブルは、UID を 32 ビット整数にマッピングします。元の UID がすでに 32 ビット整数の場合は、この手順をスキップできます。

      • ビジネスシステムやトラッキングポイントからのユーザー ID は、多くの場合、文字列または long 型の整数です。そのため、uid_mapping テーブルを作成する必要があります。Roaring Bitmap データ型では、ユーザー ID が 32 ビット整数であり、できるだけ密 (できれば連続) である必要があります。マッピングテーブルは、Hologres の SERIAL 型 (自動インクリメントされる 32 ビット整数) を使用し、安定したマッピングを自動的に管理および維持します。

      • Flink のディメンションテーブル結合の秒間クエリ数 (QPS) を向上させるには、Hologres でこのテーブルのオリエンテーションを row に設定し、行指向テーブルとして定義します。

      • 自動インクリメントフィールドを含むテーブルにデータを書き込むために最適化された実行エンジンを使用するには、適切な GUC パラメータを有効にする必要があります。詳細については、「Fixed Plan を使用した SQL 実行の高速化」をご参照ください。

      -- SERIAL 型の列を含むテーブルへの Fixed Plan 書き込みをサポートするために GUC パラメータを有効化します。
      alter database <dbname> set hg_experimental_enable_fixed_dispatcher_autofill_series=on;
      alter database <dbname> set hg_experimental_enable_fixed_dispatcher_for_multi_values=on;
      BEGIN;
      CREATE TABLE public.uid_mapping (
      uid text NOT NULL,
      uid_int32 serial,
      PRIMARY KEY (uid)
      );
      -- uid を clustering_key および distribution_key として設定し、対応する int32 値を迅速に検索できるようにします。
      CALL set_table_property('public.uid_mapping', 'clustering_key', 'uid');
      CALL set_table_property('public.uid_mapping', 'distribution_key', 'uid');
      CALL set_table_property('public.uid_mapping', 'orientation', 'row');
      COMMIT;

   2. 集計結果テーブルの作成

      基本ディメンションに基づいて集計された結果を格納するために、集計結果テーブルとして dws_app という名前のテーブルを作成します。

      Roaring Bitmap 関数を使用する前に、roaringbitmap 拡張機能を作成する必要があります。Hologres インスタンスは V0.10 以降である必要があります。

      CREATE EXTENSION IF NOT EXISTS roaringbitmap;

      オフラインの結果テーブルと比較して、このテーブルにはタイムスタンプフィールドが含まれており、Flink のタイムウィンドウに基づいた統計を可能にします。次の DDL ステートメントは、結果テーブルを定義します。

      BEGIN;
      CREATE TABLE dws_app(
        country text,
        prov text,
        city text,
        ymd text NOT NULL,  -- 日付フィールド
        timetz TIMESTAMPTZ,  -- 統計タイムスタンプ。Flink のウィンドウ期間に基づいて統計を計算できます。
        uid32_bitmap roaringbitmap, -- Roaring Bitmap を使用して UV を記録します。
       PRIMARY KEY (country, prov, city, ymd, timetz)-- クエリディメンションと時刻を主キーとして使用し、重複データの挿入を防ぎます。
      );
      CALL set_table_property('public.dws_app', 'orientation', 'column');
      -- 日付フィールドを clustering_key および event_time_column として設定し、効率的なフィルタリングを可能にします。
      CALL set_table_property('public.dws_app', 'clustering_key', 'ymd');
      CALL set_table_property('public.dws_app', 'event_time_column', 'ymd');
      -- GROUP BY フィールドを分散キーとして設定します。
      CALL set_table_property('public.dws_app', 'distribution_key', 'country,prov,city');
      COMMIT;

2. Flink を使用したリアルタイムデータ読み取りと集計結果テーブルの更新

   Flink の例の完全なソースコードについては、「alibabacloud-hologres-connectors examples」をご参照ください。次の手順では、Flink での操作について説明します。

   1. DataStream へのデータソース読み取りと Table への変換

      Flink で、CSV ファイル、Kafka、Redis などのデータソースからデータを読み取ります。

      // この例では、データソースとして CSV ファイルを使用します。Kafka や Redis などの他のデータソースも使用できます。
      DataStreamSource odsStream = env.createInput(csvInput, typeInfo);
      // ディメンションテーブルと結合するために、proctime フィールドを追加する必要があります。
      Table odsTable =
          tableEnv.fromDataStream(
          odsStream,
          $("uid"),
          $("country"),
          $("prov"),
          $("city"),
          $("ymd"),
          $("proctime").proctime());
      // テーブルをカタログ環境に登録します。
      tableEnv.createTemporaryView("odsTable", odsTable);

   2. ソーステーブルと Hologres ディメンションテーブルの結合

      Flink で Hologres のディメンションテーブルを作成するときは、insertIfNotExists パラメータを使用して、存在しないデータを自動的に挿入します。uid_int32 フィールドは、Hologres の SERIAL 型を使用して自動的に生成されます。Flink のソーステーブルを Hologres のディメンションテーブルと結合します。次のコードは例を示しています。

      // Hologres のディメンションテーブルを作成します。「insertIfNotExists」は、データが見つからない場合に自動的に挿入されることを指定します。
      String createUidMappingTable =
          String.format(
          "create table uid_mapping_dim("
          + "  uid string,"
          + "  uid_int32 INT"
          + ") with ("
          + "  'connector'='hologres',"
          + "  'dbname' = '%s'," // Hologres データベースの名前
          + "  'tablename' = '%s',"// Hologres テーブルの名前
          + "  'username' = '%s'," // アカウントの AccessKey ID
          + "  'password' = '%s'," // アカウントの AccessKey シークレット
          + "  'endpoint' = '%s'," // Hologres インスタンスのエンドポイント
          + "  'insertifnotexists'='true'"
          + ")",
          database, dimTableName, username, password, endpoint);
      tableEnv.executeSql(createUidMappingTable);
      // ソーステーブルをディメンションテーブルと結合します。
      String odsJoinDim =
          "SELECT ods.country, ods.prov, ods.city, ods.ymd, dim.uid_int32"
          + "  FROM odsTable AS ods JOIN uid_mapping_dim FOR SYSTEM_TIME AS OF ods.proctime AS dim"
          + "  ON ods.uid = dim.uid";
      Table joinRes = tableEnv.sqlQuery(odsJoinDim);

   3. 結合結果の DataStream への変換

      Flink のタイムウィンドウを使用してデータを処理し、Roaring Bitmap を使用してメトリクスを重複排除します。次のコードは例を示しています。

      DataStream<Tuple6<String, String, String, String, Timestamp, byte[]>> processedSource =
          source
          // 統計が必要なディメンション (この例では country、prov、city、ymd) でキー分割します。
          .keyBy(0, 1, 2, 3)
          // タンブリングタイムウィンドウ。CSV ファイルを使用して入力ストリームをシミュレートするため、ProcessingTime を使用します。
          // 本番環境では、EventTime を使用できます。
          .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
          // ウィンドウが閉じる前に集計結果を取得できるトリガー。
          .trigger(ContinuousProcessingTimeTrigger.of(Time.minutes(1)))
          .aggregate(
          // aggregate 関数は、keyBy で選択されたディメンションに基づいてデータを集計します。
          new AggregateFunction<
              Tuple5<String, String, String, String, Integer>,
              RoaringBitmap,
              RoaringBitmap>() {
                  @Override
                  public RoaringBitmap createAccumulator() {
                      return new RoaringBitmap();
                  }
                  @Override
                  public RoaringBitmap add(
                      Tuple5<String, String, String, String, Integer> in,
                      RoaringBitmap acc) {
                      // 32 ビット UID を Roaring Bitmap に追加して重複排除します。
                      acc.add(in.f4);
                      return acc;
                  }
                  @Override
                  public RoaringBitmap getResult(RoaringBitmap acc) {
                      return acc;
                  }
                  @Override
                  public RoaringBitmap merge(
                      RoaringBitmap acc1, RoaringBitmap acc2) {
                      return RoaringBitmap.or(acc1, acc2);
                  }
              },
          // window 関数は集計結果を出力します。
          new WindowFunction<
              RoaringBitmap,
              Tuple6<String, String, String, String, Timestamp, byte[]>,
              Tuple,
              TimeWindow>() {
                  @Override
                  public void apply(
                      Tuple keys,
                      TimeWindow timeWindow,
                      Iterable<RoaringBitmap> iterable,
                      Collector<
                      Tuple6<String, String, String, String, Timestamp, byte[]>> out)
                      throws Exception {
                      RoaringBitmap result = iterable.iterator().next();
                      // Roaring Bitmap を最適化します。
                      result.runOptimize();
                      // Roaring Bitmap をバイト配列に変換して Hologres に格納します。
                      byte[] byteArray = new byte[result.serializedSizeInBytes()];
                      result.serialize(ByteBuffer.wrap(byteArray));
                      // タイムスタンプ (Tuple6.f4) はタイムウィンドウの終了を示し、統計の粒度を決定します。
                      out.collect(
                          new Tuple6<>(
                              keys.getField(0),
                              keys.getField(1),
                              keys.getField(2),
                              keys.getField(3),
                              new Timestamp(
                                  timeWindow.getEnd() / 1000 * 1000),
                              byteArray));
              }
          });

   4. Hologres 集計結果テーブルへのデータ書き込み

      Flink によって重複排除されたデータを、Hologres の結果テーブル dws_app に書き込みます。Hologres の roaringbitmap 型は、Flink の バイト配列型に対応することに注意してください。次のコードは Flink での例を示しています。

      // 計算結果をテーブルに変換します。
      Table resTable =
          tableEnv.fromDataStream(
              processedSource,
              $("country"),
              $("prov"),
              $("city"),
              $("ymd"),
              $("timest"),
              $("uid32_bitmap"));
      // Hologres の sink テーブルを作成します。Hologres の roaringbitmap 型は、Flink のテーブル定義における BYTES 型に対応します。
      String createHologresTable =
          String.format(
              "create table sink("
              + "  country string,"
              + "  prov string,"
              + "  city string,"
              + "  ymd string,"
              + "  timetz timestamp,"
              + "  uid32_bitmap BYTES"
              + ") with ("
              + "  'connector'='hologres',"
              + "  'dbname' = '%s',"
              + "  'tablename' = '%s',"
              + "  'username' = '%s',"
              + "  'password' = '%s',"
              + "  'endpoint' = '%s',"
              + "  'connectionSize' = '%s',"
              + "  'mutatetype' = 'insertOrReplace'"
              + ")",
          database, dwsTableName, username, password, endpoint, connectionSize);
      tableEnv.executeSql(createHologresTable);
      // 計算結果を dws_app テーブルに書き込みます。
      tableEnv.executeSql("insert into sink select * from " + resTable);

3. データのクエリ

   Hologres で、集計結果テーブル (dws_app) から UV を計算します。クエリディメンションによってデータを集計し、ビットマップのカーディナリティを計算して、GROUP BY 条件に一致するユーザーをカウントします。

   • 例 1：特定の日付における各都市の UV 数をクエリします。

     -- RB_AGG 操作を実行する前に、パフォーマンス向上のために 3 段階集計スイッチを無効にすることができます。このスイッチはデフォルトで無効になっているため、この手順は任意です。
     set hg_experimental_enable_force_three_stage_agg=off;
     SELECT  country
             ,prov
             ,city
             ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) AS uv
     FROM    dws_app
     WHERE   ymd = '20210329'
     GROUP BY country
              ,prov
              ,city
     ;

   • 例 2：特定の期間における各省の UV 数をクエリします。

     -- RB_AGG 操作を実行する前に、パフォーマンス向上のために 3 段階集計スイッチを無効にすることができます。このスイッチはデフォルトで無効になっているため、この手順は任意です。
     set hg_experimental_enable_force_three_stage_agg=off;
     SELECT  country
             ,prov
             ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) AS uv
     FROM    dws_app
     WHERE   timetz > '2021-04-19 18:00:00+08' and timetz < '2021-04-19 19:00:00+08'
     GROUP BY country
              ,prov
     ;

4. 結果の可視化

   UV と PV を計算した後、通常は BI ツールを使用して可視化します。クエリには RB_CARDINALITY および RB_OR_AGG 集計関数が必要なため、カスタム集計関数をサポートする BI ツールが必要です。この機能を持つ一般的な BI ツールには、Apache Superset と Tableau があります。

   • Apache Superset

     1. Apache Superset を Hologres に接続します。詳細については、「Apache Superset と Hologres の接続」をご参照ください。

     2. dws_app テーブルをデータセットとして設定します。Apache Superset で、[Add Dataset] をクリックします。表示されるダイアログボックスで、[DATASOURCE] を postgresql holo_rb_demo に、[SCHEMA] を public に、[TABLE] を dws_app に設定します。

     3. データセットで、次の式を使用して UV という名前のメトリクスを作成します。データセット編集ページで、[METRICS] タブを選択し、[+ ADD ITEM] をクリックしてメトリクスを追加します。Metric を count に、SQL Expression を COUNT(*) に設定します。Metric を uv に、SQL Expression を RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) に設定します。[SAVE] をクリックします。

        RB_CARDINALITY(RB_OR_AGG(uid32_bitmap))

        これで、データを探索できます。

     4. (任意) ダッシュボードを作成します。

        ダッシュボードの作成方法の詳細については、「ダッシュボードの作成」をご参照ください。

   • Tableau

     1. Tableau を Hologres に接続します。詳細については、「Tableau と Hologres の接続」をご参照ください。

        Tableau のパススルー関数を使用して、カスタム関数を実行できます。詳細については、「パススルー関数 (RAWSQL)」をご参照ください。

     2. Tableau で、[UV] という名前の計算フィールドを作成し、次の数式を入力します。

        RAWSQLAGG_INT("RB_CARDINALITY(RB_OR_AGG(%1))", [Uid32 Bitmap])

        これで、データを探索できます。

     3. (任意) ダッシュボードを作成します。

        ダッシュボードの作成方法の詳細については、「ダッシュボードの作成」をご参照ください。