Hologres:Build a real-time data warehouse with Flink and Hologres

Create an RDS MySQL instance and prepare a data source

1. Create an RDS MySQL instance. For more information, see Create an ApsaraDB RDS for MySQL instance.

   The RDS MySQL instance must be in the same VPC as the Flink workspace and Hologres instance.

2. Create a database and an account.

   On the target instance, create a database named order_dw and a standard account with read and write permissions for that database. For details, see Create a database and Create an account.

3. Prepare the MySQL CDC data source.

   1. On the instance details page, click Log On to Database.

   2. In the Connect to Instance dialog box, enter the username and password for the account you created, and then click Sign in.

   3. After you log on, double-click the order_dw database on the database instance page to switch to it.

   4. In the SQL Console area, write the DDL statements to create the three business tables and the statements to insert data.

      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
      );
      -- Prepare data
      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. Click Upload, and then click Execute.

Create a Hologres instance and a virtual warehouse

1. Create an exclusive Hologres instance. For more information, see Purchase a Hologres instance.

   The Hologres instance must be in the same VPC as the RDS MySQL instance. To experience the strong resource isolation capability of Hologres through read/write splitting, select Virtual Warehouse as the instance type and set Reserved Computing Resources of Virtual Warehouse to 64. This allows you to create a new virtual warehouse.

2. After you log on to the instance, create a database and grant permissions.

   Create a database named order_dw (Simple Permission Model must be enabled) and grant admin permissions to the user. For more information about creating a database and granting permissions, see DB Management.

   Note

   • If an account does not appear in the User Account drop-down list, it has not been added to the instance. Go to the Users page to add the user as a SuperUser.

   • In Hologres V2.0 and later, the Binlog extension is enabled by default and does not require manual execution.

3. Create a new virtual warehouse.

   You can use different virtual warehouses to achieve resource isolation. Use the initial virtual warehouse init_warehouse for writing data, and use the read_warehouse_1 virtual warehouse for serving queries.

   The reserved computing resources are fully allocated to the initial virtual warehouse init_warehouse. You must first reduce its resources before creating a new one. For more information, see Create a new virtual warehouse instance.

   1. Click Security Center > Virtual Warehouse Management, and confirm that the instance name is correct.

   2. In the row for the existing virtual warehouse init_warehouse, click Modify Configuration in the Actions column. Reduce the resources and click OK.

   3. Click Create Virtual Warehouse, create a new virtual warehouse named read_warehouse_1, and click OK.

Create a Flink workspace and catalogs

1. Create a Flink workspace. For more information, see Activate Realtime Compute for Apache Flink.

   The Flink workspace must be in the same VPC as the RDS MySQL and Hologres instances.

2. Log on to the Realtime Compute for Apache Flink console, and in the row of the target workspace, click Console in the Actions column.

3. Create a session cluster to provide an execution environment for creating catalogs and query scripts. For more information, see Step 1: Create a session cluster.

