GitHub近实时数据同步以及增全量数据一体化分析

步骤

操作

预期结果

Step 0

注册MaxCompute新功能邀测申请表单

1. 打开MaxCompute新功能邀测申请表单。

2. 选中Delta Table增量表格式、基于增量物化视图的增量计算、MCQA2.0 SQL引擎查询加速。

3. 单击注册。

开通MaxCompute新功能。

Step 1

初始化MaxCompute新项目

1. 打开MaxCompute控制台，单击新建项目。

2. 在新建项目对话框中配置如下参数。

   项目名称设置为delta_compute_yunqi，项目类型选择云栖新功能邀测项目，其他参数按需设置或保持默认。

   地域可选择北京、上海或杭州。

3. 并单击确定。

1. 在控制台左侧导航栏中，选择工作区 > SQL分析，打开临时SQL分析页面。

2. 在SQL分析页面左侧，可以看到新建项目delta_compute_yunqi。

Step 2

配额（Quota）管理上新购Quota

1. 在MaxCompute控制台左侧导航栏中，选择工作区 > 配额（Quota）管理，单击新购Quota。

2. 创建一个包年包月的一级Quota，并根据页面提示完成支付。

3. 在已创建的一级Quota组下，创建2个不同二级Quota组：

   • 批处理Quota组：batch_demo_yunqi

   • 交互式Quota组：mcqa2_demo_yunqi

   

4. 分别设置批处理Quota组和交互式Quota组资源。

在配额（Quota）管理页面，确认批处理Quota组和交互式Quota组已成功创建。

Step 3

新建MaxCompute Delta Table增量表

在SQL分析页面，执行如下示例命令，创建2个MaxCompute内部表：

• 创建yunqi_github_events_odps_cdc表，后续会将数据实时通过FlinkCDC写入至该表中。

  --设置ODPS SQL支持upsertable
  SET odps.sql.upsertable.table.enable=true;
  SET odps.storage.orc.enable.memcmp.sort.key=true; 
  
  --FlinkCDC
  CREATE TABLE IF NOT EXISTS yunqi_github_events_odps_cdc
  (
      id                     BIGINT NOT NULL COMMENT '事件ID'
      ,actor_id              BIGINT COMMENT '事件发起人ID'
      ,actor_login           STRING COMMENT '事件发起人登录名'
      ,repo_id               BIGINT COMMENT 'repoID'
      ,repo_name             STRING COMMENT 'repo全名：owner/Repository_name'
      ,org_id                BIGINT COMMENT 'repo所属组织ID'
      ,org_login             STRING COMMENT 'repo所属组织名称'
      ,`type`                STRING COMMENT '事件类型'
      ,created_at            TIMESTAMP NOT NULL COMMENT '事件发生时间'
      ,action                STRING COMMENT '事件行为'
      ,commit_id             STRING COMMENT '提交记录ID'
      ,member_id             BIGINT COMMENT '成员ID'
      ,language              STRING COMMENT '编程语言'
      ,PRIMARY KEY(id, created_at))
  STORED AS ALIORC  
  TBLPROPERTIES ('acid.data.retain.hours'='24',
       'acid.incremental.query.out.of.time.range.enabled'='true',
       'columnar.nested.type'='true',
       'transactional'='true',
       'write.bucket.num'='64',
       'acid.ingest.commit.num.check.limit'='10',
       'cdc.insert.into.passthrough.enable'='true', 
       'acid.cdc.mode.enable' = 'true',--打开异步的话，就要额外设置
       'acid.cdc.build.async'='true', 
       'acid.cdc.build.interval'='60') 
  LIFECYCLE 36500;
  

