Realtime Compute for Apache Flink:Build a Hologres real-time data warehouse

Create an RDS for MySQL instance and prepare data

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

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

2. Create a database and an account.

   For the target instance, create a database named order_dw and a standard account with read and write permissions on the database. For more information, 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. On the logon page, enter the username and password for the database account that you created, and then click Log On.

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

   4. In the SQL Console, enter the following DDL statements to create the business tables and the INSERT statements to populate them.

      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 Execute, and then click Direct Execution.

Create a Hologres instance and compute groups

1. Purchase a dedicated Hologres instance. For more information, see Purchase a Hologres instance.

   The Hologres instance must be in the same VPC as the ApsaraDB RDS for MySQL instance. To experience strong resource isolation through read/write splitting, this example uses the Virtual Warehouse instance type, and sets Reserved Compute Resource to 64 so that you can create additional compute groups.

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

   Create a database named order_dw (with the simple permission model enabled), and grant the user administrative privileges. For details about database management and authorization, see Manage databases.

   Note

   • If you cannot find the account in the User drop-down list, the account has not been added to the instance. Go to the User Management page and add the user as a SuperUser.

   • In Hologres V2.0 and later, the binary logging extension is enabled by default. You do not need to enable it manually.

3. Create a new compute group.

   You can use different compute groups to isolate resources. Use the initial compute group init_warehouse for data writes and the read_warehouse_1 compute group for serving queries.

   All reserved computing resources are allocated to the initial compute group init_warehouse by default. You must first reduce its resources before you can create a new compute group. For more information, see Create a new compute group instance.

   1. Go to Security Center > Compute Group Management and confirm the instance name.

   2. In the row for the init_warehouse compute group, click Modify Configuration in the Actions column. Reduce the allocated resources and click OK.

   3. Click Create Compute Group, create a new compute group named read_warehouse_1, and then 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 ApsaraDB RDS for MySQL and Hologres instances.

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

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

4. Create a Hologres catalog.

   On the Development > Scripts page, in the Scripts tab, copy the following code, replace the placeholder values, select the code, and then click Run. This uses the session cluster you created as the execution environment.

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = 'BASIC$flinktest',
     'password' = '${secret_values.holosecrect}',
     'dbname' = 'order_dw@init_warehouse', -- Specify the database name and connect to the init_warehouse compute group.
     'binlog' = 'true', -- You can set default WITH options for source, dimension, and result tables when creating the catalog. Tables created under this catalog inherit these defaults.
     'sdkMode' = 'jdbc', -- The jdbc mode is recommended.
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- Required for wide-table merge to prevent retractions.
     'partial-insert.enabled' = 'true', -- Required for wide-table merge to enable partial column updates.
     'mutateType' = 'insertOrUpdate', -- Required for wide-table merge to enable partial column updates.
     'table_property.binlog.level' = 'replica', -- You can also pass persistent Hologres table properties when creating the catalog. Tables created later will have binary logging enabled by default.
     'table_property.binlog.ttl' = '259200'
   );

   Modify the following parameters with your actual Hologres service information.

   Parameter	Description	Notes
   endpoint	The endpoint of your Hologres instance.	On the Hologres instance details page, obtain the domain name for the specified VPC. For more information about domain names, see Endpoints.
   username	Select one of the following: The username for a custom account must be formatted as BASIC$<user_name>. The AccessKey ID of your Alibaba Cloud account or RAM user.	The configured user must have access to the corresponding Hologres database. For details, see Hologres permission model and Manage users. 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 of storing passwords in plaintext. For more information, see Project variables.
   password	The password of the custom account. The AccessKey secret of your Alibaba Cloud account or RAM user.

   Note

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

5. Create a MySQL catalog.

   Copy the following code into the Scripts tab, modify the parameter values, select the code, and then click Run. This uses the session cluster you created 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'
   );

   Modify the following parameters with your actual MySQL service information.

   Parameter	Description
   hostname	The IP address or hostname of your MySQL database. On the database basic information page, click View Connection Details in the Network Type area to obtain the internal endpoint.
   port	The port number of your MySQL database service. The default value is 3306.
   username	The username for your MySQL database service.
   password	The password for your MySQL database service. This example uses a variable named mysql_pw for the password to avoid plaintext exposure. For more information, see Manage variables.

Build the ODS layer: Ingest business data

With the catalog-based CREATE DATABASE AS (CDAS) statement, you can create the ODS layer in one step. The ODS layer typically serves as an event source for streaming jobs rather than directly for OLAP queries or point queries. Enabling binary logging is sufficient for this purpose. Binary logging is a core Hologres capability. The Hologres connector also supports a full-plus-incremental mode: it reads a full snapshot first and then consumes binary logs incrementally.

1. Create the ODS CDAS synchronization job.

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

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

      Note

      • By default, this example syncs data to the Public schema of the order_dw database. You can also sync data to a specified schema in the target Hologres database. For more information, see Use a Hologres catalog as the destination in a CREATE DATABASE AS... statement. After you specify a schema, the table name format changes when you use the catalog. For details, see Use a Hologres catalog.

      • If the schema of a source table changes, the schema of the result table will not be updated until a data change (delete, insert, or update) occurs in the source table.

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

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

2. Load data into the compute group.

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

   On the HoloWeb development page, click SQL Editor. Confirm the instance name and database name, and then execute the following commands. For more information, see Create a new compute group instance. After loading, you can see that read_warehouse_1 has loaded the data from the order_dw_tg_default table group.

   -- List 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 compute group.
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   -- Check the table groups loaded into the compute group.
   select * from hologres.hg_warehouse_table_groups;

