OpenLake：Flink+Paimon+Hologres一体化实时湖仓分析

在DataWorks中新建Notebook

1. 登录DataWorks控制台。

2. 单击控制台左侧导航栏数据开发与治理 > 数据开发 > 进入数据开发（新版） 。进入DataWorks数据开发页面。

3. 单击左侧个人目录标题栏中图标，新建Notebooks。

   

4. 输入Notebook名称，按下回车键或单击页面空白位置，使Notebook名称生效。

5. 单击页面顶部个人开发环境 > 前往新建。

   

6. 进入新建实例面板，配置相关参数。

   参数名称	说明
   实例名称	自定义设置。
   资源组	选择资源组。如果没有对应资源组，您可以直接新建资源组。
   专有网络	选择与阿里云云服务位于同一地域和同一VPC。
   安全组	选择安全组。如果没有对应资源组，您可以直接新建安全组。
   交换机	选择交换机。如果没有对应交换机，您可以直接新建交换机。
   实例RAM角色	选择DataWorks默认角色。

7. 单击确定，完成实例创建。

8. 单击Notebook名称，即可打开并进入Notebook编辑页面，进行后续步骤。

使用Hologres在DLF中创建DB和表

1. 登录DataWorks控制台，完成Notebook新建。

2. 单击目标Notebook名称，进入编辑页面。

3. 单击，Cell类型选择Hologres SQL，选择上文绑定的计算资源。

   

4. 输入SQL，单击，创建External Database来映射上文创建的DLF Catalog。

   CREATE EXTERNAL DATABASE ext_db_dlf
   WITH 
     metastore_type 'dlf-paimon' 
     catalog_type 'paimon' 
     dlf_region 'cn-hangzhou' 
     dlf_endpoint 'dlfnext-share.cn-hangzhou.aliyuncs.com' 
     dlf_catalog 'clg-paimon-d24*********'; 
    --修改为自己在DLF中创建的Paimon Catalog的ID

5. 新建SQL Cell，在上文创建的EXTERNAL DATABASEext_db_dlf数据库中创建External Schema，创建后DLF Catalog可以看到名为github_events的Schema。

   CREATE EXTERNAL SCHEMA IF NOT EXISTS ext_db_dlf.github_events;

6. 新建SQL Cell，在上文创建的EXTERNAL SCHEMAgithub_events 下创建External Table，同样，DLF对应位置可以看到此表。

   CREATE EXTERNAL TABLE ext_db_dlf.github_events.gh_event_ods(
       id TEXT,                                        -- 每个事件的唯一ID。
       created_at BIGINT,                              -- 事件时间，单位秒。
       type TEXT,                                      -- GitHub事件类型，如：。ForkEvent, WatchEvent, IssuesEvent, CommitCommentEvent等。
       actor_id TEXT,                                  -- GitHub用户ID。
       actor_login TEXT,                               -- GitHub用户名。
       repo_id TEXT,                                   -- GitHub仓库ID。
       repo_name TEXT,                                 -- GitHub仓库名，如：apache/flink, apache/spark, alibaba/fastjson等。
       org TEXT,                                       -- GitHub组织ID。
       org_login TEXT,                                 -- GitHub组织名，如： apache,google,alibaba等。
       PRIMARY KEY(id)
   ) WITH (
     "changelog-producer"='input',
     "bucket"=6,
     "bucket-key"='id'
   );

使用Flink完成GitHub公开行为事件实时入湖

1. 登录Flink控制台。

2. 选择元数据管理 > 创建Catalog > Apache Paimon > 下一步。

3. 进入配置Catalog页面。单击确定完成创建。

   说明

   metastore：dlf。

   catalog name：上文创建的Paimon Catalog名称。

4. 登录DataWorks控制台。在目标Notebook中新建SQL Cell，输入SQL创建Flink作业来实时摄取SLS中的数据。

   说明

   为方便您更好地体验实时计算Flink版产品服务，本文代码示例中提供了仅有只读权限的AK供您试用，涉及的AK仅可用于本文示例中读取样例数据，真实数据生产环境请替换为您自己的数据源AK。

   -- @conf executionMode = STREAMING
   -- @conf name = flink_vvp_job_github_events
   -- @conf engineVersion=vvr-8.0.9-flink-1.17
   -- @conf flinkConf."execution.checkpointing.interval"=10second
   -- @conf flinkConf."execution.checkpointing.min-pause"=10second
   
   -- 创建一张临时表来实时摄取sls中的数据
   CREATE TEMPORARY TABLE sls_input (
       id STRING, -- 每个事件的唯一ID。
       created_at BIGINT, -- 事件时间，单位秒。
       type STRING, -- GitHub事件类型，如：ForkEvent, WatchEvent, IssuesEvent, CommitCommentEvent等。
       actor_id STRING, -- GitHub用户ID。
       actor_login STRING, -- GitHub用户名。
       repo_id STRING, -- GitHub仓库ID。
       repo_name STRING, -- GitHub仓库名，如：apache/flink, apache/spark, alibaba/fastjson等。
       org STRING, -- GitHub组织ID。
       org_login STRING -- GitHub组织名，如： apache,google,alibaba等。
   ) 
   WITH (
       'connector' = 'sls', -- 实时采集的GitHub事件存放在阿里云SLS中
       'project' = 'github-events-hangzhou', -- 存放公开数据的SLS项目。需改为您的实际地域信息
       'endPoint' = 'https://cn-hangzhou-intranet.log.aliyuncs.com', -- 公开数据仅限VVP通过私网地址访问。本示例以杭州为例，您需要修改为您的实际地域信息。
       'logStore' = 'realtime-github-events',           -- 存放公开数据的SLS logStore
       'accessId' =  'LTAI5tNF1rP8PKVyYjr9TKgh',        -- 访问公开数据集的只读AK，无需替换     
       'accessKey' = 'FDgsh6fjSmkbFsx083tN6HOiqNVWTP'   -- 访问公开数据集的只读secret，无需替换
   );
   
   -- 将临时表数据写入Hologres中创建的Paimon格式的External Table
   
   INSERT INTO ext_db_dlf.github_events.gh_event_ods
   SELECT * FROM sls_input 
   WHERE id IS NOT NULL 
   AND created_at IS NOT NULL 
   AND TO_TIMESTAMP(created_at * 1000) >= date_add(CURRENT_DATE,-1);

