友盟+｜如何通过阿里云Flink+Paimon实现流式湖仓落地方案 ...

1. 友盟+介绍

友盟+ 以“数据智能，驱动业务增长”为使命，为移动应用开辟者和企业提供包罗统计分析、性能监测、消息推送、智能认证等一站式办理方案。停止 2023 年 6 月，已累计为 270 万移动应用和 980 万家网站，提供十余年的专业数据服务。
作为国内最大的移动应用统计服务商，其统计分析产品 U-App & U-Mini & U-Web 为开辟者提供基础报表及自定义用户举动分析服务，能够资助开辟者更好地明白用户需求，优化产品功能，提升用户体验，助力业务增长。

为了满意产品、运营等多业务脚色对数据差别视角的分析需求，统计分析 U-App 提供了包罗用户分析、页面路径、卸载分析在内的多种「开箱即用」的预置报表，集成 SDK 上报数据后即可检察这些指标。除此以外，为了满意个性化的分析诉求，业务也可以自定义报表的计算规则，提供了事件细分、漏斗分析、留存分析等用户举动分析模子，用户可以根据本身的分析需求灵活地选择时间范围、设置事件名称、where 筛选和 Groupby 分组等。

如上所述，U-App 服务了众多应用场景，每天处置惩罚接近千亿条日记，需要思量平衡好数据新鲜度、查询延迟和本钱的关系，同时保障体系的稳定性，这对数据架构和技术选型提出了极高的要求。
针对报表类型差别的看数场景和业务需求，我们底层技术架构通过多种产品来支撑。在数据新鲜度方面，分别是提供了 T+0 的实时计算 和 T+1的离线批量计算，主要支持预置报表的计算场景，并将计算好的结果导出到存储，能够支持高并发的报表查询。在分析时效性方面，实现自定义报表支持秒级的 OLAP 分析，但鉴于本钱和稳定性思量，对于大数据量和大跨度的时间查询会走离线触发式计算。
在本文中，我们会分享友盟+ U-App 团体的技术架构，以及在实时和离线计算上面的优化方案
2. 友盟+数据架构及现状

如下图所示，在大数据范畴这是一个比力通用的数据处置惩罚 pipeline，贯穿数据的加工&利用过程包罗，数据收罗&接入、数据清洗&传输、数据建模&存储、数据计算&分析 以及 查询&可视化，其中友盟U-App 数据处置惩罚的核心架构是红框部分。

U-App 团体架构大体可以分为四层：数据服务、数据计算、数据存储以及核心组件
● 数据服务：将查询 DSL 剖析为底层引擎执行的 DAG，同时智能采样、查询排队等来尽可能淘汰体系过载情况，保证查询顺滑
● **数据计算：**根据差别分析场景抽象沉淀了自定义分析模子，包罗举动分析和画像分析两大类；而且提供预置的基础统计指标的计算
● **数据存储：**利用了以 User-Event 为核心的数据模子，提供基于明细数据的举动分析
● **核心组件：**离线批量计算利用 MaxCompute，流式计算利用阿里云上实时计算 Flink，OLAP 计算利用 Hologres
3. 基于Flink + Paimon的流式湖仓利用实践

本节首先将介绍Apache Paimon主要上风，然后介绍基于Paimon在U-App实时基础指标计算和友盟设备ID维表更新场景的优化方案
3.1 Apache Paimon简介

3.1.1 概览

Apache Paimon 是一项流式数据湖存储技术，可以为用户提供高吞吐、低延迟的数据摄入、流式订阅以及实时查询能力。通俗表明即 Paimon 是一个流批一体的湖存储格式，它不是一个服务只是一个格式一个Jar包, 数据存储在的 OSS 大概 HDFS 上。可以利用 Flink CDC 来一键入湖到 Paimon 中，也可以通过 Flink SQL 或 Spark SQL 来批写、流写到 Paimon 当中。Paimon 也支持主流开源引擎，包罗几乎现在全部的开源引擎。Paimon 也可以被 Flink 或 Spark 流读，这也是它作为流式数据湖的特有能力之一。
3.1.2 典型应用场景

● CDC 更新入湖，可被准实时查询(1-5min)，并大幅简化入湖架构
● 支持 Partial-Update 能力，基于相同的主键可以各个流实时地打宽，别的支持多种聚合引擎( Deduplicate、Aggregation 等)，在 Paimon 当中能被分钟级给卑鄙各种计算引擎查询
● 支持流入的数据生成变更日记 changelog，给卑鄙更好的流计算，即支持流读
● Paimon 作为湖存储格式，有很强的 Append 处置惩罚，并给 Append 表上多了流读流写、排序后加速查询的能力
3.2 U-App实时基础指标计算

