使用内置公开数据集快速体验实时计算Flink版

复制如下代码到作业开发文本编辑区。

-- 通过DDL语句创建SLS源表，SLS中存放了Github的实时数据。
CREATE TEMPORARY TABLE gh_event(
  id STRING,                                        -- 每个事件的唯一ID。
  created_at BIGINT,                                -- 事件时间，单位秒。
  created_at_ts as TO_TIMESTAMP(created_at*1000),   -- 事件时间戳（当前会话时区下的时间戳，如：Asia/Shanghai）。
  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-shenzhen',             -- 存放公开数据的SLS项目。本示例以深圳为例，您需要修改为您的实际地域信息。
  'endPoint' = 'https://cn-shenzhen-intranet.log.aliyuncs.com', -- 公开数据仅限VVP通过私网地址访问。本示例以杭州为例，您需要修改为您的实际地域信息。
  'logStore' = 'realtime-github-events',            -- 存放公开数据的SLS logStore。
  'accessId' =  'LTAI5tNF1rP8PKVyYjr9****',         -- 只读账号的AK。
  'accessKey' = 'FDgsh6fjSmkbFsx083tN6HOiqN****',   -- 只读账号的SK。
  'batchGetSize' = '500'                            -- 批量读取数据，每批最多拉取500条。
);

-- 配置开启mini-batch, 每2s处理一次。
SET 'table.exec.mini-batch.enabled'='true'; 
SET 'table.exec.mini-batch.allow-latency'='2s'; 
SET 'table.exec.mini-batch.size'='4096'; 

-- 作业设置4个并发，聚合更快。
SET 'parallelism.default' = '4';

-- 查看当天Github新增star数Top 10仓库。
SELECT DATE_FORMAT(created_at_ts, 'yyyy-MM-dd') as `date`, repo_name, COUNT(*) as num
FROM gh_event
WHERE type = 'WatchEvent' AND DATE_FORMAT(created_at_ts, 'yyyy-MM-dd') = DATE_FORMAT(NOW(), 'yyyy-MM-dd')
GROUP BY DATE_FORMAT(created_at_ts, 'yyyy-MM-dd'), repo_name
ORDER BY num DESC
LIMIT 10; 

按照实际地域配置WITH参数中的project和endPoint（本示例以深圳地域为例），其他参数保持默认值，无需修改。

单击调试。

在弹出的对话框中，选择调试集群为flink-sql-test-session（第一步中创建的Session集群），确认调试数据状态为线上数据后，单击确定。

查看可视化结果。

如果需要继续下一步，请单击结果区域左侧红色停止按钮。

因为这是一个流式作业，作业会持续运行，结果会持续更新。此步骤完成后如果您忘记单击停止按钮且继续了后续示例的执行，可能会出现资源不足无法显示结果的情况。这时，您需要在Session集群页面先停止Session集群，再启动Session集群至运行状态。

复制如下代码到作业开发文本编辑区。

-- 通过DDL语句创建SLS源表，SLS中存放了Github的实时数据。
CREATE TEMPORARY TABLE gh_event(
  id STRING,                                        -- 每个事件的唯一ID。
  created_at BIGINT,                                -- 事件时间，单位秒。
  created_at_ts as TO_TIMESTAMP(created_at*1000),   -- 事件时间戳（当前会话时区下的时间戳，如：Asia/Shanghai）。
  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-shenzhen',             -- 存放公开数据的SLS项目。本示例以深圳为例，您需要修改为您的实际地域信息。
  'endPoint' = 'https://cn-shenzhen-intranet.log.aliyuncs.com',  -- 公开数据仅限VVP通过私网地址访问。本示例以深圳为例，您需要修改为您的实际地域信息。
  'logStore' = 'realtime-github-events',            -- 存放公开数据的SLS logStore。
  'accessId' =  'LTAI5tNF1rP8PKVyYjr9****',         -- 只读账号的AK。
  'accessKey' = 'FDgsh6fjSmkbFsx083tN6HOiqN****',   -- 只读账号的SK。
  'batchGetSize' = '500',                           -- 批量读取数据，每批最多拉取500条。
  'startTime' = '2023-07-24 00:00:00'               -- 开始时间，尽量设置到需要计算的时间附近，否则无效计算的时间较长。默认值为当前值。
);

