HASH GAME - Online Skill Game GET 300

　　本文整理自阿里云智能集团苏轩楠老师在Flink Forward Asia 2024论坛的分享，涵盖流式湖仓架构的背景介绍、技术演进和未来发展规划。背景部分介绍了ODS、DWD、DWS三层数据架构及关键组件Flink与Paimon的作用；技术演进讨论了全量与增量数据处理优化、宽表构建及Compaction操作的改进；发展规划则展望了Range Partition、Materialized Table等新功能的应用前景。通过这些优化，系统不仅简化了复杂度，还提升了实时与离线处理的灵活性和效率。

　　这是一个典型的流式湖仓架构，首先业务数据会存储在 MySQL 表中，然后借助 Flink 及其 CDC Connector 的作业，将这些数据库的数据同步到 Paimon 的 ODS 层中，从而构成 ODS 层数据。这层数据实际上包含了 MySQL 的全量数据，并且会根据 MySQL 的更新实时地进行相应的更新。在有了 ODS 层数据之后通常会进行数据过滤，并进行数据的 Join 操作，以生成一个宽表，这就是 DWD 层的数据。 DWD 层数据形成后会进一步进行数据过滤、数据的聚合和打宽表等操作，以生成 DWS 层的数据，用于进行指标的统计。这就是一个简化的流式湖仓分层设计。在这个分层设计中， Apache Flink 与 Paimon 是两个至关重要的组件。

　　但幸运的是 Flink 提供了一个非常有效的机制来实现宽表构建，即 Partial Update 。如图所示它可以使用两个 Flink 分别对一个表中的不同字段进行更新。例如左边的表只需要对 Column A 进行更新，并将 Column B 设置为 null ，而另一个 Sink 则可以对 Column B 进行更新。这样当下游读取这个表时，它会自动将这两个部分合并起来，将 null 值替换为对应的实际值。

　　在实际使用过程中经常会遇到一个问题：当一个作业包含多个Source ，并将数据写入同一个 Paimon 表时，如果多个 Flink 尝试同时对该表进行 Compaction 操作，Paimon 通常不支持这种行为。这会导致作业在执行 Compaction 时失败，进而引发作业持续 Failover ，最终导致作业不可用。为了解决这个问题，用户可以通过配置来关闭作业的自动 Compaction 功能。然而，这样做意味着需要启动另一个专门的作业来对该表执行 Compaction 操作。在很多情况下，用户可能只希望通过一个作业的多个 Flink 来完成必要的 Passbook 操作，而不希望额外启动一个专门的作业来进行 Compaction 。但遗憾的是当前可能无法直接让该作业自行处理 Compaction 。

　　为了实现这一目标需要在 Flink 的 Single Planner 中进行一些改造。改造完成后， Flink Single Planner 将能够自动识别是否多个 Flink 组件正在向同一张表写入数据。在满足特定条件时将多个 Flink 的上游结果进行 Union 操作，并仅使用一个 Flink 组件来统一写入所有数据。这样在进行 Compaction 操作时，就只有一个判断逻辑，从而避免了之前提到的冲突问题。

　　由于 Paimon 通常是 bucket 存储的，在主键表的情况下，分桶是一种常见的做法。然而为什么事实表不能根据Paimon表的分桶分配方式进行 shuffle 呢？原因是目前还没有办法让 Paimon 表指定事实表的 shuffle 方式。因此在 Flink Single Planner 中引入了一个名为“Support Lookup Custom Shuffle”的接口。这个接口的本质是允许 connector 为维表实现指定事实表的数据 shuffle 方式。有了这个接口后 Paimon 表的维表就能够执行一些特定的操作了。

　　首先关于 Paimon 的维表， Paimon 的主键表中包含两种分Bucket 的方式。最简单的一种是 Fixed Bucket 。 Fixed Bucket 指的是在作业定义时，而非在 Paimon 表定义时，就已经确定了 Bucket 的数量。对于任意一条数据，其对应的 Bucket 可以通过一个简单公式计算得出。本质上这个计算过程是对 Bucket Key 取哈希值，然后再对 Bucket 的总数取模，从而确定数据具体属于哪个 Bucket 。实际上只需要让事实表也按照同样的方式进行 shuffle 。例如在事实表中可以将 K 1 和 K 2 分配到 Lookup 算子上。这个 Lookup 算子知道，它只需要读取 Bucket 1 的数据，并且只需将 Bucket 1 的数据存储在本地即可。通过这种方式，可以大大降低每个 Lookup 算子的并发量，减少其需要读取的 Paimon 数据量。同时也能降低其实际要存储到本地以及内存中的数据量。

