实时计算Flink版体验评测

产品概述

阿里云实时计算Flink版（Alibaba Cloud Realtime Compute for Apache Flink，Powered by Ververica）是阿里云基于Apache Flink构建的企业级、高性能实时大数据处理系统。是一种全托管Serverless的Flink云服务，开箱即用，计费灵活。具备一站式开发运维管理平台，支持作业开发、数据调试、运行与监控、自动调优、智能诊断等全生命周期能力。100%兼容Apache Flink，支持开源Flink平滑迁移上云，核心企业级增强Flink引擎较开源Flink有约两倍性能的提升。提供端到端亚秒级实时数据分析能力，并通过标准SQL降低业务开发门槛，助力企业向实时化、智能化大数据计算升级转型。

相比开源Flink，阿里云实时计算Flink版具备如下优势：

• 在成本上，基于Serverless的服务支持作业弹性扩缩容，只需增加计算节点无需修改硬件配置，提供弹性体验。
• 在性能上，具备单核CPU每秒数十万条记录处理能力，可实现端到端亚秒级数据处理延迟，可支持数万并发超大规模实时任务计算。
• 在稳定性上，服务分布式集群计算无单点，保障高可用。
• 在功能上，提供数十种作业指标监控和一站式开发界面，支持SQL作业调试，支持作业运行期间智能调优。提供了丰富的数据实时入湖入仓和实时风控场景能力。
• 在使用过程中，已有开源Flink任务可平滑迁移上云，无缝对接主流开源大数据生态，具备三十余种上下游连接器，极大节约企业的人力支出与资源投入成本。

产品开通

新用户可以点击链接前往领取0元免费试用 Flink，有（3000CU*小时，3 个月内）额度。

如果你是老用户，那就得老老实实进行按量付费购买了，点击链接即可前往控制台进行购买。

当然产品经理还是考虑到诸多老用户体验感受的，特地推出了优惠套餐，59元/3个月，对于需要短期体验的用户来说还是非常合适的。

由于实时计算Flink版采用的是存储计算分离架构，所以根据存储类型不同，有全托管和对象存储两种规格。这里为了体验方便，选择全托管。在购买页面选择付费模式为按量计费，地域和可用区根据实际情况选择，这里选华东1（杭州），工作空间名称自定义一个，存在重复会有提示，存储类型选择全托管（目前是免费公测，对于老用户来说是个小福利），勾选协议，点击确认订单即可。

在接下来的支付页面点击确认支付即可。

看到该界面就说明产品开通成功了，实例创建需要花费一点时间，完成后接下来就可以进行入门体验了。

入门体验

点击实例ID即可进入管理控制台，如下：

下面将使用内置的GitHub公开事件样例数据，快速体验实时计算Flink版产品的实时数据分析功能。

首先需要创建Session集群。

填上名称session-test，状态选择RUNNING，引擎版本选择第一个推荐的vvr-8.0.9-flink-1.17，Task Managers 数量配置为4。

完成参数配置，点击创建Session集群。稍等片刻，集群状态变成运行中。

接下来需要创建一个SQL作业，点击左侧导航栏的数据开发——ETL，单击新建。

选择新建空白的流作业草稿，点击下一步。

这里的参数可以保持默认，也可以自定义，按需选择即可。

将下面的SQL粘贴到作业开发的文本编辑区，用来统计当天各个仓库新增Star数Top 10。

-- 通过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-hangzhou',             -- 存放公开数据的SLS项目。本示例以杭州为例，您需要修改为您的实际地域信息。
  'endPoint' = 'https://cn-hangzhou-intranet.log.aliyuncs.com', -- 公开数据仅限VVP通过私网地址访问。本示例以杭州为例，您需要修改为您的实际地域信息。
  'logStore' = 'realtime-github-events',            -- 存放公开数据的SLS logStore。
  'accessId' =  'xxxx',         -- 只读账号的AK，无需修改。
  'accessKey' = 'FDgsh6fjSmkbFsx083tN6HOiqNVWTP',   -- 只读账号的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;