• 创建yunqi_github_events_odps_dw表，后续会将数据实时通过DataWorks数据集成写入至该表中。

  --设置ODPS SQL支持upsertable
  SET odps.sql.upsertable.table.enable=true;
  SET odps.storage.orc.enable.memcmp.sort.key=true; 
  
  --DataWorks实时数据同步
  CREATE TABLE IF NOT EXISTS yunqi_github_events_odps_dw
  (
      id                     BIGINT NOT NULL COMMENT '事件ID'
      ,actor_id              BIGINT COMMENT '事件发起人ID'
      ,actor_login           STRING COMMENT '事件发起人登录名'
      ,repo_id               BIGINT COMMENT 'repoID'
      ,repo_name             STRING COMMENT 'repo全名：owner/Repository_name'
      ,org_id                BIGINT COMMENT 'repo所属组织ID'
      ,org_login             STRING COMMENT 'repo所属组织名称'
      ,`type`                STRING COMMENT '事件类型'
      ,created_at            TIMESTAMP COMMENT '事件发生时间'
      ,action                STRING COMMENT '事件行为'
      ,commit_id             STRING COMMENT '提交记录ID'
      ,member_id             BIGINT COMMENT '成员ID'
      ,language              STRING COMMENT '编程语言'
      ,PRIMARY KEY(id))
  STORED AS ALIORC  
  TBLPROPERTIES ('acid.data.retain.hours'='24',
       'acid.incremental.query.out.of.time.range.enabled'='true',
       'columnar.nested.type'='true',
       'transactional'='true',
       'write.bucket.num'='64',
       'acid.ingest.commit.num.check.limit'='10',
       'cdc.insert.into.passthrough.enable'='true', 
       'acid.cdc.mode.enable' = 'true',--默认同步CDC
       'acid.cdc.build.async'='true', --打开异步的话，就要额外设置
       'acid.cdc.build.interval'='60') 
  LIFECYCLE 36500;

在SQL分析页面，选择项目delta_compute_yunqi，确认以下两个Delta Table表已存在：

• yunqi_github_events_odps_cdc

• yunqi_github_events_odps_dw

步骤

操作

预期结果

Step 1

使用交互式资源组-近实时数据分析

1. 在MaxCompute控制台左侧导航栏中，选择工作区 > SQL分析，打开临时SQL分析页面。

2. 选择步骤一创建的交互式Quota组，进行近实时查询分析。

具体SQL分析步骤如下：

• Q1：过去24小时最活跃项目

  --Q1:过去24小时最活跃项目
  SELECT
      repo_name,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      repo_name
  ORDER BY
      events DESC
  LIMIT 5;

• Q2：过去24小时最活跃开发者

  --Q2:过去24小时最活跃开发者
  SELECT
      actor_login,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      actor_login
  ORDER BY
      events 
  LIMIT 5;

• Q3：今日公开事件总数

  --Q3:今日公开事件总数
  SELECT count(*) FROM yunqi_github_events_odps_dw WHERE TO_DATE(created_at) >= date_add(now(),-1);

• Q4：实时事件展示

  --Q4:实时事件展示
  SELECT
      cast(created_at as STRING ),
      actor_login,
      type,
      repo_name
  FROM
      yunqi_github_events_odps_dw
  ORDER BY
      created_at DESC
  LIMIT 5;

观察查询SQL结果。

Step 2

使用批处理资源组-增量计算-增量物化视图MV聚合查询

1. 在MaxCompute控制台左侧导航栏中，选择工作区 > SQL分析，打开临时SQL分析页面。

2. 选择步骤一创建的批处理Quota组，批量MV应用。

具体SQL分析步骤如下：

1. 确认Delta Table数据表已生成，历史和增量数据更新中。

   创建跟踪Delta Table数据表的CDC信息 （可选：默认CDC自动生成）。

   --Q1: 创建跟踪Delta Table数据表的CDC信息
   ALTER TABLE yunqi_github_events_odps_dw build cdc; --触发继续CDC后续流程

2. 支持实时增量物化视图MV查询。基于历史和增量数据，创建轻度聚合-增量MV：按照repo_name/language/create_date聚合统计事件数量。

   --Q2: 生成增量物化视图MV1-yunqi_incre_mv1
   SET useTunnel=FALSE;
   CREATE materialized VIEW IF NOT EXISTS yunqi_incre_mv1
   refresh every 2 minutes
   tblproperties("enable_auto_refresh"="true", "refresh_mode"="incremental")
   AS SELECT
       repo_name,
       language,
       COUNT(*) AS events,
       to_date(created_at) AS create_date
   FROM
       yunqi_github_events_odps_dw
   WHERE
       date_add(created_at,-1)>'2024-07-26' AND language IS NOT NULL
   GROUP BY
     repo_name,language,to_date(created_at);

   --Q3: 观测源数据表状态 - 验证show history查看表行为 / CDC Build状态确认
   SHOW history FOR TABLE yunqi_github_events_odps_dw;
   SELECT * FROM table_changes('yunqi_github_events_odps_dw', 1);