3.2.1 产品模块介绍

友盟基础指标分为实时和离线指标两类，分别对应实时和离线两条计算链路，通过计算新增、活泼和启动等基础指标为客户提供团体概览数据

3.2.2 计算架构

​
（TT–阿里巴巴集团内部 datahub）（OTS–阿里巴巴集团内部 TableStore 表存储服务）
​
上述计算链路即传统的 lambda 架构，数据经过预处置惩罚后写入消息队列，离线链路同步消息队列数据到离线数仓进行加工处置惩罚将计算结果同步到 OTS (类 Hbase 存储)中；实时链路通过 Flink 直接消费消息队列的数据聚合成统计指标后写入 OTS，查询服务将离线和实时两份指标进行统一展示如上图所示。传统 lambda 架构的优缺点如下:
(1) 优点
● 使命容错性比力高
针对早期实时链路不稳定的特点，每天破晓通过离线批处置惩罚计算结果覆盖实时计算结果的方式，保证T+1的离线数据的精确性。对于数据订正的场景可以通过回溯离线数据完成数据的订正；
● 职责边界划分清晰
实时链路只负责增量数据的计算，数据时效性比力高; 离线批处置惩罚链路计算全量历史数据，两条链路职责划分比力明白互相不影响，支持灵活的单独对每条链路进行扩展。
(2) 缺点
● 同时维护实时和离线两套计算逻辑，存储和计算都造成肯定的浪费
实时和离线的计算逻辑是相同的，实时链路只计算当天的结果，第二天破晓再用离线计算去覆盖实时计算结果，带来的问题就是一天的数据实时和离线重复计算，带来资源本钱的浪费；
● 两套计算链路开辟运维本钱比力高，而且涉及实时和链路的数据口径会不一致等问题
两条链路必然带来运维本钱的增加，对于友盟来说实时和离线使命照旧分两个团队在维护。别的因为实时指标每天破晓会被覆盖，可能会出现指标不一致的结果，给客户带来困扰；
实时链接直接基于TT的明细数据进行聚合数据不可查，给排查问题带来困难；
● 对于 U-App 数据量大的特性，基于 Flink 计算实时聚合指标会存在 State 大，实时使命稳定性差的问题
U-App 启动日记每天是千亿级数据量，直接基于明细数据通过 Flink 进行实时聚合，造成Flink使命的state比力大，别的上游使命稍微有颠簸就会对卑鄙计算造成比力大的影响，对使命的稳定性要求比力高，以是我们现在采用的方案是拿资源换稳定，使命资源的 buffer 给的比力足，缺点就是造成肯定资源的浪费。

3.2.3 基于阿里云 Flink + Paimon 的优化方案

针对上述提到的痛点问题，利用 Paimon 自带的聚合引擎能力，将指标的聚合下沉到 Paimon 表中实现，从而统计实时和离线计算链路

​
（TT–阿里巴巴集团内部 datahub）（OTS–阿里巴巴集团内部 TableStore 表存储服务）

1. CREATE TABLE paimon-ump .default.dwd_ump_app_install_paimon_table (
2. app_key       STRING,
3. umid         STRING,
4. cli_datetime     BIGINT,
5. launch_time  BIGINT,
6. launch_flag  INT,
7. new_install_umid  STRING,
8. new_install_flag   INT,
9. app_channel     STRING,
10. country    STRING,
11. province_name    STRING,
12. city_name      STRING,
13. puid         STRING,
14. device_brand     STRING,
15. device_model     STRING,
16. os          STRING,
17. os_version      STRING,
18. sdk_version     STRING,
19. app_version     STRING,
20. inst_datetime    STRING,
21. inst_channel     STRING,
22. inst_app_version   STRING,
23. terminate_duration DOUBLE,
24. resolution   STRING,
25. access STRING,
26. carrier STRING,
27. server_datetime     BIGINT,  
28. upload_traffic     DOUBLE,
29. download_traffic    DOUBLE,
30. app_upgrade INT,
31. hh   STRING,
32. ds   STRING
33. ) PARTITIONED BY (ds)
34. WITH (
35. 'metastore.partitioned-table' = 'true',
36. 'maxcompute.life-cycle' = '360',
37. 'bucket' = '-1',
38. 'sink.parallelism' = '64',
39. 'consumer.expiration-time' = '86400 s',
40. 'snapshot.expire.limit' = '100',
41. 'consumer.ignore-progress' = 'true'
42. );

