Flink+Hologres搭建实时数仓

1.创建专有网络VPC和交换机

1. 登录专有网络管理控制台。

2. 在顶部菜单栏，选择华东1（杭州）地域（本方案以杭州地域为例）。

3. 在左侧导航栏，单击专有网络，单击创建专有网络。

4. 在创建专有网络页面，按照资源配置创建1个专有网络（VPC）和1台交换机，然后单击确定。详情请参见创建专有网络和交换机。

   资源配置

   规划项	说明	方案默认示例
   专有网络VPC
   VPC名称	建议您在部署过程中新建一个VPC作为本方案的专有网络。部署过程中填写VPC名称即可创建对应名称的VPC。 长度为2~128个字符，以英文大小字母或中文开头，可包含数字、下划线（_）和连字符（-）。	vpc_flink
   网段	在创建VPC时，您必须按照无类域间路由块（CIDR block）的格式为您的专有网络划分私网网段。阿里云VPC支持的网段信息请参见专有网络组成部分。 在网络规划时可以按照管理网段-开发网段-测试网段-生产网段等规则做好规划。网段一旦投入使用，调整过程复杂，因此规划十分重要。	192.168.0.0/16
   虚拟交换机
   台数	本方案需要至少1台交换机，用来连接云资源实例。	1
   名称	建议您在部署过程中在新建的VPC内创建虚拟交换机。部署过程中填写vswitch名称即可创建对应名称的虚拟交换机。 长度为2~128个字符，以英文大小字母或中文开头，可包含数字、下划线（_）和连字符（-）。	vsw_001
   可用区	在规划的地域内选择1个可用区，用于部署虚拟交换机。	可用区J
   IPv4网段	每台虚拟交换机需要一个IPv4网段。	192.168.5.0/16

2.创建RDS MySQL实例

1. 访问RDS实例创建页面。

2. 按照资源配置完成相应参数配置后，根据页面提示完成支付。

   资源配置

   规划项	说明	方案默认示例
   计费方式	RDS实例的计费方式。详情请参见快速创建RDS MySQL实例。	按量付费
   地域	您的云服务部署的地域。选择地域的基本原则请参见地域和可用区。	华东1（杭州）
   引擎	引擎及版本。	MySQL8.0
   产品系列	实例系列。	高可用系列
   主节点可用区	本方案要求RDS实例与虚拟交换机位于同一可用区。	杭州 可用区J
   部署方案	单可用区指主备节点位于同一可用区。多可用区指主备节点位于不同可用区，实现跨可用区容灾。请您根据实际需求进行选择。	单可用区部署
   实例规格	可通过测试业务负载来确认需要的实例规格。	mysql.n2.small.2c
   VPC	选择上述已创建的VPC。	vpc_flink
   主节点交换机	选择上述已创建的交换机。	vsw_001

3. 为目标实例创建数据库order_dw和具有对应数据库读写权限的普通账号。具体操作请参见创建数据库和账号。

4. 准备MySQL CDC数据源。

   1. 单击上方的登录数据库。

   2. 在弹出的DMS页面中，填写在步骤3中创建的数据库账号名和密码，然后单击登录。

   3. 登录成功后，在左侧双击order_dw数据库，切换数据库。

   4. 在SQL Console区域编写三张业务表的建表DDL以及插入的数据语句。

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

   5. 单击执行，单击直接执行。

3.创建独享Hologres实例

1. 进入Hologres产品详情页，单击立即购买。

2. 选择商品类型，按照资源配置完成相应参数配置后，根据页面提示完成开通。

   资源配置

   规划项	说明	方案默认示例
   商品类型	Hologres实例的计费方式。计费的详细信息请参见计费概述。	独享实例（按量付费）
   地域	选择与上述服务相同的地域。	华东1（杭州）
   实例类型	计算组实例是主从实例的升级模式，支持将计算资源分解为不同的计算组，可支撑读写分离、资源隔离、业务隔离等场景，详情请参见计算组实例快速入门。	计算组型
   可用区	本方案要求Hologres实例与虚拟交换机位于同一可用区。	可用区J
   计算组预留计算资源	需要确保预留资源大于等于实例所有计算组的总计算资源。计算组默认最小资源为32 CU，本方案需要2个计算组。	64 CU
   专有网络（VPC）	选择上述已创建的VPC。	vpc_flink
   专有网络交换机	选择上述已创建的交换机。	vsw_001
   实例名称	自定义填写。	hologres_test

3. 创建成功后，单击实例名称，单击右上方的登录实例，进入HoloWeb页面。

4. 创建名为order_dw的数据库，并为用户授权（赋予admin权限），使用简单权限模型创建数据库，详情请参见DB管理。

   说明

   如果在被授权账号的下拉列表找不到对应的账号，则说明该账号并未添加至当前实例，您需要前往用户管理页面添加用户为SuperUser。