3. SQL分析：按照repo_name/language信息，查询在一定时间内的各repo的总共事件数量和语言。

   --Q4:按照repo_name / language 信息，降序查询总共事件数据与Language
   SELECT repo_name, language, SUM(events) AS events_sum, COUNT(create_date) AS day_count 
   FROM yunqi_incre_mv1 
   WHERE events > 100 
   GROUP BY repo_name,language ORDER BY events_sum DESC;

   --Q5：增量物化视图MV1历史操作变更
   SHOW history FOR TABLE yunqi_incre_mv1;
   --数据更新批次5分钟间隔，如无变更需执行
   ALTER materialized VIEW yunqi_incre_mv1 rebuild;
   SELECT * FROM table_changes('yunqi_github_events_odps_dw', 1);

4. 构建级联增量物化视图MV2。基于已有的增量物化视图MV1，创建Aggragate重度汇聚层。

   • 生成增量物化视图MV2-yunqi_incre_mv2。在增量MV1的基础上，物化SQL分析结果到增量MV2。参考Q4 SQL。

     --Q6:生成级联增量物化视图MV2-yunqi_incre_mv2
     SET useTunnel=FALSE;
     CREATE materialized VIEW IF NOT EXISTS yunqi_incre_mv2
     refresh every 2 minutes
     tblproperties("enable_auto_refresh"="true", "refresh_mode"="incremental")
     AS 
     SELECT repo_name, language, SUM(events) AS events_sum, COUNT(create_date) AS day_count 
     FROM yunqi_incre_mv1 
     WHERE events > 100 
     GROUP BY repo_name,language;

   • SQL分析：在一定时间内按照编程language分类，按照Repo数量多少来排序查询每个语言对应的总共事件数量。

     --Q7: 编程Language热门排序
     SELECT language, COUNT(repo_name) AS repo_cnt,MAX(events_sum) AS max_events FROM yunqi_incre_mv2 
     GROUP BY language ORDER BY repo_cnt DESC;

     --Q8：观测增量MV2历史操作状态 / 表变化信息
     SHOW history FOR TABLE yunqi_incre_mv2;
     --数据更新批次5分钟间隔，如无变更需执行
     ALTER materialized VIEW yunqi_incre_mv2 rebuild;
     SELECT * FROM table_changes('yunqi_incre_mv2', 2);

观察查询SQL结果。

Step 3

使用批处理资源组-增量计算-Stream&Task应用

1. 在MaxCompute控制台左侧导航栏中，选择工作区 > SQL分析，打开临时SQL分析页面。

2. 选择步骤一创建的批处理Quota组，增量CDC。

具体SQL分析步骤如下：

1. 创建stream对象，查询新增的cdc记录，进行后续的增量计算。

   --Q1: 查询最新的表数据的版本信息
   SELECT get_latest_version('yunqi_github_events_odps_dw'); 

   创建Stream对象。通过上面get_lastest_version信息，设置开始跟踪的增量数据的起始版本。选择小于等于最新版本的一个历史版本开始跟踪增量的CDC记录。

   --Q2：创建Stream对象
   CREATE stream stream_githubs ON TABLE yunqi_github_events_odps_dw version AS OF 10 strmproperties('read_mode'='cdc') ;
   
   DESC stream stream_githubs;

2. 创建并执行Task，Task自动执行增量更新repo_name。

   • 创建聚合表yunqi_github_events_ads。

     --创建聚合表yunqi_github_events_ads
     SET odps.sql.upsertable.table.enable=true;
     CREATE TABLE IF NOT EXISTS 
     yunqi_github_events_ads
     (
         repo_name             STRING COMMENT 'repo全名：owner/Repository_name'
         ,language              STRING COMMENT '编程语言'
         ,events_cnt            BIGINT COMMENT '事件数量'
         ,create_date           DATE COMMENT '事件发生时间'
     )
     STORED AS ALIORC  
     TBLPROPERTIES (
          'transactional'='true',
          'write.bucket.num'='64') 
     LIFECYCLE 36500;

   • Task自动执行增量更新repo_name。

     --Task自动执行增量更新repo_name
     SET odps.sql.periodic.task.enabled=true;
     CREATE task yunqi_incre_task1
     schedule='45 second' 
     taskproperties('schedule_strategy'='200')
     WHEN stream_has_data('stream_githubs')
     AS INSERT overwrite yunqi_github_events_ads SELECT
         repo_name,
         language,
         COUNT(*) AS events_cnt,
         to_date(created_at) AS create_date
     FROM
         yunqi_github_events_odps_dw
     WHERE
         date_add(created_at,-1)>'2024-08-26' AND language IS NOT NULL
     GROUP BY
     repo_name,language,to_date(created_at);
     

     每隔一分钟，自动查询出来DWD的CDC增量数据，进行分析处理后写入ADS表-yunqi_github_events_ads，自定义Pipeline，写起来复杂但用户可以灵活通过SQL自定义增量计算分析处理逻辑。（时长设置：1分钟内可按照秒级设置schdule，超过1分钟按照分钟级设置schdule，最长设置为59分钟）

