Flink Table Store v0.2 应用场景和核心功能「建议收藏」

大家好，欢迎来到IT知识分享网。

导读：本文讲分享Flink Table Store v0.2适用的应用场景，以及支撑这些应用场景的核心功能。同时介绍Flink Table Store未来的规划。

今天的主题围绕以下四点展开：

• 应用场景
• 核心功能
• 未来展望
• 项目信息

01

应用场景

首先了解一下Flink Table Store v0.2的架构（如图 1），它首先是一个湖存储，以低成本无服务的方式存储大量数据。湖存储中通过Manifest管理文件，每个Bucket中是一个 LSM Tree。在湖存储上也支持了和Kafka的集成，让你一张表同时存储离线和实时数据。

上游支持Streaming或Batch写入数据，下游支持Flink Streaming、Flink Batch、Hive、Spark和Trino消费。湖存储的数据存储在DFS上，也可以使用Object Store、Cloud Storage来存储。

图1 Flink Table Store v0.2的架构

接下来分享v0.2主要面向的四个场景。

1. 场景一：离线数仓加速

第一个场景是离线数仓的加速（如图 2），这个场景中是典型湖存储的应用场景，支持Flink Streaming写入，下游支持各种计算引擎的批量查询（Batch Read）或OLAP查询。

目前Hudi、Iceberg也具有以上能力，那Flink Table Store v0.2 对比它们有什么特点？

第一个特点，Flink Table Store 湖存储面向Flink写入，需要支持Flink Streaming SQL产生的实时更新的所有类型。包括主键更新、无主键更新和AppendOnly数据。

第二个特点，由于它面向Flink Streaming SQL，因此它支持实时更新。比如业界把Hive、Iceberg作为偏离线的数仓，Hudi作为近实时的数仓。而Table Store想要支持的是更快更大吞吐的实时更新。

Table Store需要哪些能力满足以上两个特点？从写入和读取两个方面来说明。

在写端，首先，所有的更新都应基于存储自身而不应依赖Flink的状态，这样的好处是易用性非常高。由于存储的更新不依赖写入作业的状态，写入作业可以在流处理和批处理之间随意切换；其次，存储是基于非常高效的LSM的Sorted Merge，因此它的更新性能非常高。

对于读端，基于LSM非常快速的Sorted Merge，可以实现高效的 MOR（Merge On Read），有序的合并开销非常低，对比Iceberg或者Hive这种Copy On Write（COW）表的查询时延差不多。同时，由于数据按照主键排序，相当于主键上有索引。如果有基于主键或部分主键的Filter或Range Filter 查询，查询速度会非常快。因为底层基于排序的查询已经排除掉大量的文件，点查最快可以在100毫秒返回。

图 2 离线数仓加速

2. 场景二：Partial Update（COALESCE）

如图3中的 CREATE TABLE 语句中定义了一个 pk，然后定义 merge-engine 为partial-update。图3中INSERT语句写入数据时merge两张表，每张表更新的字段不同，其中表Src1更新column_1，表Scr2更新column_2。这里写的方式与传统的 Partial Update 有些区别，是将不需要更新的字段设置为 NULL 。这里的 Partial Update类似于SQL引擎中的COALESCE函数，非NULL字段更新。

Partial Update适用于基于主键的大宽表的更新，比如实时更新10个表，它们基于相同主键更新不同字段。在读端可以进行实时批量查询。(流消费还在开发中)

Table Store 在支持这个应用场景的特点与第一个场景类似，支持写端和读端的所有四个特点（如图3）。

图3 Partial Update

3. 场景三：预聚合 Rollup

由于Table Store是基于LSM面向更新的湖存储，因此可以做一些更有趣的事，比如预聚合（如图 4）。图4中的SQL建表语句中定义了merge-engine为aggregation，分别对column_1和column_2做sum和max聚合操作，当写入数据时就会自动的进行merge，实现与Flink流中使用agg函数类似的效果。

