Hologres:Real-time UV counting with Hologres and Flink using pre-aggregation

Real-time UV counting workflow

This solution features a simple data pipeline, allows flexible computation across any dimension, and uses a single bitmap for storage, which avoids storage explosion issues. It also guarantees real-time updates to create a more responsive, flexible, and powerful multi-dimensional analytical data warehouse.

Procedure

1. Create base tables in Hologres

   1. Create a user mapping table

      In Hologres, run the following statements to create a user mapping table named uid_mapping. This table maps UIDs to 32-bit integers. If your original UIDs are already 32-bit integers, you can skip this step.

      • User IDs from business systems or tracking points are often strings or long integers. Therefore, you need to create a uid_mapping table. The roaring bitmap data type requires user IDs to be 32-bit integers that are as dense (preferably consecutive) as possible. The mapping table uses the Hologres SERIAL type, which is an auto-incrementing 32-bit integer, to automatically manage and maintain a stable mapping.

      • To improve the queries per second (QPS) of Flink dimension table joins, set this table in Hologres as a row-oriented table.

      • You must enable the appropriate GUC parameters to use the optimized execution engine for writing data to tables that contain an auto-increment field. For more information, see Accelerate SQL execution by using Fixed Plan.

      -- Enable GUC parameters to support Fixed Plan writes for tables that contain SERIAL-type columns.
      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)
      );
      -- Set uid as the clustering_key and distribution_key to quickly find its corresponding int32 value.
      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. Create an aggregation result table

      Create a table named dws_app as an aggregation result table to store the results that are aggregated based on basic dimensions.

      Before you use roaring bitmap functions, you must create the roaringbitmap extension. Your Hologres instance must be V0.10 or later.

      CREATE EXTENSION IF NOT EXISTS roaringbitmap;

      Compared to an offline result table, this table includes a timestamp field to enable statistics based on the Flink time window. The following DDL statements define the result table.

      BEGIN;
      CREATE TABLE dws_app(
        country text,
        prov text,
        city text,
        ymd text NOT NULL,  -- Date field
        timetz TIMESTAMPTZ,  -- Statistics timestamp, which allows statistics to be calculated based on the Flink window period.
        uid32_bitmap roaringbitmap, -- Use a roaring bitmap to record UVs.
       PRIMARY KEY (country, prov, city, ymd, timetz)-- Use query dimensions and time as the primary key to prevent duplicate data insertion.
      );
      CALL set_table_property('public.dws_app', 'orientation', 'column');
      -- Set the date field as the clustering key and event_time_column for efficient filtering.
      CALL set_table_property('public.dws_app', 'clustering_key', 'ymd');
      CALL set_table_property('public.dws_app', 'event_time_column', 'ymd');
      -- Set the GROUP BY fields as the distribution key.
      CALL set_table_property('public.dws_app', 'distribution_key', 'country,prov,city');
      COMMIT;