4. Create a Hologres catalog.

   On the Script tab of the Development > Scripts page, copy the following code to the script editor. Modify the target parameter values, select the target snippet, and then click Run. In the lower-right corner, ensure the session cluster you created is selected as the execution environment.

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = 'BASIC$flinktest',
     'password' = '${secret_values.holosecret}',
     'dbname' = 'order_dw@init_warehouse', -- Specify the database name and connect to the init_warehouse virtual warehouse.
     'binlog' = 'true', -- When you create a catalog, you can set the WITH parameters for source, dimension, and result tables. These default parameters are automatically added when you use tables under this catalog.
     'sdkMode' = 'jdbc', -- The jdbc mode is recommended.
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- This must be enabled for wide table merges to prevent retractions.
     'partial-insert.enabled' = 'true', -- This parameter must be enabled for wide table merges to allow partial column updates.
     'mutateType' = 'insertOrUpdate', -- This parameter must be enabled for wide table merges to allow partial column updates.
     'table_property.binlog.level' = 'replica', -- You can also pass persistent Hologres table properties when creating a catalog. Binlog is then enabled by default when you create tables.
     'table_property.binlog.ttl' = '259200'
   );

   You need to modify the following parameter values with your actual Hologres service information.

   Parameter	Description	Notes
   endpoint	The endpoint of the Hologres instance.	Obtain the domain name for the specified VPC network type on the Hologres instance details page. For more information about domain names, see Endpoints.
   username	Select one of the following: The username for the custom account is in the format BASIC$<user_name>. The AccessKey ID of your Alibaba Cloud account or a RAM user.	The user configured here needs access to the corresponding Hologres database. For more information about Hologres database permissions and user management, see Hologres permission model and User management. This example uses a custom account named BASIC$flinktest and sets its password by using a project variable named holosecrect to avoid the security risks associated with plaintext storage. For more information, see Project variables.
   password	The password of the custom account. The AccessKey Secret of your Alibaba Cloud account or a RAM user.

   Note

   When you create a catalog, you can set default WITH parameters for source, dimension, and result tables. You can also set default properties for creating Hologres physical tables, such as the parameters starting with table_property. For more information, see Manage Hologres catalogs and Hologres connector WITH parameters.

5. Create a MySQL catalog.

   Copy the following code to the Script editor. Modify the target parameter values, select the target snippet, and then click Run on the left side of the code line. In the lower-right corner, ensure the session cluster you created is selected as the execution environment.

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

   You need to modify the following parameter values with your actual MySQL service information.

   Parameter	Description
   hostname	The IP address or hostname of the MySQL database. You can obtain the internal address by clicking View Details in the Network Type area on the database basic information page.
   port	The port number for the MySQL database service. The default value is 3306.
   username	The username for the MySQL database service.
   password	The password for the MySQL database service. This example uses a variable named mysql_pw for the password value to avoid risks such as plaintext storage. For more information, see Manage variables.

Build the ODS layer: Real-time data ingestion

You can build the ODS layer in one step by using the catalog's CREATE DATABASE AS (CDAS) statement. The ODS layer is not typically used directly for OLAP or serving key-value point queries. Instead, it serves as an event driver for streaming jobs, so enabling Binlog is sufficient. Binlog is a core capability of Hologres, and the Hologres connector supports a full and incremental mode that first reads all existing data and then incrementally consumes the Binlog.

1. Create a CDAS synchronization job named ODS.

   1. On the Data Development>ETL page, create a new SQL streaming job named ODS, and copy the following code into the SQL editor.

      CREATE DATABASE IF NOT EXISTS dw.order_dw   -- The table_property.binlog.level parameter was set when the catalog was created, so Binlog is enabled for all tables created via CDAS.
      AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- You can select which tables from the upstream database to ingest.
      /*+ OPTIONS('server-id'='8001-8004') */ ;   -- Specify the server-id range for the mysql-cdc instance.

      Note

      • By default, this example synchronizes data to the Public Schema of the order_dw database. You can also synchronize data to a specified schema in the target Hologres database. For more information, see Use as a target-side catalog for CDAS. After specifying a schema, the table name format for using the catalog will also change. For more information, see Use a Hologres catalog.

      • Schema changes to a source table are propagated to the result table only after a subsequent DML operation (INSERT, UPDATE, or DELETE) occurs on the source table.

   2. In the upper-right corner, click Deploy to deploy the job.

   3. In the left-side navigation pane, click O&M > Deployments. In the row for the ODS job you just deployed, click Start in the Actions column. Select Start Without State and then click Start.

