Flink+Paimon在阿里云大数据云原生运维数仓的实践

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 构建实时云原生运维数仓以提升大数据集群的运维能力,采用 Flink+Paimon 方案,解决资源审计、拓扑及趋势分析需求。

1. 背景

随着大数据产品云原生化的推进,云原生集群的规模和数量都在增加,云原生集群的运维难度也在不断增加,云原生集群的资源审计、资源拓扑、资源趋势的需要就比较迫切。云原生集群的资源审计主要是 node 资源、pod 资源,如当前集群的 node 数量以及Pod 数量;资源拓扑主要构建用户->实例->pod->node->集群的关联关系,例如一个公有云用户,提供给运维团队可以通过 uid 查询到用户实例以及实例所在的节点/集群信息。资源趋势是整个团队所管理的集群资源使用趋势,pod 数量趋势、node 数量趋势、以及用户使用资源趋势。

为了提升大数据产品云原生集群的运维能力,我们构建一个实时云原生运维数仓,业务能力提供资源审计、资源拓扑、资源趋势等能力。在数据性能方面最终数据需要具备实时性(分钟级)和一致性。

目前 Flink 已经是实时计算的标准解决方案,在这个基础上阿里云开源大数据团队推出 Paimon 为低成本的数据湖解决方案。本文我们使用 Flink+Paimon 为 流式数仓解决方案应用到云原生运维数仓中,提供实时(分钟级)的数据服务,提升云原生集群的运维能力。

2. 总体设计

2.1 方案选型

如 1 中背景所述,因为数据的实时性需求,所以我们不考虑离线数仓解决方案,直接考虑实时数仓解决方案。目前阿里云实时数仓解决方案中,比较推荐使用的是 Flink+Hologres 和 Flink+Paimon 两种解决方案。其中 Flink+Hologres 已经比较成熟,本文中我们不详细描述,其主要缺点是 Hologres 存储成本比较高。最终我们选择使用 Flink+Paimon 的方案,优势在于存储成本低,实时性高。

2.2 最终方案

数据采集采用 exporter operator 实现 kubernet 负载数据到 sls,为了适配阿里云多 region 网络隔离的问题,我们在数据出 region 之前先同步到 sls 中,再直接同步到 sls 的跨域网络同步到 paimon ods 表中,同步的过程需要处理数据排序问题。flink 作业通过运维数仓本身有比较高的实时性的要求,我们在数据处理过程中采用 Flink+Paimon 的解决方案,paimon 支持元数据注册到 maxcompute 中,可以直接使用 dataworks /flink 查询数据;最终数据报表采用 FBI 进行展示,资源拓扑数据写入到 CMDB 中给运维系统使用。

3. 落地实践

3.1 数据采集

我们研发了 exporter-operator 工具,嵌入 Kubernetes 集群中的哨兵,实时监听 Kubernetes api server 中 workload 数据变化。作为集群生态系统的 addon 组件,exporter-operator 以高效的响应机制驻留内部。

通过配置 Informer 中的自定义资源(CR),exporter-operator 灵活启用了多个 workload-informer 实例,针对各类工作负载实施监听,赋予系统卓越的灵活性,使其能够灵活监控多种工作负载信息。

exporter-operator 不断捕获并即时传输这些关键的工作负载数据,无缝对接阿里云日志服务(SLS)或 Apache Kafka 流处理平台,实现实时记录与深度分析集群运行状态,为优化集群效能提供强有力的数据支持。

3.2 数据清洗

为了数据分析可以使用 dataworks 进行分析,数据清洗前需要按照文档Paimon外部表进行配置。配置完成后创建 Paimon 表的元数据会直接同步到 dataworks 中,就可以直接进行数据分析。

3.2.1 ods 层

ods 层数据直接把 sls/kafka 中的数据同步到 paimon ods 表,保存有所有同步到的负载数据。

CREATE TABLE IF NOT EXISTS `abm-exporter-paimon`.`abm_exporter_db`.`ods_realtime_exporter_lakehouse`
 (
     `sls_time`  bigint
     ,`cluster`  varchar
     ,`content`  varchar
 ) PARTITIONED BY (cluster) with( 'orc.write.batch-size' = '128','file.format' = 'avro','bucket' = '8')
insert into `abm-exporter-paimon`.`abm_exporter_db`.`ods_realtime_exporter_lakehouse` /*+ OPTIONS('orc.write-buffer-size' = '128','file.format' = 'avro') */
select
    `__timestamp__` as sls_time
    ,`__topic__` as cluster
    ,`content`
from source_k8s_meta;

注意:

paimon 默认文件格式是 orc,orc 是列存,所以需要先在内存里攒批然后写出去。默认一批是 1024 行,目前写入数据一行几 mb ,1024 行的 buffer 就几 G 了,堆内存会不够。所以选择使用行存:设置 'file.format' = 'avro',用 avro 格式;另外也可以设置 'orc.write.batch-size' = '128',把批的大小改成 128 来解决。

3.2.2 dwd 层