5. 新增计算组。

   您可以通过不同计算组实现资源隔离，使用初始计算组init_warehouse用于写入数据，使用read_warehouse_1计算组用于服务查询。

   预留计算资源会全部分配给初始计算组init_warehouse，需先减少计算组资源，再新增计算组。详情请参见场景1：创建全新的计算组实例。

   1. 单击安全中心 > 计算组管理，确认实例名为目标实例名称。

   2. 单击已有资源组init_warehouse操作列下的调整配置，本示例减少资源至32 CU，单击确认。

   3. 单击新增计算组，新增名称为read_warehouse_1，资源为32 CU的计算组，单击确认。

4.创建OSS Bucket

用于存储Flink作业系统检查点、作业快照、日志和JAR包等信息，详情请参见开通实时计算Flink版。您可以创建新的OSS Bucket用于体验本方案，体验完毕后删除。

1. 登录OSS管理控制台，单击Bucket 列表，然后单击创建 Bucket。

2. 在创建页面，填写Bucket 名称（本方案默认示例flink-test-oss），选择您的云服务部署的地域（本方案默认示例华东1（杭州）），单击确定。

5.开通Flink全托管

1. 登录实时计算控制台。单击立即购买。

2. 在购买页面，按照资源配置完成相应参数配置后，根据页面提示完成开通。开通详情请参见开通实时计算Flink版。

   资源配置

   规划项	说明	方案默认示例
   付费模式	Flink工作空间的计费方式。计费详情请参见计费概述。	按量付费
   地域	选择与上述服务相同的地域。	华东1（杭州）
   可用区	本方案要求Flink工作空间与虚拟交换机位于同一可用区。	可用区J
   专有网络	选择上述已创建的VPC。	vpc_flink
   虚拟交换机	选择上述已创建的交换机。	vsw_001
   工作空间名称	开通成功后，不可修改名称。	flink-test
   OSS存储	用于存储作业系统检查点、作业快照、日志和JAR包等信息。需要与Flink工作空间服务在同一地域，且Flink开通成功后，不可修改OSS存储。详情请参见开通实时计算Flink版。	flink-test-oss
   监控服务	支持选择免费监控服务或按量付费的Prometheus监控服务。具体信息和计费说明请参见开通实时计算Flink版。	免费监控服务

1.创建Catalog

在Flink开发控制台的SQL开发页面，通过SQL方式分别创建Hologres和MySQL的Catalog。

1. 登录实时计算控制台，单击目标工作空间操作列下的控制台。

2. 在左侧导航栏，单击Session管理，单击创建Session集群，为后续创建Catalog和查询脚本提供执行环境，详情请参见Session集群注意事项。

   1. 填写名称，并将状态设置为Running。

   2. 选择引擎版本，本方案示例选择最新版本。

   3. 其他参数本方案示例保持默认，单击创建Session集群。

3. 在左侧导航栏，单击SQL开发，单击查询脚本页签，单击新建。

   

4. 创建Hologres Catalog。

   在文本编辑区域，输入右侧配置Hologres Catalog的命令，创建名称为dw的Hologres Catalog。其中，您需要修改以下参数取值为您实际Hologres服务信息。关于Hologres Catalog的更多信息，详情请参见管理Hologres Catalog。

   参数	说明	备注
   endpoint	Hologres的Endpoint地址。	本示例使用指定VPC域名，详情请参见实例配置。
   username	阿里云账号的AccessKey。	当前配置的AccessKey对应的用户需要能够访问所有的Hologres数据库，Hologres数据库权限请参见Hologres权限模型概述。
   password	阿里云账号的AccessSecret。

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = '<USERNAME>',
     'password' = '<PASSWORD>',
     'dbname' = 'order_dw@init_warehouse', --数据库名称,并指定连接init_warehouse计算组
     'binlog' = 'true', -- 创建catalog时可以设置源表、维表和结果表支持的with参数，之后在使用此catalog下的表时会默认添加这些默认参数。
     'sdkMode' = 'jdbc', -- 推荐使用jdbc模式。
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- 宽表merge需要开启，防止回撤。
     'partial-insert.enabled' = 'true', -- 宽表merge需要开启此参数，实现部分列更新。
     'mutateType' = 'insertOrUpdate', -- 宽表merge需要开启此参数，实现部分列更新。
     'table_property.binlog.level' = 'replica', --也可以在创建catalog时传入持久化的hologres表属性，之后创建表时，默认都开启binlog。
     'table_property.binlog.ttl' = '259200'
   );

5. 选中创建Catalog的代码后，单击左侧代码行数上的运行。

   返回The following statement has been executed successfully!表示创建成功。