可以看到这两个能力都有一定的限制，比 如agg函数的计算逻辑不能太复杂，如count distinct这种操作就无法实现。Flink Table Store与Flink Streaming作业比起来通用性较弱。那为什么湖存储还需要提供场景二和场景三的能力？因为在存储侧提供这些能力成本要低很多。在存储侧提供这种能力时不需要TTL，而Flink Streaming 作业就有State的 TTL，当Flink Streaming有错误的延迟事件数据时会导致错误的结果。而存储侧提供这种能力使得操作更宽泛。

图 4 预聚合 Rollup

4. 场景四：实时数仓增强

前面三个场景主要面向湖存储本身，而第四个场景面向实时数仓增强（如图5）。

如图5所示的SQL中指定Log System为kafka及其相关的配置信息，从而使表具有两种物理存储形态——湖存储和Log System。在Flink Streaming写入时会双写两个系统，且能通过offset处理好两个存储系统之间的一致性。传统的实时数仓只有Kafka，且实时数据不可查，而Table Store可以很方便的通过批处理引擎或者OLAP引擎查询中间表的状态，以及做数据印证和灵活的OLAP。

另一个能力是通过流处理引擎处理数据时可以实现Hybird的Backfill读，也就是先读取湖存储，然后读取Log Store。比如查询一个月的数据等。这种能力与Kafka Tired Storage 类似，因为消息引擎也在解决自身存储能力不足的问题（不能无限的存储）。消息引擎一般有的TTL（Time-To-Live）retention指标来保存最近某段时间内的数据，因此Kafka设计了Tired Storage实现数据存储在DFS上。但是消息队列设计这种存储存在一个问题，数据无法被各种计算引擎灵活地查询。Table Store就解决了这个问题，可以让外部计算引擎灵活的查询数据。

那么就会引出另外一个问题，要把Table Store当作Kafka的Tired Storage，就要保证读数据的顺序与独立使用Kafka是一样的。如果没有定义PK和write-mode是AppendOnly模式，这种情况下无论Hybrid读湖存储还是Log System它读到的数据顺序是一样的。

图5 实时数仓增强

以上是Table Store v0.2四个主要应用场景，那么它通过哪些能力来支撑这四个场景呢？下面介绍一下它的核心功能。

—

02

核心功能

1. 湖存储的结构

Table Store v0.1提供了湖存储的能力，其湖存储的特点有：

第一，提供 Snapshot级别的事务语义。

第二，支持对象存储上的大规模数据存储。

如图6展示了v0.1在结构上对这两个特点的支持，可以看到重要的细粒度的Meta数据都放在DFS上。

每个Snapshot相当于一次Commit，每个Snapshot更新的文件存放在同样位于DFS的Manifest中。但所有更新的文件不能只记录在一个Manifest中，因为当Table的数据达到TB甚至PB级时，这种量级的数据文件自身的Meta就非常大。湖存储通过分层的手段支持Manifest的增量更新（如图6中的m0、m1等等），从而实现文件Meta可更新。Table Store通过这种分层结构来支持大规模数据存储，同时达到Snapshot级别的事务语义。

以下是Table Store的上层结构。

图6 湖存储结构

2. Table Store v0.1分区内部结构

在v0.1分区内部，用户在定义表的时候需要设置其包含的Bucket数量。这样每个分区中包含多个独立的Bucket，数据通过Hash分布到Bucket中。每个Bucket中都是一个支持更新的LSM Tree。由于LSM Tree是支持更新的数据结构，避免了单个Bucket 的数据存储在一个文件中。

如果Bucket内只有一个文件，每次更新时需要对文件中的数据全部重写，当文件中的数据达到GB量级时的代价就非常大。但是又不能把Bucket数量设置太多，当Bucket number的数量很大时每个Bucket都对应一个小文件，此时对对象存储或者DFS的压力很大。

所以Table Store在Bucket中是LSM Tree的文件结构，而不是单个文件，以此来支持更新以及查询性能的提升。

图7 Table Store v0.1分区内部结构

3. Table Store的生产化

以上是Table Store v0.1提供的基本结构，在此基础上Table Store v0.2 为了更接近生产化做了很多改进。生产化需要哪些能力？主要包括API和生态两个方面。