3. In the upper-right corner, switch the compute group to read_warehouse_1. Subsequent queries and analyses will use this compute group.

   In the upper-right corner of the HoloWeb page, select read_warehouse_1 from the compute group drop-down list.

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

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

   The query result for the product_catalog table contains two columns, product_id (1 to 5) and catalog_name (phone_aaa, phone_bbb, phone_ccc, phone_ddd, phone_eee), with a total of 5 records. This indicates that the data is successfully synchronized to Hologres.

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

This step uses the Hologres connector’s partial column update capability. You can express partial updates with INSERT DML. The job queries multiple dimension tables by using high-performance point queries, enabled by Hologres row storage and hybrid row-column storage. With strong resource isolation, write, read, and analytics workloads 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 Development > Scripts page, copy the following code into the Scripts tab, select the code, and then click Run.

   -- Wide table columns must be nullable because different streams write to the same result table, and any column can be 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 '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 binary log TTL to one week.
   );

2. Consume binary logging changes from the ODS layer tables orders and orders_pay in real time.

   On the Development > ETL page, create a new SQL stream draft named DWD. Copy the following code into the SQL editor, and then Deploy and Start the job. This SQL job joins the orders and product_catalog tables by using a temporal join and writes the result to the dwd_orders table, enriching the data in real time.

   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.

   Connect to the Hologres instance on the HoloWeb development page, log on to the target database, and then execute the following command in the SQL Editor.

   SELECT * FROM dwd_orders;

   After successful execution, the query result returns data from the dwd_orders wide table, including 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.

Build the DWS layer: Calculate real-time metrics

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

   On the Development > Scripts page, copy the following code into the Scripts tab, select the code, and then click Run.

   -- User-dimension aggregate 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 of payments completed on the day',
     primary key(user_id,ds) NOT ENFORCED
   );
   -- Shop-dimension aggregate 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 of payments completed on the day',
     primary key(shop_id,ds) NOT ENFORCED
   );

2. Consume the DWD layer 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 Development > ETL page, create a new SQL stream draft named DWS. Copy the following code into 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 -- Both order and payment stream data have 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 -- Both order and payment stream data have been written to the wide table.
      GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   END;

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

   1. View the data in the Hologres console before the change.

      dws_users table

      SELECT * FROM dws_users;

      After running the query, the result returns data from the dws_users table, which includes three columns: user_id, ds, and paied_buy_fee_sum. In the example result, the user_id column contains the values user_001, user_002, and user_003; the ds column contains the value 20230215; and the paied_buy_fee_sum column contains the values 8000.08, 5000.05, and 5000.05, respectively. The user_id column uniquely identifies each user.

      dws_shops table

      SELECT * FROM dws_shops;

      The query result shows that the dws_shops table contains three columns: shop_id (shop ID), ds (date partition), and paied_buy_fee_sum (payment amount). Four rows of sample data are returned, confirming that the DWS layer table has been successfully built.

   2. In the RDS console, insert one new record into each of the orders and orders_pay tables in 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. View the updated data in the Hologres console.

      dwd_orders table

      SELECT * FROM dwd_orders;

      After you run the query, eight order records are returned from the dwd_orders table, including 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. The eighth record (order_id=100008, user_003, phone_eee, 6000.02) is the newly written data.

      dws_users table

      SELECT * FROM dws_users;

      The query returns three rows with three columns: user_id, ds, and paied_buy_fee_sum: user_001 / 20230215 / 8000.08, user_002 / 20230215 / 5000.05, and user_003 / 20230215 / 11000.07. The highest total payment amount is for user_003 (11000.07).

      dws_shops table

      SELECT * FROM dws_shops;

      After the query is run, the result contains three columns, shop_id, ds, and paied_buy_fee_sum, with four rows of data showing the fee values (11000.07, 4000.04, 7000.07, and 2000.02) for shops 12345, 12346, 12347, and 12348 on 20230215. The shop_id and paied_buy_fee_sum are the key metric columns.

Profile the data

Because binary logging is enabled, you can directly inspect data changes. If you need to perform ad-hoc business data exploration on intermediate results or verify the correctness of the final computation, each layer of this solution is persisted, making it easy to examine the intermediate process.

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 batch job execution efficiency.

Point query

Query the DWS layer aggregated metric tables by primary key, with support for millions of RPS.

On the HoloWeb development page, run the following SQL to query the consumption amount for a specific user on a specific date.

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

The query result returns three columns: user_id, ds, and paied_buy_fee_sum (consumption amount). The consumption amount for user_001 on 20230215 is 8000.08.

OLAP query

Run OLAP queries on the DWD layer wide table.

On the HoloWeb development page, run the following SQL to query the order details for a specific customer on a specific payment platform in February 2023.

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

The query result returns two order records with fields such as order_id, order_user_id, order_shop_id, order_product_id, order_fee, order_create_time, and order_update_time. The example order IDs are 100006 and 100004.

Real-time reports

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

On the HoloWeb development page, run the following SQL to query the total number of orders and the total order amount for each product category in February 2023.

-- 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 run the SQL statement, the Result tab displays a table with four columns: order_create_date, order_product_catalog_name, count, and sum. For example, the results for the date 20230215 show that the phone_aaa category had 2 orders totaling 6000.06, the phone_bbb category had 1 order for 4000.04, and so on for all five categories.