6. 创建MySQL Catalog。

   输入右侧的命令创建MySQL Catalog，您需要修改表格中的参数取值为您实际的MySQL服务信息。选中创建MySQL Catalog的代码后，单击左侧代码行数上的运行。关于MySQL Catalog的更多信息，详情请参见管理MySQL Catalog。

   参数	说明
   hostname	MySQL数据库的IP地址或者Hostname。
   port	MySQL数据库服务的端口号，默认值为3306。
   username	MySQL数据库服务的用户名。 本示例为具有数据库order_dw读写权限的账号名。
   password	MySQL数据库服务的密码。

   CREATE CATALOG mysqlcatalog WITH(
     'type' = 'mysql',
     'hostname' = '<hostname>',
     'port' = '<port>',
     'username' = '<username>',
     'password' = '<password>',
     'default-database' = 'order_dw'
   );

2.构建ODS层：业务数据库实时入仓

基于Catalog的CREATE DATABASE AS（CDAS）语句功能，可以一次性把ODS层建出来。ODS层一般不直接做OLAP或SERVING（KV点查），主要作为流式作业的事件驱动，开启binlog即可满足需求。Binlog是Hologres的核心能力之一，Hologres连接器也支持先全量读取再增量消费Binlog的全增量模式。

1. 创建CDAS同步作业。

   1. 在实时计算控制台上，创建新的SQL作业草稿，并将如下代码拷贝到SQL编辑器。

      CREATE DATABASE IF NOT EXISTS dw.order_dw   -- 创建catalog时设置了table_property.binlog.level参数，因此通过CDAS创建的所有表都开启了binlog。
      AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- 可以根据需要选择上游数据库需要入仓的表。本案例需选择前面创建了三张业务表的数据库。
      /*+ OPTIONS('server-id'='8001-8004') */ ;   -- 指定mysql-cdc源表。

   2. 单击右上方的部署，单击确定，进行作业部署。

   3. 单击左侧导航栏的作业运维，单击刚刚部署的ODS作业操作列的启动，选择无状态启动启动作业。

2. 向计算组加载数据。

   Table Group是Hologres中数据的载体。使用read_warehouse_1查询order_dw数据库中Table Group（本示例为order_dw_tg_default）的数据时，为计算组read_warehouse_1加载order_dw_tg_default，以实现使用init_warehouse计算组写入数据，使用read_warehouse_1计算组进行服务查询。

   在HoloWeb开发页单击SQL编辑器，确认实例名和数据库名称后，执行右侧命令。更多详情请参见场景1：创建全新的计算组实例。加载后，可以查看到read_warehouse_1已经加载了order_dw_tg_default Table Group的数据。

   --查看当前数据库有哪些Table Group
   SELECT tablegroup_name FROM hologres.hg_table_group_properties GROUP BY tablegroup_name;
   
   --为计算组加载Table Group
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   
   --查看计算组加载Table Group的情况
   select * from hologres.hg_warehouse_table_groups;

3. 右上角切换计算组为read_warehouse_1，后续使用read_warehouse_1进行查询分析。

   

4. 在HoloWeb中执行右侧命令，查看MySQL同步到Hologres的3张表数据。

   ---查orders中的数据。
   SELECT * FROM orders;
   
   ---查orders_pay中的数据。
   SELECT * FROM orders_pay;
   
   ---查product_catalog中的数据。
   SELECT * FROM product_catalog;

   

3.构建DWD层：实时主题宽表

1. 通过Flink Catalog功能在Hologres中创建DWD层的宽表dwd_orders。

   将代码拷贝到Flink的查询脚本中，选中目标片段后单击左侧代码行上的运行。

   -- 宽表字段要nullable，因为不同的流写入到同一张结果表，每一列都可能出现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
   );
   
   -- 支持通过catalog修改Hologres物理表属性。
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' --修改binlog的超时时间为一周。
   );

2. 实现实时消费ODS层orders、orders_pay表的binlog。

   创建新的Flink 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开发页面的SQL编辑器执行右侧所示命令。

   SELECT * FROM dwd_orders;

   

4.构建DWS层：实时指标计算

1. 通过Flink Catalog功能，在Hologres中创建DWS层的聚合dws_users以及dws_shops。

   将代码拷贝到Flink的查询脚本中，选中目标片段后单击左侧代码行上的运行。

   -- 用户维度聚合指标表。
   CREATE TABLE dw.order_dw.dws_users (
     user_id string not null,
     ds string not null comment 'pay_create_date',
     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 comment 'pay_create_date',
     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表。

   创建新的Flink 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. 在HoloWeb开发页面的SQL编辑器执行所示命令，查询dws_users表结果，dws_shops表结果查询类似。

   SELECT * FROM dws_users;

   

   2. 在RDS控制台向orders和orders_pay表中分别插入1条新数据，Hologres控制台查看变更后的表结果（以dwd_orders为例）。

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