2. Load data to the virtual warehouse.

   A table group is the data carrier in Hologres. When you use the read_warehouse_1 virtual warehouse to query data from a table group in the order_dw database, such as order_dw_tg_default (for the creation steps, see Table Group Management), you load order_dw_tg_default for read_warehouse_1. This allows you to use the init_warehouse virtual warehouse to write data and use the read_warehouse_1 virtual warehouse to perform service queries.

   On the HoloWeb development page, click SQL Editor, confirm the instance name and database name, and then run the following commands. For more information, see Create a new virtual warehouse instance. After loading, you can see that read_warehouse_1 has loaded the data from the order_dw_tg_default Table Group.

   -- View the Table Groups in the current database.
   SELECT tablegroup_name FROM hologres.hg_table_group_properties GROUP BY tablegroup_name;
   -- Load a Table Group into a virtual warehouse.
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   -- View the loading status of Table Groups for virtual warehouses.
   select * from hologres.hg_warehouse_table_groups;

3. In the upper-right corner, switch the virtual warehouse to read_warehouse_1. Subsequent queries and analyses will use the read_warehouse_1 virtual warehouse.

   The virtual warehouse drop-down list in the upper-right corner shows read_warehouse_1. The editor shows the executed table group loading statement CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1); and the query statement select * from hologres.hg_warehouse_table_groups;.

4. On the SQL Editor page, run the following commands to view the data synchronized from MySQL to the three Hologres tables.

   --- Query data in the orders table.
   SELECT * FROM orders;
   --- Query data in the orders_pay table.
   SELECT * FROM orders_pay;
   --- Query data in the product_catalog table.
   SELECT * FROM product_catalog;

   After you run the third query, the Result[3] tab shows five rows in the product_catalog table, including two columns: product_id (with values from 1 to 5) and catalog_name (with values phone_aaa, phone_bbb, phone_ccc, phone_ddd, and phone_eee).

Build the DWD layer: Create a real-time wide table

Building the DWD layer leverages the Hologres connector's unique partial-column update capability. You can use INSERT DML statements to express the semantics of partial-column updates. This process relies on high-performance point queries against dimension tables, which is provided by Hologres's row storage and hybrid row-column storage. At the same time, the strong resource isolation architecture of Hologres ensures that write, read, and analysis jobs do not interfere with each other.

1. Use the Flink catalog feature to create the DWD layer wide table dwd_orders in Hologres.

   On the Script tab of the Development > Scripts page, copy the following code into the script editor, select the snippet, and then click Run on the left side of the code line.

   -- Wide table fields must be nullable because when different streams write to the same result table, any column can potentially have a null value.
   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 'platform 0: phone, 1: pc', 
     pay_create_time timestamp,
     PRIMARY KEY(order_id) NOT ENFORCED
   );
   -- You can modify Hologres physical table properties through the catalog.
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' -- Change the Binlog timeout to one week.
   );

2. Consume the Binlog of the ODS layer tables orders and orders_pay in real time.

   On the Data Development>ETL page, create an SQL streaming job named DWD, copy the following code into the SQL editor, and then Deploy and Start the job. This SQL job joins the orders table with the product_catalog dimension table, writes the final result to the dwd_orders table, and performs real-time data enrichment.

   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. View the data in the dwd_orders wide table.

   On the HoloWeb development page, connect to the Hologres instance and log on to the target database. Then, run the following command in the SQL editor.

   SELECT * FROM dwd_orders;

   The dwd_orders wide table includes the following fields: order_id, order_user_id, order_shop_id, order_product_id, order_product_catalog_name, order_fee, order_create_time, order_update_time, order_state, pay_id, pay_platform, and pay_create_time. The query returns 7 order records.

Build the DWS layer: Calculate real-time metrics

1. Use the Flink catalog feature to create the DWS layer aggregation tables dws_users and dws_shops in Hologres.

   On the Script tab of the Development > Scripts page, copy the following code into the script editor, select the snippet, and then click Run on the left side of the code line.

   -- User-dimension aggregate metric table.
   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 'Total amount paid on the current day',
     primary key(user_id,ds) NOT ENFORCED
   );
   -- Shop-dimension aggregate metric table.
   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 'Total amount paid on the current day',
     primary key(shop_id,ds) NOT ENFORCED
   );