2. Use Flink to read data in real time and update the aggregation result table

   For the complete source code of the Flink example, see alibabacloud-hologres-connectors examples. The following steps describe the operations in Flink.

   1. Read data from a data source as a DataStream and convert the DataStream to a Table

      In Flink, read data from a data source, such as a CSV file, Kafka, or Redis.

      // A CSV file is used as the data source in this example. You can also use other data sources such as Kafka or Redis.
      DataStreamSource odsStream = env.createInput(csvInput, typeInfo);
      // A proctime field must be added to join with the dimension table.
      Table odsTable =
          tableEnv.fromDataStream(
          odsStream,
          $("uid"),
          $("country"),
          $("prov"),
          $("city"),
          $("ymd"),
          $("proctime").proctime());
      // Register the table in the catalog environment.
      tableEnv.createTemporaryView("odsTable", odsTable);

   2. Join the source table with the Hologres dimension table (uid_mapping)

      When you create the Hologres dimension table in Flink, use the insertIfNotExists parameter to automatically insert data that does not exist. The uid_int32 field is automatically generated using the SERIAL type of Hologres. Join the Flink source table with the Hologres dimension table. The following code provides an example.

      -- Create the Hologres dimension table. 'insertIfNotExists' indicates that if the data is not found, it is automatically inserted.
      String createUidMappingTable =
          String.format(
          "create table uid_mapping_dim("
          + "  uid string,"
          + "  uid_int32 INT"
          + ") with ("
          + "  'connector'='hologres',"
          + "  'dbname' = '%s'," // The name of the Hologres database.
          + "  'tablename' = '%s',"// The name of the Hologres table.
          + "  'username' = '%s'," // The AccessKey ID of your account.
          + "  'password' = '%s'," // The AccessKey secret of your account.
          + "  'endpoint' = '%s'," // The endpoint of the Hologres instance.
          + "  'insertifnotexists'='true'"
          + ")",
          database, dimTableName, username, password, endpoint);
      tableEnv.executeSql(createUidMappingTable);
      -- Join the source table with the dimension table.
      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. Convert the join result to a DataStream

      Process the data using a Flink time window and use a roaring bitmap to deduplicate metrics. The following code provides an example.

      DataStream<Tuple6<String, String, String, String, Timestamp, byte[]>> processedSource =
          source
          -- Filter the dimensions that require statistics. In this example, the dimensions are country, prov, city, and ymd.
          .keyBy(0, 1, 2, 3)
          -- Tumbling time window. Since a CSV file is used to simulate the input stream, ProcessingTime is used. In a production environment, you can use EventTime.
          .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
          -- A trigger that allows you to obtain aggregate results before the window closes.
          .trigger(ContinuousProcessingTimeTrigger.of(Time.minutes(1)))
          .aggregate(
          -- The aggregate function aggregates data based on the dimensions selected in 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) {
                      -- Add the 32-bit UID to the roaring bitmap for deduplication.
                      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);
                  }
              },
          -- The window function outputs the aggregated result.
          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();
                      // Optimize the roaring bitmap.
                      result.runOptimize();
                      // Convert the roaring bitmap to a byte array to store it in Hologres.
                      byte[] byteArray = new byte[result.serializedSizeInBytes()];
                      result.serialize(ByteBuffer.wrap(byteArray));
                      // The timestamp (Tuple6.f4) marks the end of the time window, which determines the granularity of the statistics.
                      out.collect(
                          new Tuple6<>(
                              keys.getField(0),
                              keys.getField(1),
                              keys.getField(2),
                              keys.getField(3),
                              new Timestamp(
                                  timeWindow.getEnd() / 1000 * 1000),
                              byteArray));
              }
          });

   4. Write data to the Hologres aggregation result table

      Write the data deduplicated by Flink to the Hologres result table dws_app. Note that the roaringbitmap type in Hologres corresponds to the byte array type in Flink. The following code provides an example in Flink.

       -- Convert the calculation results to a table.
      Table resTable =
          tableEnv.fromDataStream(
              processedSource,
              $("country"),
              $("prov"),
              $("city"),
              $("ymd"),
              $("timest"),
              $("uid32_bitmap"));
      -- Create the Hologres sink table. The roaringbitmap type in Hologres corresponds to the BYTES type in Flink's table definition.
      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);
      -- Write the calculation results to the dws_app table.
      tableEnv.executeSql("insert into sink select * from " + resTable);

3. Query data

   In Hologres, calculate UVs from the aggregation result table (dws_app). Aggregate the data by your query dimensions and calculate the bitmap's cardinality to count the users that match the GROUP BY conditions.

   • Example 1: Query the UV count for each city on a specific day.

     -- Before you run the RB_AGG operation, you can disable the three-stage aggregation switch for better performance. This step is optional because the switch is disabled by default.
     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
     ;

   • Example 2: Query the UV and PV counts for each province within a specific time period.

     -- Before you run the RB_AGG operation, you can disable the three-stage aggregation switch for better performance. This step is optional because the switch is disabled by default.
     set hg_experimental_enable_force_three_stage_agg=off;
     SELECT  country
             ,prov
             ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) AS uv
             ,SUM(pv) AS pv
     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. Visualize the results

   After you calculate UVs and PVs, you typically use a BI tool for visualization. Because the queries require the RB_CARDINALITY and RB_OR_AGG aggregate functions, you need a BI tool that supports custom aggregate functions. Common BI tools with this capability include Apache Superset and Tableau.

   • Apache Superset

     1. Connect Apache Superset to Hologres. For more information, see Connect Apache Superset to Hologres.

     2. Set the dws_app table as a dataset. In Apache Superset, click Add Dataset. In the dialog box that appears, set DATASOURCE to postgresql holo_rb_demo, SCHEMA to public, and TABLE to dws_app.

     3. In the dataset, create a metric named UV using the following expression. On the dataset editing page, select the METRICS tab and click + ADD ITEM to add metrics. Set Metric to count and SQL Expression to COUNT(*). Set Metric to uv and SQL Expression to RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)). Click SAVE.

        RB_CARDINALITY(RB_OR_AGG(uid32_bitmap))

        You can now explore the data.

     4. (Optional) Create a dashboard.

        For more information about how to create a dashboard, see Create a Dashboard.

   • Tableau

     1. Connect Tableau to Hologres. For more information, see Connect Tableau to Hologres.

        You can use Tableau's pass-through functions to run custom functions. For more information, see Pass-Through Functions (RAWSQL).

     2. In Tableau, create a calculated field named UV and enter the following formula.

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

        You can now explore the data.

     3. (Optional) Create a dashboard.

        For more information about how to create a dashboard, see Create a Dashboard.