实时计算Flink版：物化表快速入门（构建流批一体湖仓）

1. （可选）创建Paimon Catalog。

   基于Apache Paimon提供了物化表能力，需要创建元数据存储类型为Filesystem的Paimon Catalog，如果已经创建，可以跳过此步骤。更多详情请参见创建Paimon Catalog。

   创建Paimon Catalog

   1. 登录实时计算管理控制台。

   2. 单击目标工作空间操作列下的控制台。

   3. 在左侧导航栏选择数据管理，单击创建Catalog。选择Apache Paimon，单击下一步。

      

      参数说明：

      配置项	说明	备注
      metastore	元数据存储类型。	本示例采用filesystem为元数据存储类型。
      catalog name	Paimon Catalog名称。	请填写为自定义的英文名。本示例为paimon。
      warehouse	OSS服务中所指定的数仓目录。	格式为oss://<bucket>/<object>。其中： <bucket>：表示您创建的OSS Bucket名称。 <object>：表示您存放数据的路径。 请在OSS管理控制台上查看您的bucket和object名称。
      fs.oss.endpoint	OSS服务的连接地址。	如果Flink与OSS位于同一地域，则使用内网Endpoint，否则使用外网Endpoint。详情请参见OSS地域和访问域名。
      fs.oss.accessKeyId	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey ID。	获取方法请参见创建AccessKey。为了防止明文密钥泄露，建议使用变量写入，详情请参见变量管理。
      fs.oss.accessKeySecret	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey secret。

2. 创建用户行为日志表ods_user_log和商品信息表ods_dim_product。

   1. 登录实时计算管理控制台。

   2. 单击目标工作空间操作列下的控制台。

   3. 在左侧导航栏选择数据开发 > 数据查询。

      本样例已经创建名为paimon的Paimon Catalog，使用默认数据库default。

      CREATE TABLE `paimon`.`default`.`ods_user_log` (
        item_id INT NOT NULL,
        user_id INT NOT NULL,
        vtime TIMESTAMP(6),
        ds VARCHAR(10)
      ) 
      PARTITIONED BY(ds)
      WITH (
        'bucket' = '4',            --指定分桶数为4
        'bucket-key' = 'item_id'   --指定确定数据分桶的键。相同的item_id会被放到一个桶里。
      );
      
      CREATE TABLE `paimon`.`default`.`ods_dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
      ) WITH (
        'bucket' = '4',
        'bucket-key' = 'item_id'
      );

   4. 单击右上角运行，创建相应的数据表。

   5. 在左侧导航栏选择数据管理，单击对应的Paimon Catalog后，单击刷新查看新增表。

3. 使用模拟数据生成Faker连接器生成用户行为数据，并写入Paimon表中。

   1. 在左侧导航栏选择数据开发 > ETL。

   2. 单击新建，选择空白的流作业草稿，单击下一步，单击创建。

   3. 将如下SQL语句复制到SQL编辑器。

      CREATE TEMPORARY TABLE `user_log` (
        item_id INT,  //商品ID
        user_id INT,  //用户ID
        vtime TIMESTAMP,  
        ds AS DATE_FORMAT(CURRENT_DATE,'yyyyMMdd')
      ) WITH (
        'connector' = 'faker',    --faker连接器
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',    --生成0-1000其中一个随机数
        'fields.user_id.expression'='#{number.numberBetween ''0'',''100''}',
        'fields.vtime.expression'='#{date.past ''5'',''HOURS''}',           --基于当前日期时间前5小时生成数据
        'rows-per-second' = '3'   --每秒生成3条数据
       );
        
       CREATE TEMPORARY TABLE `dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
       ) WITH (
        'connector' = 'faker',    --faker连接器
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',
        'fields.title.expression'='#{book.title}',
        'fields.pict_url.expression'='#{internet.domainName}',
        'fields.brand_id.expression'='#{number.numberBetween ''1000'',''10000''}',   
        'fields.seller_id.expression'='#{number.numberBetween ''1000'',''10000''}',
        'rows-per-second' = '3'        --每秒生成3条数据
       );
      
      BEGIN STATEMENT SET; 
      
      INSERT INTO `paimon`.`default`.`ods_user_log` 
        SELECT 
        item_id,
        user_id,
        vtime,
        CAST(ds AS VARCHAR(10)) AS ds
      FROM `user_log`;
      INSERT INTO `paimon`.`default`.`ods_dim_product`
        SELECT 
        item_id,
        title,
        pict_url,
        brand_id,
        seller_id
      FROM `dim_product`;
      
      END; 

   4. 单击右上方的部署，进行作业部署。

   5. 单击左侧导航栏的运维中心 > 作业运维，单击目标作业操作列的启动，选择无状态启动后单击启动。

4. 查询模拟数据。

   在左侧导航栏选择数据开发 > 数据查询。将如下SQL语句复制到SQL编辑器后，单击右上角的运行。

   SELECT * FROM `paimon`.`default`.ods_dim_product LIMIT 10;
   
   SELECT * FROM `paimon`.`default`.ods_user_log LIMIT 10;

   

数据预览

可以预览物化表最新的100条数据。

1. 在左侧导航栏选择数据血缘，搜索目标物化表。

2. 单击目标物化表视图，在页面右下角单击详情。

3. 在物化表数据预览页签，单击查询图标。

   

数据查询

在左侧导航栏选择数据开发 > 数据查询，将如下SQL语句复制到SQL编辑器后，选中代码片段，单击运行，可以查询dws_overall物化表。

SELECT * FROM `paimon`.`default`.dws_overall ORDER BY hh;