首先，在API方面提供了Catalog，如图8中的Flink SQL创建一个CATALOG 指定它的type为Table Store。Catalog的元数据默认存在File System上，可通过metastore配置元数据的存储位置，比如Hive，这样就可以在hive中直接读这些表。

Catalog的使用也比较简单，直接USE CATELOG，然后再创建表。在创建表时你可以选择性的提供Kafka的LogSystem信息。

图8 Table Store Catalog

其次，是Table Store的生态（如图9）。

由于v0.1只支持Flink Batch，严重束缚了Table Store的生产可用性。v0.2将核心支持Hive、Spark和Trino这三个计算引擎。使用方式如图9所示，对于Hive SQL如果已经配置Metastore的Catalog可以直接查；如果没有配置Metastore则需要创建外表并指定LOACTION信息才可以直接查询。对于Spark SQL可以直接配置Catalog，进而看到Table Store的所有表。

图9 生态

4. Table Store中Bucket的更新

前面提到，一个分区内Bucket的数量使固定的。建表时如果定义的Bucket数量太少，更新性能太差无法满足吞吐量，Bucket数量太多导致小文件太多导致查询性能很差。

v0.2引入了Rescale的能力（如图10），当Bucket数量太少时通过Rescale实现扩展。Rescale的前提是最好更新当前或者未来的分区，而不影响老的分区。如果影响老的分区就意味着Rescale需要将全部数据重写一遍，业务上是不允许这种操作的。Rescale 的设计是只影响新分区，新的分区会使用更新后的Bucket个数，而老分区不动。如果当前分区的性能太低怎么办？需要暂停流写作业，使用Batch作业Rescale当前分区，然后再恢复流写作业，从而实现调整当前分区的Bucket数量。

图10 Change Bucket

5. Append Only模式

前面也提到Table Store可以实现类似Kafka的Tiered Storage的能力，因此需要存储侧支持Append Only模式（如图11）。该模式的特点是：

• 由于没有合并和更新，所以写入成本非常低，可以作为离线表使用。
• 流读的顺序与输入序一样，提供与Kafka流读相同的体验；同时数据是可查询的，可以通过不同的查询引擎查询所有的数据。
• 提供自动Compaction功能，由于写时会产生大量小文件，需要通过Compaction合并小文件，避免小文过多件导致的性能问题。

图11 Append Only模式

—

03

未来展望

接下来分享一下Table Store未来的规划及长期展望。

1. 满足Flink SQL对存储的需求

Table Store首要目的是满足Flink SQL对存储的需求。这些需求包括（如图12）：

①最基本的是对消息队列的需求。

②表数据实现OLAP可查的功能。

③支持Batch ETL的写入和大规模Scan。

④在以上三个能力的基础上支持Dim Lookup，也就是Flink Stream SQL中Dim Join的能力。这种点查无法达到Hbase那种低延迟毫秒级的点查，这里的点查是基于Batch的点查即流计算中批次的点查。

有了以上四个能力就可以满足Flink SQL对存储的所有需求。

图12 满足Flink SQL对存储的需求

2. 满足不同Tradeoff的选择

Table Store未来的目标是能在新鲜度、成本、查询延时（如图13）这三个指标之间达到一个平衡。

成本包括服务器、费用以及开发成本等；新鲜度是指数据从产生到查询整个过程消耗的时间。查询延时是指用户的查询耗时。Table Store的目标是让用户在这三个Tradeoff之间灵活的选择。比如，当需要较短的查询延时，就要降低对新鲜度低以及成本的要求。如果要更好的新鲜度，那么数据准备的过程就不应太复杂且需要更高的成本。

图13 满足不同Tradeoff的选择

3. Flink Table Store的架构

目前Table Store只是一个湖存储，以后作为长期的一个架构如图14所示，主要包括湖存储、DFS、Log System三部分。后续的架构正在POC当中，包括在湖存储基础上支持加速能力的Service，达到Flink Streaming Pipeline基于Service实现流写和流读。同时存储本身支持很强的OLAP查询性能。由于实时Pipeline的成本很高，如果要查询历史数据比如一年的数据怎么办？Service作为湖存储的加速，所有的数据都会周期性的存储到湖存储上，这样就可以实现Batch Pipeline查询，比如通过Hive、Spark、Flink Batch或者Presto来查询一整年的数据。批处理只通过Metastore和湖存储实现，不会干扰Service，所以不会带来很大的成本。