-- 配置开启mini-batch, 每2s处理一次。
SET 'table.exec.mini-batch.enabled'='true'; 
SET 'table.exec.mini-batch.allow-latency'='2s'; 
SET 'table.exec.mini-batch.size'='4096'; 

-- 作业设置4个并发，聚合更快。
SET 'parallelism.default' = '4'; 

-- 查看某组织(如：apache)最近7天的开发者活跃度实时趋势。
SELECT NOW() as `current_ts`, COUNT(*) as pr_count
FROM gh_event
WHERE org_login = 'apache' and  type = 'PullRequestEvent' and
created_at_ts >= NOW() - INTERVAL '7' DAY;

按照实际地域配置WITH参数中的project和endPoint（本示例以深圳地域为例），其他参数保持默认值，无需修改。

单击调试。

在弹出的对话框中，选择调试集群为flink-sql-test-session（第一步中创建的Session集群），确认调试数据状态为线上数据后，单击确定。

查看可视化结果。

如果需要继续下一步，请单击结果区域左侧红色停止按钮。

因为这是一个流式作业，作业会持续运行，结果会持续更新。此步骤完成后如果您忘记单击停止按钮且继续了后续示例的执行，可能会出现资源不足无法显示结果的情况。这时，您需要在Session集群页面先停止Session集群，再启动Session集群。

复制如下代码到作业开发文本编辑区。

-- 通过DDL语句创建SLS源表，SLS中存放了Github的实时数据。
CREATE TEMPORARY TABLE gh_event(
  id STRING,                                        -- 每个事件的唯一ID。
  created_at BIGINT,                                -- 事件时间，单位秒。
  created_at_ts as TO_TIMESTAMP(created_at*1000),   -- 事件时间戳（当前会话时区下的时间戳，如：Asia/Shanghai）。
  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-shenzhen',              -- 存放公开数据的SLS项目。本示例以深圳为例，您需要修改为您的实际地域信息。
  'endPoint' = 'https://cn-shenzhen-intranet.log.aliyuncs.com', -- 公开数据仅限VVP通过私网地址访问。本示例以深圳为例，您需要修改为您的实际地域信息。
  'logStore' = 'realtime-github-events',            -- 存放公开数据的SLS logStore。
  'accessId' =  'LTAI5tNF1rP8PKVyYjr9****',         -- 只读账号的AK。
  'accessKey' = 'FDgsh6fjSmkbFsx083tN6HOiqN****',   -- 只读账号的SK。
  'batchGetSize' = '500',                           -- 批量读取数据，每批最多拉取500条。
  'startTime' = '2023-07-24 00:00:00'               -- 开始时间，尽量设置到需要计算的时间附近，否则无效计算的时间较长。默认值为当前值。
);

-- 配置开启mini-batch, 每2s处理一次。
SET 'table.exec.mini-batch.enabled'='true'; 
SET 'table.exec.mini-batch.allow-latency'='2s'; 
SET 'table.exec.mini-batch.size'='4096'; 

-- 作业设置4个并发，聚合更快。
SET 'parallelism.default' = '4'; 

-- 实时统计过去一周里Github小时级别的评论热力图。
SELECT  DATE_FORMAT(created_at_ts, 'yyyy-MM-dd') as comment_date, HOUR(created_at_ts) AS comment_hour ,COUNT(*) AS comment_count
FROM gh_event
WHERE created_at_ts >= NOW() - INTERVAL '7' DAY AND (type='CommitCommentEvent'
    OR type='IssueCommentEvent' or type = 'PullRequestReviewCommentEvent')
GROUP BY DATE_FORMAT(created_at_ts, 'yyyy-MM-dd'), HOUR(created_at_ts) ;

按照实际地域配置WITH参数中的project和endPoint（本示例以深圳地域为例），其他参数保持默认值，无需修改。

单击调试。

在弹出的对话框中，选择调试集群为flink-sql-test-session（第一步中创建的Session集群），确认调试数据状态为线上数据后，单击确定。

查看可视化结果。

完成本示例后，请单击结果区域左侧红色停止按钮。

因为这是一个流式作业，作业会持续运行，结果会持续更新。体验完成之后，请及时单击停止按钮，否则会继续消耗计算资源。