3. 更新Task状态，Task自动执行增量更新repo_name。

   --更新Task状态，Task自动执行增量更新repo_name
   ALTER task  yunqi_incre_task1  suspend;  --暂停
   ALTER task  yunqi_incre_task1  resume; --恢复

4. 查询CDC build task情况。

   --查询CDC build task情况
   SHOW history FOR task yunqi_incre_task1;
   SELECT *  FROM stream_githubs

观察查询SQL结果。

步骤

操作

预期结果

Step 1

Quota扩容

1. 在MaxCompute控制台左侧导航栏中，选择工作区 > 配额（Quota）管理，单击目标一级Quota组右侧的Quota配置。

2. 在基础配置页面，单击编辑基础配置。

3. 修改交互式Quota组mcqa2_demo_yunqi，将其从64CU扩容至128CU。

目标Quota组资源扩容成功。

Step 2

查询对比分析与性能优化

1. 在MaxCompute控制台左侧导航栏中，选择工作区 > SQL分析，打开临时SQL分析页面。

2. 选择步骤一创建的交互式Quota组，进行近实时查询分析，观测数据变化。

具体SQL分析步骤如下：

• Q1（可选）: 执行Major Compaction操作，优化查询。

  --Q1: 执行Major Compaction操作，优化查询
  SET odps.merge.task.mode=service;
  ALTER TABLE yunqi_github_events_odps_dw compact major;

  MCQA2.0 查询对比性能-交互式资源组。

  --Q2: 近实时查询-实时事件展示
  SELECT
      cast(created_at AS STRING ),
      actor_login,
      type,
      repo_name
  FROM
      yunqi_github_events_odps_dw
  ORDER BY
      created_at DESC
  LIMIT 5;

• 增量MV1查询：按照repo_name/language信息，查询在一定时间内的各repo的总共事件数量和语言。

  --Q3:按照repo_name / language 信息，降序查询总共事件数据与Language
  SELECT repo_name, language, SUM(events) AS events_sum, COUNT(create_date) AS day_count 
  FROM yunqi_incre_mv1 
  WHERE events > 100 
  GROUP BY repo_name,language ORDER BY events_sum DESC;

• 增量MV2查询：在一定时间内按照编程language分类，查询每个语言对应的repo数量与总共事件数量。

  --Q4: 编程Language热门排序
  SELECT language, COUNT(repo_name) AS repo_cnt,MAX(events_sum) AS max_events FROM yunqi_incre_mv2 
  GROUP BY language;

观察查询SQL结果。

步骤

操作

预期结果

Step 1

在DataWorks IDE 数据开发模块配置MaxCompute计算资源

1. 访问DataWorks IDE数据开发页面，在左侧导航栏单击图标，打开数据目录页面。

2. 单击右上角的工作空间管理，选择计算资源，单击绑定计算资源。

3. 在页面右侧配置计算资源、计算配额和资源组。

4. 绑定MaxCompute项目，执行MaxComputeSQL任务。

无

Step 2

在DataWorks IDE数据开发平台进行数据分析

具体SQL分析步骤如下：

• Q1：过去24小时最活跃项目

  SELECT
      repo_name,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      repo_name
  ORDER BY
      events DESC
  LIMIT 5;

• Q2：过去24小时最活跃开发者

  SELECT
      actor_login,
      COUNT(*) AS events
  FROM
      yunqi_github_events_odps_dw
  WHERE
      date_add(created_at,-1)>'2024-08-26'
  GROUP BY
      actor_login
  ORDER BY
      events 
  LIMIT 5;

• Q3：今日公开事件总数

  SELECT count(*) FROM yunqi_github_events_odps_dw WHERE TO_DATE(created_at) >= date_add(now(),-1);

• Q4：实时事件展示

  SELECT
      cast(created_at AS STRING ),
      actor_login,
      type,
      repo_name
  FROM
      yunqi_github_events_odps_dw
  ORDER BY
      created_at DESC
  LIMIT 5;

观察查询SQL结果。