Paimon在之家的业务实践

背景在使用Paimon之前，之家的实时/离线数仓分别使用不同的技术方案：离线数仓的方案是使用Hive将数据加工成天/小时级别的表这个方案非常成熟，几乎所有数据仓库团队成员都能熟练掌握。对于业务方而言，其开发和维护成本较低。但是这种方案生成的表通常会有较长时间的数据延迟（如天或小时），数据新鲜度相对较低。此外，在夜间可能还会出现大量的ETL任务竞争资源，导致资源紧张。实时数仓采用的方案是基于Flink、Kafka，关系型数据库和Redis等技术栈，并结合StarRocks做实时OLAP由于数据都是实时处理，因此可以保证数据的新鲜度，一般情况下延迟能做到秒级。然而，当SQL比较复杂时，尤其是存在多个Group by, Join算子时，会导致Flink处理的回撤流翻倍，Flink状态体积非常庞大，使用大量的计算存储资源。这种场景在任务的开发和维护方面可能会带来较大的挑战。任务的开发周期也因此会较离线的方案长很多。在20年底，我们开始调研Iceberg通过流式湖仓的方式在存储上作为实时和离线数据的统一存储方案。Iceberg架构非常的简洁健壮、集成Flink可以做到分钟级别的数据延迟、并且通过自身维护元数据减少了Hive MetaStore的压力、可以灵活且高效地处理表结构变更、支持排序索引等功能可以有效地提升查询效率等优点非常多。但是经过一段时间的使用后，我们发现Iceberg更适合批处理场景，在流场景的一些必要功能的缺失比如增量且有序读取，实时任务在线schema变更，缺少部分更新导致无法满足我们流式湖仓的需求。我们在23年开始调研Apache Paimon，Paimon提供了类似于Iceberg的简洁健壮的架构，并且功能非常强大。与Flink也集成得非常完善，提供了增量且有序的数据读取、部分更新等能力，结合Flink CDC可以实现整库同步，在线Schema变更等，满足了我们对于流式湖仓的需求。而且Paimon作为一个相对较新的数据湖，没有太多的历史负担，这对于其他数据湖来说这一点有着非常巨大的优势。我们最终使用了Paimon作为存储流批一体的解决方案。实时和离线数据共享同一份存储，这样做降低了开发和维护的难度，提高了数据仓库的整体数据新鲜度。此外，我们还通过使用StarRocks建立物化视图和使用Sort Compaction功能等技术手段来进一步提升查询效率，节省计算资源。业务实践在下文中将给大家分享之家基于Paimon的业务实践：新用户转化分析使用Paimon主键表的部分更新功能使用StarRocks建物化视图加速Paimon表的查询流量日志入湖使用Paimon的Append表使用Paimon的Sort Compaction功能加速Paimon表的查询，提升查询效率，减少资源消耗资源入湖使用Paimon主键表的Upsert功能一. 新用户转化分析：新用户转化分析是一种对新用户在产品中行为变化的分析方法，通过观察新用户在其首次进入APP的行为，分析他们在后续阶段的转化过程。这种分析对于理解产品的用户采用情况、改进用户体验、提高用户留存率等方面具有重要的意义。为了实时获取新用户进入之家APP后在同步落地页的转化情况，需将用户的行为路径和订单数据关联，通过综合分析新用户在产品或服务中的行为，可以更加全面地改进产品体验，并制定更具针对性的策略，从而提高用户转化率。在这个场景中，我们使用到了Paimon的部分更新功能，按照用户ID部分更新对应主题的数据。?实现1. 够建新用户宽表CREATE ?TABLE if not exists new_user_transform ( ? ? ? ? ? ? ? user_id String, ? ? ? ? ? ? ? new_user_type String, ? ? ? ? ? ? ? channel_name String, ? ? ? ? ? ? ? land_page String,.... ? ? ? ? ? ? ? ?primary key (deviceid,dt) not enforced ? ? ? ? ? ? ? ?) partitioned by (dt) ? ? ? ? ? ? ? ?WITH ( ? ? ? ? ? ? ? ?'bucket-key' ='deviceid', ? ? ? ? ? ? ? ?'bucket' = 'xx', ? ? ? ? ? ? ? ?'full-compaction.delta-commits'='5', ? ? ? ? ? ? ? ?'merge-engine' ='partial-update', ? ? ? ? ? ? ? ?'partial-update.ignore-delete'='true', ? ? ? ? ? ? ? ?)2. 处理数据写入到Paimon2.1 通过Flink的interval window 计算新用户在5分钟内进入过的承接页CREATE TEMPORARY VIEW new_user_view ASselectl.deviceid,r.page_id as land_pagefrom(select time_ltz, deviceid, channelid, proctimefrom new_user) lleft join(select page_id, deviceid, time_ltzfrom user_page_view_log) ron upper(l.deviceid) = upper(r.deviceid)and r.time_ltz > l.time_ltz and r.time_ltz