dwd 层的功能有三个部分,1.将数据按照 primary key 聚合 ,确保资源数据的唯一性 2.将数据进行排序,确保 delete 的数据在最后发生 3.集群维度进行维表 join,丰富集群维度字段。

  CREATE TABLE IF NOT EXISTS `abm-exporter-paimon`.`abm_exporter_db`.`dwd_realtime_exporter_lakehouse`
 (
   type                 varchar
   ,obj                 varchar
   ,k8s_group           varchar
   ,version             varchar
   ,kind                varchar
   ,namespace           varchar
   ,name                varchar
   ,resourceversion     varchar
   ,region_id           varchar NOT NULL
   ,cluster             varchar NOT NULL
   ,watchname           varchar NOT NULL
   ,watchsetuptimestamp bigint NOT NULL
   ,uid                 varchar PRIMARY KEY NOT ENFORCED
   ,collect_time        bigint NOT NULL
   ,sls_time            bigint
   ,cluster_id          varchar
   ,region              varchar
   ,zone                varchar
 )
 with (
   'merge-engine' = 'partial-update'
   ,'changelog-producer' = 'lookup'
,'partial-update.ignore-delete'='true'
 );
set 'table.exec.sink.upsert-materialize' = 'NONE';
insert into `abm-exporter-paimon`.`abm_exporter_db`.`dwd_realtime_exporter_lakehouse`  /*+ OPTIONS('orc.write-buffer-size' = '128','file.format' = 'avro') */
select
  json_value(`content`, '$.type')             as `type`
  ,json_value(`content`, '$.obj')              as `obj`
  ,json_value(`content`, '$.group')            as `k8s_group`
  ,json_value(`content`, '$.version')          as `version`
  ,json_value(`content`, '$.kind')             as `kind`
  ,json_value(`content`, '$.namespace')        as `namespace`
  ,json_value(`content`, '$.name')             as `name`
  ,json_value(`content`, '$.resourceversion')  as `resourceversion`
  ,'cn-zhangjiakou'                               as `region_id`
  ,`cluster`                                as `cluster`
  ,json_value(`content`, '$.watchname')        as `watchname`
  ,cast(json_value(`content`, '$.watchsetuptimestamp') as bigint)    as `watchsetuptimestamp`
  ,json_value(`content`, '$.uid')              as `uid`
  ,cast(UNIX_TIMESTAMP() as bigint)                      as `collect_time`
  ,`sls_time`                            as `sls_time`
  ,NULL
  ,NULL
  ,NULL
from (
  SELECT
    *
  FROM (
    SELECT
      *
      ,ROW_NUMBER()
      OVER (
        PARTITION BY `cluster`, json_value(`content`, '$.uid')
        ORDER BY cast(json_value(`content`, '$.resourceversion') as bigint) DESC,cast(json_value(`content`, '$.__time_milli__') as bigint) DESC
      ) as rowNum
    FROM `abm-exporter-paimon`.`abm_exporter_db`.`ods_realtime_exporter_lakehouse`
  )
  WHERE rowNum = 1
);

3.2.3 ads 层

ads 层进行数据聚合,flink 处理逻辑只负责将数据写入到 Paimon 中,使用 Paimon 的merge-engine='aggregation'进行字段的聚合。这样计算逻辑不在 flink 中,不需要在 flink state 中进行,大量减少 state 的资源消耗。因为 Paimon 的计算是在 flink checkpoint 结束后触发,所以减少 flink checkpoint 的间隔时间可以提高数据的实时性。

CREATE TABLE IF NOT EXISTS `abm-exporter-paimon`.`abm_exporter_db`.`ads_realtime_exporter_lakehouse`
(
    kind          varchar
    ,cluster      varchar NOT NULL
    ,workload_num bigint
    ,PRIMARY KEY (kind, cluster) NOT ENFORCED
)
with (
    'merge-engine' = 'aggregation'
    ,'changelog-producer' = 'lookup'
    ,'fields.workload_num.aggregate-function' = 'sum'
)
;
insert into `abm-exporter-paimon`.`abm_exporter_db`.`ads_realtime_exporter_lakehouse`
select
    kind
    ,cluster
    ,case
        when type = 'ADDED' then 1
        when type = 'DELETED' then -1
        ELSE  0
    end as workload_num
from `abm-exporter-paimon`.`abm_exporter_db`.`dwd_realtime_exporter_lakehouse`;

3.3 数据分析

如果只需要查询一次的情况可以使用 dataworks 进行数据查询,当然也可以直接使用 flink 进行数据查询,当前 VVP 里面已经开发了查询的功能。如果需要生成报表就使用fbi 等工具进行数据分析,也可以使用 flink 将相关的数据清洗到 cmdb 系统中去。

use bigdata_sre;
SET odps.table.scan-options.odps.external.sub.disable.hyper=true;
SET odps.sql.common.table.planner.ext.hive.bridge = true;
set odps.sql.submit.ddltask.via.common.table=true;
SET odps.sql.hive.compatible = false;
set odps.compiler.verify=true;
set odps.isolation.session.enable=true;
-- select * from ods_realtime_exporter_lakehouse limit 100;
-- select * from dwd_realtime_exporter_lakehouse where kind='Node' limit 100;
select * from ads_realtime_exporter_lakehouse order by workload_num desc limit 100;