　　但是依然可以做一些事情来应对这种情况，即可以通 Custom Shuffle 接口来指定其 Sort of 的方式。这里的分配方式是指根据 Join Key 取一个哈希值，然后在取模时根据下游 Subtask 的数量，即 Lookup Join 的 Subtask 数量来进行。这时每个 Lookup 算子，或者说每个 Lookup 的并发实例，在读取维表时就会知道它可能会接收到哪些与事实表相关的数据。因此它就可以对其存储的缓存进行一些裁剪。

　　比如说，虽然事实表仍然是按照如 K 1 、 K 3 这样的 Key 发送给上游的并发实例，但这些并发实例在读取数据时还是需要全量读取。但是当数据存储到本地时可以进行过滤，只存储与 K 1 和 K 3 相关的数据。因为他知道事实表的分配算法策略是他指定的，所以他可以只存储与 K 1 、 K 3 相关的维表数据。尽管在读取时仍然需要访问全量的数据，但实际上他只需要在本地保存一部分维表数据。这就是针对 Lookup Join 所做的一个优化策略。

　　比如说在这个例子中，设定 N 等于 2 ，事实表中的 K 1 可能会被分发到第一个和第二个并发实例上，从而尽可能地将一个 Hot Key 打散。由于需要进行一个类似于复制的操作，因此第一个和第二个 Lookup 并发算子都需要额外读取一个 Bucket 。这实际上是一个 Trade-off 。如果将 N 设置得很大，那么数据被打散得会更加平均，但每个算子需要读取的数据量也会相应增加。以上就是关于数据读写以及查询优化方面的一些讲解。

　　具体来说在 1.18 及 Paimon 0.9 之前的版本中，要求在执行 COPY 或类似操作时，必须按照定义的顺序填入所有参数，即使某些参数有默认值且用户并不关心，也必须用空字符串来替代。例如在进行 Compact 操作时，如果希望对名为 Default 的表进行 Compact ，并将并发度设置为 4 ，理论上只需要传递两个参数。但在实际操作中，在参数列表中需要填入三个空字符串，这无疑增加了用户的操作难度，降低了使用的便捷性。

　　在 Flink 1.18 之后引入了一个名为 Named Argument 的功能。这一功能允许以任意顺序传入参数，并且只需填写必要的参数即可。在相同的场景下只需要填写表名和并发度配置即可。此外之前提到 Paimon 提供了许多 Action ，包括进行 Compaction 、 Snapshot 管理，以及 Clone Table 或 Clone Database 等操作，然而在之前的版本中发现许多 Action 并没有对应的 Flink Procedure 实现。这导致在使用时，需要通过 Action 的流程来完成操作。

　　Action 的流程通常是怎样的呢？以创建一个表的操作为例。在 0.9版本之前，当 Paimon 还没有实现对应的 Procedure 来创建空表时，用户首先需要从 Paimon 官网上下载一个 Action 的 JAR 包。然后，用户需要将这个 JAR 包上传到 Flink 的运行环境上。接下来，用户需要通过执行 Flink Run 命令来启动创建空表的作业。同时，作业的参数也需要通过命令行参数来传入。这样的操作方式显然不够便捷。

　　接下来展望一下未来的技术发展方向。首先关于前面提到的 Range Partition ，即无主键表，它已经实现了对用户指定列的排序功能，这一功能已经在 Paimon 0.9 版本中发布。此外刚刚提到的 Procedure 易用性优化也已经在该版本中得以呈现。现在大家应该都能开始使用这些功能了。值得一提的是 Paimon 是首个在第一时间对接了 Flink 2.0 的 Materialized Table 功能的平台，这一功能将在 Paimon 1.0 版本上线时与大家见面。

　　本文介绍了阿里云实时数仓Hologres负责人姜伟华在Flink Forward Asia 2024上的分享，涵盖实时数仓的发展历程、从实时数仓到实时湖仓的演进，以及总结。文章通过三代实时数仓架构的演变，详细解析了Lambda架构、Kafka实时数仓分层+OLAP、Hologres实时数仓分层复用等方案，并探讨了未来从实时数仓到实时湖仓的演进方向。最后，结合实际案例和Demo展示了Hologres + Flink + Paimon在实时湖仓中的应用，帮助用户根据业务需求选择合适的方案。

　　本文整理自鹰角网络大数据开发工程师朱正军在Flink Forward Asia 2024上的分享，主要涵盖四个方面：鹰角数据平台架构、数据湖选型、湖仓一体建设及未来展望。文章详细介绍了鹰角如何构建基于Paimon的数据湖，解决了Hudi入湖的痛点，并通过Trino引擎和Ranger权限管理实现高效的数据查询与管控。此外，还探讨了湖仓一体平台的落地效果及未来技术发展方向，包括Trino与Paimon的集成增强、StarRocks的应用以及Paimon全面替换Hive的计划。