5. 新建SQL Cell，Cell类型选择Hologres SQL，输入SQL查询Flink实时写入的数据。

   -- 查看数据
   SELECT * FROM ext_db_dlf.github_events.gh_event_ods limit 10;

使用Hologres Dynamic Table完成湖上数据实时入仓

Dynamic Table是一种声明式数据处理架构，可以自动处理并存储一个或者多个基表对象的数据聚合结果，内置不同的数据刷新策略，业务可以根据业务需求设置不同的数据刷新策略，实现数据从基表对象到Dynamic Table的自动流转，满足业务统一开发、数据自动流转、处理时效性等诉求。

DROP TABLE IF EXISTS hologres_github_event ;

CREATE DYNAMIC TABLE hologres_github_event 
WITH (
    auto_refresh_enable='true',
    refresh_mode = 'incremental',
    incremental_auto_refresh_schd_start_time='immediate',
    incremental_auto_refresh_interval = '1 minute',
    clustering_key='created_at_tz',
    distribution_key='id'  
) AS
SELECT 
    id::BIGINT ,
    TO_TIMESTAMP(created_at)::TIMESTAMPTZ AS created_at_tz,
    type::TEXT,                                      -- GitHub事件类型，如：。ForkEvent, WatchEvent, IssuesEvent, CommitCommentEvent等。
    actor_id::TEXT,                                  -- GitHub用户ID。
    actor_login::TEXT,                               -- GitHub用户名。
    repo_id::TEXT,                                   -- GitHub仓库ID。
    repo_name::TEXT,                                 -- GitHub仓库名，如：apache/flink, apache/spark, alibaba/fastjson等。
    org::TEXT,                                       -- GitHub组织ID。
    org_login::TEXT,
    to_char(TO_TIMESTAMP(created_at),'YYYY-MM-DD')::text AS ds
FROM ext_db_dlf.github_events.gh_event_ods;

对于仓内数据进行聚合分析

登录DataWorks控制台。在目标Notebook编辑页面单击+SQL，Cell类型选择Hologres SQL，输入SQL来分析GitHub行为事件。

• 查询今日最活跃项目。

  SELECT
      repo_name,
      COUNT(*) AS events
  FROM
      hologres_github_event
  WHERE
      created_at_tz >= CURRENT_DATE 
  GROUP BY
      repo_name
  ORDER BY
      events DESC
  LIMIT 10;

• 查询1天内最活跃（事件数最多）的几位开发者。

  SELECT
      actor_login,
      COUNT(*) AS events
  FROM
      hologres_github_event
  WHERE
      created_at_tz >= now() - interval '1 day'
      AND actor_login NOT LIKE '%[bot]'
  GROUP BY
      actor_login
  ORDER BY
      events DESC
  LIMIT 10;

• 查询过去1天最多的10项操作。

  SELECT
      type,
      count(*) total
  FROM
      hologres_github_event
  WHERE
      created_at_tz > now() - interval '1 day'
  GROUP BY
      1
  ORDER BY
      total DESC
  LIMIT 10;

• 查看当天漏斗转化情况。

  CREATE EXTENSION IF NOT EXISTS flow_analysis; --开启Extension
  
  -- 查看当天的转换漏斗
  WITH level_detail AS (
      SELECT
          level,
          COUNT(1) AS count_user
      FROM (
          SELECT
              actor_id,
              windowFunnel (1800, 'default', created_at_tz, type = 'CreateEvent', type = 'PushEvent',type = 'IssuesEvent') AS level
          FROM
              hologres_github_event
          WHERE
              created_at_tz >= now() - interval '1 day'
              AND created_at_tz < now()
          GROUP BY
              actor_id) AS basic_table
      GROUP BY
          level
      ORDER BY
          level ASC
  )
  SELECT  CASE level    WHEN 0 THEN 'total'
                        WHEN 1 THEN 'CreateEvent'
                        WHEN 2 THEN 'PushEvent'
                        WHEN 3 THEN 'IssuesEvent'
                        END
          ,SUM(count_user) OVER ( ORDER BY level DESC )
  FROM
      level_detail
  GROUP BY
      level,
      count_user
  ORDER BY
      level ASC;