4. 总结

目前 Flink+Paimon 实时数据湖的方案已经比较成熟,使用成本不高,相关的生态也比较完善,在低成本和低延迟的要求下,这个方案还是非常好的选择。如果对成本没太高的要求,Flink+Hologres 在延迟方面会有些优势。

参考文件:

  1. 《Flink+Paimon构建流式数据湖仓》

    https://ata.atatech.org/articles/11000268231?spm=ata.25287382.0.0.6be65f6948opbG

  2. 《Paimon外部表》

    https://help.aliyun.com/zh/maxcompute/user-guide/paimon-external-table

  3. 《当流计算邂逅数据湖:Paimon 的前生今世》

    https://xie.infoq.cn/article/63890f6dc8afcbfaac312444f


更多内容

img


活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算 Flink 版现开启活动:
新用户复制点击下方链接或者扫描二维码即可0元免费试用 Flink + Paimon
实时计算 Flink 版(3000CU*小时,3 个月内)
了解活动详情:https://free.aliyun.com/?utm_content=g_1000395379&productCode=sc

retouch_2024070417440476.jpg

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
2月前
|
运维 监控 负载均衡
高效运维实践:常见问题的应对策略与实践经验
本文探讨了运维工作中的五大核心挑战及应对策略,涵盖负载均衡优化、数据库性能提升、系统监控预警、容器化与微服务运维等方面,旨在帮助企业提升系统稳定性与运维效率。
|
3月前
|
分布式计算 运维 监控
Fusion 引擎赋能:流利说如何用阿里云 Serverless Spark 实现数仓计算加速
本文介绍了流利说与阿里云合作,利用EMR Serverless Spark优化数据处理的全过程。流利说是科技驱动的教育公司,通过AI技术提升用户英语水平。原有架构存在资源管理、成本和性能等痛点,采用EMR Serverless Spark后,实现弹性资源管理、按需计费及性能优化。方案涵盖数据采集、存储、计算到查询的完整能力,支持多种接入方式与高效调度。迁移后任务耗时减少40%,失败率降低80%,成本下降30%。未来将深化合作,探索更多行业解决方案。
133 1
|
2月前
|
运维 监控 安全
从实践到自动化:现代运维管理的转型与挑战
本文探讨了现代运维管理从传统人工模式向自动化转型的必要性与路径,分析了传统运维的痛点,如效率低、响应慢、依赖经验等问题,并介绍了自动化运维在提升效率、降低成本、增强系统稳定性与安全性方面的优势。结合技术工具与实践案例,文章展示了企业如何通过自动化实现运维升级,推动数字化转型,提升业务竞争力。
|
4月前
|
SQL 关系型数据库 MySQL
客户说|保险极客引入阿里云AnalyticDB,多业务场景效率大幅提升
“通过引入AnalyticDB,我们在复杂数据查询和实时同步方面取得了显著突破,其分布式、弹性与云计算的优势得以充分体现,帮助企业快速响应业务变化,实现降本增效。AnalyticDB的卓越表现保障了保险极客数据服务的品质和效率。”
|
6月前
|
数据采集 机器学习/深度学习 人工智能
智能运维在IT管理中的实践与探索
【10月更文挑战第21天】 本文深入探讨了智能运维(AIOps)技术在现代IT管理中的应用,通过分析其核心组件、实施策略及面临的挑战,揭示了智能运维如何助力企业实现自动化监控、故障预测与快速响应,从而提升整体运维效率与系统稳定性。文章还结合具体案例,展示了智能运维在实际环境中的显著成效。
224 26
|
12月前
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。
|
10月前
|
存储 分布式计算 流计算
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
本文介绍了阿里云开源大数据团队在实时计算领域的最新成果——向量化流计算引擎Flash。文章主要内容包括:Apache Flink 成为业界流计算标准、Flash 核心技术解读、性能测试数据以及在阿里巴巴集团的落地效果。Flash 是一款完全兼容 Apache Flink 的新一代流计算引擎,通过向量化技术和 C++ 实现,大幅提升了性能和成本效益。
3088 73
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
zdl
|
10月前
|
消息中间件 运维 大数据
大数据实时计算产品的对比测评:实时计算Flink版 VS 自建Flink集群
本文介绍了实时计算Flink版与自建Flink集群的对比,涵盖部署成本、性能表现、易用性和企业级能力等方面。实时计算Flink版作为全托管服务,显著降低了运维成本,提供了强大的集成能力和弹性扩展,特别适合中小型团队和业务波动大的场景。文中还提出了改进建议,并探讨了与其他产品的联动可能性。总结指出,实时计算Flink版在简化运维、降低成本和提升易用性方面表现出色,是大数据实时计算的优选方案。
zdl
396 56

相关产品

  • 实时计算 Flink版