复制代码

由于 Paimon 的聚合引擎不支持去重，以是设计 DWM 层实现去重逻辑

1. CREATE TABLE `paimon-ump`.`default`.`dwm_ump_app_install_paimon_table` (
2.   app_key           STRING,
3.   dimSTRING,
4.   granularitySTRING,
5.   distinct_idSTRING,
6.   dsSTRING,
7.   PRIMARY KEY (ds, app_key, dim, granularity, distinct_id) NOT ENFORCED
8. )PARTITIONED BY (ds)
9. WITH (
10.   'metastore.partitioned-table' = 'true',
11.   'merge-engine'='first-row',
12.   'first-row.ignore-delete'='true'
13.   'changelog-producer' = 'lookup',
14.   'maxcompute.life-cycle' = '360',
15.   'bucket' = '512',
16.   'sink.parallelism' = '128',
17.   'consumer.expiration-time' = '86400 s',
18.   'snapshot.expire.limit' = '100',
19.   'consumer.ignore-progress' = 'true'
20. );
21. CREATE TABLE `paimon-ump`.`default`.`dws_ump_app_install_paimon_table` (
22.   app_key           STRING,
23.   dimSTRING,
24.   granularitySTRING,
25.   `value`DOUBLE,
26.   dsSTRING,
27.   PRIMARY KEY (ds, app_key, dim, granularity) NOT ENFORCED
28. )PARTITIONED BY (ds)
29. WITH (
30.   'merge-engine'='aggregation',
31.   'metastore.partitioned-table' = 'true',
32.   'changelog-producer' = 'lookup',
33.   'changelog-producer.lookup-wait' = 'false',
34.   'maxcompute.life-cycle' = '360',
35.   'bucket' = '16',
36.   'sink.parallelism' = '16',
37.   'fields.value.aggregate-function' = 'sum',
38.   'consumer.expiration-time' = '86400 s',
39.   'snapshot.expire.limit' = '100',
40.   'consumer.ignore-progress' = 'true'
41. );

复制代码

该方案带来的收益如下：
● 计算资源本钱的节省
在实时基础指标计算场景下，在相同34实时个指标下，用 Paimon 更换 Flink 纯实时计算，计算资源方面可以来了 28% 的资源节省；
在离线指标计算场景下，Paimon 可以直接将离线计算链路使命更换掉不再需要，极大节省离线链路的计算和存储本钱；
● 开辟运维效率的提升
后续使命的开辟和运维不再需要区分实时和离线两条链路，只需要开辟维护一套代码逻辑即可，也不存在数据口径不一致等问题，极大的提高开辟和运维效率；
数据可查，之前直接基于消息队列(TT)的数据不可直接查询，需要同步到离线或其他存储才可以，导致排查问题效率比力低，基于 Paimon 的表可以直接查询，极大提供问题排查和定位的效率；
同时 Paimon 表支持批读批写，支持数据的订正和回溯；
● 计算链路架构的统一
随着实时和批处置惩罚技术的发展，早期的 lambda 架构的缺点在当前业务场景下被渐渐放大变得越来越明显。通过 Paimon + Flink 构建的流式湖仓统一了实时和批处置惩罚链路架构，后续不需要再维护两套计算链路，低落了整个计算链路的复杂性。
3.3 U-App 设备 ID 维表的更新

3.3.1 利用场景

现在设备属性表包罗两部分内容，一部分是设备相干的属性信息；同时还包罗该设备对应的账号的用户属性。现在设备属性维表主要在各种分析模子管理用户属性、人群的用户列表和个体细查等模块。

3.3.2 计算架构

现在友盟设备属性维表的实现方案如上图所示，采用全量+增量的实现方式，这套架构的缺点如下：
● 时延高
现在这套逻辑都是在离线实现的，至少 T + 1 延时，而且需要等全量和增量合并完成后(使命运行2-3小时)卑鄙使命才气利用，数据时效性比力差，用户无法看到当天设置的设备及用户属性信息;
● 存储计算本钱高
每天需要读取全量数据(百亿级)，与增量数据进行全量合并，在全量数据特殊大，增量数据不多时使命计算本钱加高，而且带来资源的浪费;
每天全量表一个分区存储全部数据，在增量数据不多的场景下，意味全量分区存在大量的重复数据，造成存储资源的浪费;
● 架构链路复杂度高
由于设备属性表中带有该设备关联的用户属性信息，加之这种全量和增加合并的实现方式导致链路复杂，导致每天产出全量分区容易有问题导致不能按时产出，新增业务也比力复杂，全量和增量割裂。
3.3.3 基于阿里云Flink + Paimon的办理方案