单击作业区页面右上角的调试，并选择上述创建的Session集群，点击确认。

调试完成后即可在结果栏看到统计效果。

在作业开发文本编辑区粘贴如下SQL，可以统计过去一周里Github小时级别的评论热力图。

-- 通过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-hangzhou',              -- 存放公开数据的SLS项目。本示例以杭州为例，您需要修改为您的实际地域信息。
  'endPoint' = 'https://cn-hangzhou-intranet.log.aliyuncs.com', -- 公开数据仅限VVP通过私网地址访问。本示例以杭州为例，您需要修改为您的实际地域信息。
  'logStore' = 'realtime-github-events',            -- 存放公开数据的SLS logStore。
  'accessId' =  'xxxx',         -- 只读账号的AK，无需修改。
  'accessKey' = 'FDgsh6fjSmkbFsx083tN6HOiqNVWTP',   -- 只读账号的SK，无需修改。
  'batchGetSize' = '500',                           -- 批量读取数据，每批最多拉取500条。
  'startTime' = '2024-09-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) ;

单击作业区页面右上角的调试，并选择上述创建的Session集群，点击确认。呈现的热力图如下：

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

体验到这是不是意犹未尽，这里为了避免文章冗长就不再演示其他入门实验了，感兴趣的小伙伴可以点击链接参照官网文档进行自行体验。

对比分析

目前市面上可以看到并且具有广泛应用案例的实时计算产品有百度的流式计算（Baidu Stream Computing，简称BSC）、腾讯的流计算Oceanus、火山引擎的流计算Flink版、亚马逊的AWS Kinesis Data Analytics for Apache Flink、谷歌的Google Cloud Dataflow with Apache Flink、以及微软的Azure Stream Analytics with Apache Flink Compatibility。他们之间的共同点可以简要归纳如下：

• 基于Apache Flink构建，利用了Flink的高性能、分布式、实时计算等特性
• 提供了云托管服务，用户无需自行搭建和维护计算集群，降低了运维成本和技术门槛
• 支持实时数据处理，能够处理大规模的数据流，并提供高效的数据分析和实时决策支持
• 支持多种数据源接入和多种数据输出方式，满足用户不同的数据处理需求
• 提供了全链路监控和运维工具，帮助用户更好地管理和优化计算任务

它们之间的差异除了主观的使用体验和文档支持存在差异外，其他对比项以表格罗列如下：

基于表格可以将各个产品间的不同点归纳如下：

• 产品都由不同的云服务提供商提供，都有自己的生态系统和服务体系，用户可以根据自己的需求和偏好选择相应的产品
• 产品在功能和特性上都有所不同。例如，阿里云实时计算Flink版提供了端到端亚秒级实时数据分析能力，并通过标准SQL降低业务开发门槛；腾讯云流计算Oceanus提供了丰富的数据处理和分析能力，并支持多种数据处理方式和算法；AWS Kinesis Data Analytics for Apache Flink则降低了构建和管理Apache Flink应用程序的复杂性
• 各个产品的定价和成本模型可能有所不同
• 虽然都支持Apache Flink社区，但每个产品的官方支持和用户社区可能有所不同

体验总结

首先提出一个体验过程中发现的问题，或者说是错误。在产品首页的产品规格中，有一个59元/3月的优惠产品，本以为是针对老用户的一个福利，没想到点击购买却出现了不一样的情况。

明明标注的是容量规格是3000CUH，跳转后却成了5000CUH，也就是说出现了产品规格不一致的情况，也就直接导致了无法优惠购买。

下面逐条逐项进行体验总结

（1）在体验实时计算Flink版产品的过程中，我总体上得到了较为全面的产品内引导和文档帮助。产品界面设计清晰，关键功能点都有相应的说明和指引，这对于初次接触的用户来说非常友好。此外，官方文档也相当详尽，从安装部署到功能使用，再到高级配置，都提供了详细的步骤和示例。