2. Consume data from the DWD wide table dw.order_dw.dwd_orders in real time, perform aggregations in Flink, and write the final results to the DWS tables in Hologres.

   On the Data Development>ETL page, create a new SQL streaming job named DWS, copy the following code to the SQL editor, and then Deploy and Start the job.

   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 -- Data from both order and payment streams has been written to the wide table.
       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 -- Data from both order and payment streams has been written to the wide table.
      GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   END;

3. View the aggregation results at the DWS layer. The results are updated in real time as the upstream data changes.

   1. In the Hologres console, view the data before the change.

      dws_users

      SELECT * FROM dws_users;

      The query result contains three columns: user_id (for example, user_001, user_002, user_003), ds (for example, 20230215), and paied_buy_fee_sum (for example, 8000.08, 5000.05). The user_id column is the key association field.

      dws_shops

      SELECT * FROM dws_shops;

      The query returns 4 records with three columns: shop_id, ds, and paied_buy_fee_sum. In the sample data, shop_id is 12345 to 12348, all ds values are 20230215, and the paied_buy_fee_sum values are 5000.05, 4000.04, 7000.07, and 2000.02, respectively.

   2. In the RDS console, insert a new data record into each of the orders and orders_pay tables of the order_dw database.

      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. In the Hologres console, view the data after the change.

      dwd_orders

      SELECT * FROM dwd_orders;

      The execution result shows 8 total order records in the dwd_orders table (order_id 100001–100008). Fields include order_user_id, order_shop_id, order_product_id, order_product_catalog_name, order_fee, order_create_time, and order_update_time. The newly inserted record for order_id=100008 has an order_fee of 6000.02.

      dws_users

      SELECT * FROM dws_users;

      The query returns 3 records from the dws_users table with columns for user_id, ds, and paied_buy_fee_sum. The data is: user_001 / 20230215 / 8000.08, user_002 / 20230215 / 5000.05, and user_003 / 20230215 / 11000.07. The aggregated amount for user_003 is the highest at 11000.07.

      dws_shops

      SELECT * FROM dws_shops;

      The query returns four rows with three columns: shop_id, ds, and paied_buy_fee_sum. The shop_id values are 12345, 12346, 12347, and 12348; all ds values are 20230215; and the paied_buy_fee_sum values are 11000.07, 4000.04, 7000.07, and 2000.02, respectively.

Data profiling

Because Binlog is enabled, you can directly inspect data changes. This solution persists data at each layer, which simplifies ad hoc data profiling and verifying the final results.

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'; -- Batch mode supports filter pushdown to improve the execution efficiency of batch jobs.

Point queries

Query the aggregated metric tables at the DWS layer based on the primary key, supporting millions of RPS.

The following is a code example for querying the consumption amount of a specific user on a specific date on the HoloWeb development page.

-- holo sql
SELECT * FROM dws_users WHERE user_id ='user_001' AND ds = '20230215';

In the query result, the value of the consumption amount field (paied_buy_fee_sum) is 8000.08.

OLAP analysis

Perform OLAP analysis on the DWD layer wide table.

The following is a code example for querying the order details of a specific customer on a specific payment platform in February 2023 on the HoloWeb development page.

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

After the query is executed, the result table displays the order detail records that match the filter criteria, showing fields such as order_id, order_user_id, order_shop_id, order_product_id, order_product_catalog_name, order_fee, order_create_time, and order_update_time.

Real-time reports

Display real-time reports based on data from the DWD layer wide table. Hologres's hybrid row-column storage and column-oriented tables provide excellent OLAP analysis capabilities, supporting responses in seconds.

The following is a code example for querying the total number and total amount of orders for each category in February 2023 on the HoloWeb development page.

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

After you execute the SQL, the Results tab displays four columns of data in a table: order_create_date, order_product_catalog_name, count, and sum. The example result for the date 20230215 shows the order counts (2, 1, 1, 2, and 2) and total amounts (6000.06, 4000.04, 3000.03, 4000.04, and 7000.03) for five product categories from phone_aaa to phone_eee.