该方案利用 Paimon 的核心能力:主键更新能力，利用 Paimon Partial Update 引擎的能力，将整理计算链路的时效性从之前的 T+1 低落到分钟级。

1. CREATE TABLE paimon-ump.default.dim_ump_umid_paimon_table (
2. app_key     STRING,
3. umid      STRING,
4. cli_datetime  BIGINT,
5. app_channel   STRING,
6. province_name  STRING,
7. city_name    STRING,
8. idfa      STRING,
9. imei      STRING,
10. oaid      STRING,
11. puid      STRING,
12. zid       STRING,
13. device_brand  STRING,
14. device_model  STRING,
15. os       STRING,
16. os_version   STRING,
17. app_version   STRING,
18. inst_datetime  STRING,
19. inst_channel  STRING,
20. inst_app_version STRING,
21. active_ds    STRING,
22. mobile     STRING,
23. email        STRING,
24. custom_properties  STRING,
25. PRIMARY KEY(app_key,umid) NOT ENFORCED
26. ) COMMENT 'paimon设备属性表'
27. WITH (
28. 'merge-engine'='partial-update',
29. 'metastore.partitioned-table' = 'false',
30. 'changelog-producer' = 'lookup',
31. 'partial-update.ignore-delete' = 'true',
32. 'maxcompute.life-cycle' = '7',
33. 'bucket' = '64',
34. 'tag.automatic-creation' = 'process-time',
35. 'tag.creation-period' = 'daily',
36. 'tag.creation-delay' = '10 m',
37. 'tag.num-retained-max' = '7',
38. 'sink.parallelism' = '64',
39. 'num-sorted-run.stop-trigger' = '2147483647',
40. 'sort-spill-threshold' = '10',
41. 'changelog-producer.lookup-wait' = 'false',
42. 'sequence.field' = 'cli_datetime'  
43. );

复制代码

该方案带来的收益如下:
● 提高数据时效性低落时延
该方案将整个计算链路的时效性从T+1低落到 分钟级，用户当天设置的属性信息当天就可以利用进行分析利用，助力提升业务代价；
● 低落存储计算本钱高
得益于Paimon的 Snapshot 管理，加上 LSM 的文件复用，好比同样是存储 100天的快照，原有离线数仓 100 天需要 100 份的存储，其中在增量数据不多的场景下大部分数据都是重复的，但是Paimon只需要 1 份的存储，大幅节省存储资源；
得益于 LSM 的增量合并能力，此条链路只有增量数据的处置惩罚，没有全量的合并；
● 简化计算链路架构复杂度
简化了之前的全量和增量计算链路，只需要维护一个Flink使命就可以实现全增量合并的目的，提升开辟运维效率。
4. 总结猜测

综上所述，通过 Flink + Paimon 的组合方式在低落计算资源本钱，提高数据时效性，提升开辟运维效率和统一数据链路架构方面，相比于传统的实现方案，体现出相当大的上风。后续友盟会继续跟进 Paimon 的新特性并探索 Paimon 在友盟+具体业务场景中的落地方案。
后续规划：

• 利用 Paimon 对 U-App 自定义事件的计算场景进行优化
  
• 跟进 Paimon 新特性，对现有使命的性能和资源利用进行进一步的优化
  
• 基于 Paimon 自带的 Metric 特性完善 Paimon 使命的监控
  

末了，由衷感谢@之信、 @才智老师在方案落地过程中的引导

---

更多内容

阿里云提供的基于Flink和Paimon的云上流式湖仓办理方案，旨在搭建高效、低延时的流式数据湖仓。此方案利用Flink的实时计算能力，结合Paimon的高效更新能力，实现数据在数仓分层间的实时活动。其上风包罗将数据变更的传递延时从小时级甚至天级低落至分钟级，无需覆写分区即可直接继续变更数据，从而极大地低落了数据更新与订正的本钱。此外，ETL链路的逻辑基于Flink SQL实现，统一了模子并简化了架构，提高了数据处置惩罚效率。点击下方链接了解更多详情。
点击：基于Flink+Paimon搭建流式湖仓


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。