然而，在深入探索某些高级功能时，我发现文档和产品内引导略显不足。例如，在进行复杂的数据处理任务时，关于如何优化Flink作业性能、如何调整并行度等高级参数的文档相对较少，且部分参数的解释不够详细，容易让新手用户感到困惑。此外，虽然产品内提供了部分示例代码，但针对特定业务场景的完整案例相对较少，这在一定程度上限制了用户的快速上手和实际应用。

因此，我认为在以下方面还可以进一步加强：

• 增加针对高级功能的详细文档和指引，特别是关于性能优化和参数调整的部分。
• 提供更多针对特定业务场景的完整案例和示例代码，帮助用户更好地理解和应用Flink。
• 在产品内增加更多互动式的引导和学习资源，如在线教程、视频讲解等，以提升用户体验和学习效率。

（2）从功能层面来看，实时计算Flink版产品基本满足了我的预期。在数据开发方面，该产品提供了丰富的API和算子库，支持复杂的数据处理逻辑；同时，它还支持多种数据源和数据格式的接入，使得数据开发过程更加灵活和高效。在运维方面，该产品提供了完善的监控和报警功能，能够实时监控作业的运行状态和性能指标，及时发现问题并进行处理。

此外，该产品还支持多种部署方式，如单机部署、集群部署等，能够满足不同规模的业务需求。同时，它还提供了丰富的安全功能，如数据加密、访问控制等，确保数据的安全性和隐私性。

然而，在部分细节方面，我认为还有改进的空间。例如，在作业调度和资源管理方面，该产品目前还缺乏一些高级功能，如动态调整作业资源、作业优先级管理等。这些功能的缺失可能会在一定程度上影响作业的性能和稳定性。

（3）针对我所体验的业务场景，我认为实时计算Flink版产品还可以在以下方面进行改进或增加更多功能：

• 增强实时数据分析能力：目前该产品主要侧重于实时数据处理和流计算，但在实时数据分析方面还有所欠缺。可以考虑增加一些实时数据分析工具和算法库，以便用户能够更方便地进行实时数据分析和挖掘。
• 优化作业部署和升级流程：在作业部署和升级方面，目前该产品还需要用户手动进行一系列操作，这在一定程度上增加了运维成本。可以考虑引入自动化部署和升级工具或流程，降低运维难度和成本。
• 增加更多数据源支持：虽然该产品已经支持了多种数据源和数据格式的接入，但随着业务的发展和新数据源的出现，可能需要不断增加对新数据源的支持。因此，建议持续关注和更新数据源支持列表，以满足用户的多样化需求。

（4）作为大数据流式处理的一款产品，我觉得它与其他产品联动的类型会有很多，比如：

• 与大数据平台结合：可以将实时计算Flink版产品与Hadoop、Spark等大数据平台结合使用，实现离线数据分析和实时数据处理的有机结合。例如，可以利用Hadoop进行大规模数据存储和预处理，然后利用Flink进行实时数据分析和处理。
• 与数据库结合：可以将实时计算Flink版产品与关系型数据库（如MySQL、PostgreSQL）或非关系型数据库（如MongoDB、Cassandra）结合使用，实现数据的实时入库和查询。这有助于提升数据处理的实时性和准确性。
• 与机器学习平台结合：可以将实时计算Flink版产品与机器学习平台（如TensorFlow、PyTorch）结合使用，实现实时数据分析和预测。例如，可以利用Flink进行实时数据预处理和特征提取，然后利用机器学习模型进行实时预测和决策。
• 与消息队列结合：可以将实时计算Flink版产品与Kafka、RabbitMQ等消息队列结合使用，实现数据的实时传输和处理。这有助于提升数据处理的实时性和可靠性，特别是在分布式系统中。

如果你阅读到这感觉还是意犹未尽，下面提供的视频课程和在线实验也许可以帮到你。如下：

• 视频课程

Apache Flink 入门

实时计算 Flink 版产品入门与实操

实时计算 Flink 实战

• 云起实验室

基于Hologres+PAI+计算巢，5分钟搭建企业级AI问答知识库

基于Hologres+Flink搭建GitHub实时数据大屏

基于Flink+Tair搭建实时监控大屏

使用Flink实时发现最热Github项目

5分钟上手Flink MySQL连接器