整体上想通过这套架构在新鲜度、成本、查询延时三个指标之间有一定的Tradeoff，在存储所有的数据的同时满足实时OLAP的能力。

图14 Flink Table Store架构

4. Dim Join，存算分离

除了流读流写、批读批写，Flink Table Store也具备Dim Join的能力，即支持一定程度的点查（如图15）。由于LSM支持类似点查的能力，Table Store类似于HBase的计算存储分离的查询，在某个节点，比如Dim Join的Task上会构建一个点查的Cache（如图15）。其中的Cache具有从内存到本地磁盘，再到DFS的分层结构。

图15 Dim Join，计算存储分离

—

04

项目信息

Flink Table Store是Apache Flink的子项目。项目开源地址：

https://github.com/apache/flink-table-store

项目文档：

https://nightlies.apache.org/flink/flink-table-store-docs-master/

邮件列表：

• 讨论邮件列表：dev@flink.apache.org
• 用户列表：user@flink.apache.org
• 用户中文列表：user-zh@flink.apache.org

v0.2已在2022年8月份发布！

—

05

问答环节

Q：Flink Table Store如何管理数据权限？有计划支持库表的权限管理吗？

A：在湖存储中数据权限依赖于文件系统的权限，目前Flink Table Store的权限认证只有文件级别的权限，或者Hive Metastore的权限认证，依赖底层存储的权限管理，目前没有支持库表的权限管理。

Q：目前有对Hadoop组件，如Hive的Kerberos认证环境支持吗？

A：目前通过Flink来支持Kerberos认证的环境，后续可能要做一些调整。

Q：对比Hudi、Iceberg等，Flink Table Store的优势在什么地方？

A：它们面向的场景不同，第一Flink Table Store的特点是不仅支持Update with PK还支持Update without PK。其二，Flink Table Store是面向更大吞吐更实时的更新，成本低，吞吐大。同时读端有很好的MOR性能和主键索引加速能力，点查的效率非常高。另外，Flink Table Store更新是无状态的，不依赖Flink SQL，这样Flink可以随时启停。欢迎大家来测试。

Q：LSM也支持快照，是否可以做到像Iceberg的增量读取？

A：可以的，从场景四可以看到，Log System（Kafka）是可选的，也可以不配置Kafka，直接增量读取湖存储。Flink Table Store提供的特点的是，有更多的顺序保证，更好的流消费性能。比如在AppendOnly模式下增量读可以保证输入序读取，它可以当作一个Queue使用，只是延时相对比较高。

Q：后期的路线图是如何规划的？

A：从Flink Table Store的架构图（图14），目前v0.2的目标是把Lake Store做好。Service的核心在于能提供很好的新鲜度以及在线离线的整合能力，明年计划发布成熟的Service版本。

今天的分享就到这里，谢谢大家。

分享嘉宾：李劲松 阿里巴巴 技术专家

编辑整理：刘浩平 东北大学

出品平台：DataFunTalk

01/分享嘉宾

李劲松｜阿里巴巴技术专家

目前就职于阿里云开源大数据，长期从事分布式流 / 批处理系统领域的工作，也对数据湖和 OLAP MPP 有一些研究。我是 Apache Beam / Flink / Iceberg 的 Committer，对底层调度、通信机制、用户模型、SQL 流批计算、存储有一定了解。目前专注于 Flink Table Store 项目的开发，希望给 Flink 带来一个最适合的存储。

02/关于我们

DataFun：专注于大数据、人工智能技术应用的分享与交流。发起于2017年，在北京、上海、深圳、杭州等城市举办超过100+线下和100+线上沙龙、论坛及峰会，已邀请超过2000位专家和学者参与分享。其公众号 DataFunTalk 累计生产原创文章700+，百万+阅读，14万+精准粉丝。