Flink+StarRocks 实时更新数据

看看 StarRocks 在有更新的实时链路里，怎么提供高效的分析查询服务。首先，谈到实时数仓，每个企业每个客户的理解都不尽相同，技术路线的选择也会有所不同。

有的场景处理逻辑非常复杂，借助 Flink 强大的计算能力和丰富的时间语义，客户可以在 Flink 里完成建模。然后，把加工后的结果持久化到消息总线。StarRocks 可以去订阅对应的 Kafka 里的分层数据，再把结果同步过来。对于固定报表类的场景，往往聚焦在 ADS/DWS 层的聚合指标查询，要求查询有极高的性能。这种在 Flink 里计算，在 StarRocks 负责极速查询分析的方案就比较适合。有些场景，数据量不大，利用离线数仓跑批的思路，用调度系统在 StarRocks 里一层一层的做上去，也能实现数仓分层的建设。我们通常说的 OLAP 多维分析，一般会聚焦在 DWD 宽表和 DWS 轻度汇聚层，更灵活的 Adhoc 查询，可能还会对 ODS 原始数据进行查看。

前面聊的一些实时数仓建设的思路，大部分是建立在 append 流的基础上的。假定我们的数据只有追加，没有 Upsert/Delete 操作。在有更新的场景下，不管是增量构建，还是微批调度，都很难保证上层的聚合指标，下钻下来，还能跟明细层对应上。

已经有客户尝试在一些场景下，用 Primary Key 模型做 ODS，保证实时的数据 Upsert/Delete。然后，上面的分层用逻辑视图，保证聚合指标和明细的完全同步吻合。

另外，我们跟 Flink 去结合，如果只支持 append 流是远远不够的。那么 StarRocks 能不能解这个难题呢？答案是肯定的。

可以看到，在新的数据“StarRocks, 1”进来后，如果没有数据回撤，来标记上一轮 Sink 出去的数据失效的情况下，再叠加新进来的数据，就会造成结果的错误。反之，有了 Flink Retract，可以收回上一批次的结论，然后吐出正确的指标。

这种情况下，Flink 端能搞定回撤的问题了，但是 OLAP 端怎么办呢？如果没有高效稳定的 Upsert/Delete 能力，非常容易造成数据的重复和结果的错误。

在一年前，我们在 1.9 版本中发布了新的存储引擎 Primary Key 表模型，在支持实时更新的同时，还能保持查询的高性能。它内置了 OP 字段，以 0 或 1 的形式来标记数据的 Upsert/Delete，恰恰吻合了 Flink 回撤流的数据特征。结合我们提供的 Flink Connector，可以直接将 Flink 的回撤流，对接进 Primary Key 模型。

它基于 Delete+Insert 的方式或者叫 merge-on-write 的方式，实现更新。相比原来 merge-on-read 的 unique 模型，在导入性能几乎不受影响的前提下，查询性能提升了 3-10 倍。

它非常适合 TP->AP 实时同步数据，并加速查询的场景。通过 Flink-CDC 工具，将 TP 业务系统，比如 MySQL 直接同步到 StarRocks，极大的简化了实时分析数据流，简单易用。目前，己经有多个用户在线上系统中采用，是实时数据分析的典型范式。

2022 年，我们对 pk 模型做了持久化主键索引的功能，来降低主键模型的内存开销。原来的主键索引是基于全内存哈希表的，新的持久化索引同样使用了基于 Hash 的设计，并且使用了类似 LSM 的多层设计。

第一层 Hash 为内存 Hash 表，第二层是基于磁盘的 Hash 表结构。为了节约存储空间，使用了类似原全内存 Hash 表的 Shard by length 设计。测试结果显示，内存占用一般只有原来 1/10。由于查询索引本质上，是大量的随机 IO 操作，如果需要持久化索引，推荐使用固态硬盘。

在持久化索引模式下，随着数据持续导入，总内存最高到 70-80G，索引内存最高到 3-4G 左右，内存使用下降非常明显。

另一个 Feature 是，部分列更新的支持。在去年的 FFA 峰会上，我分享了基于聚合模型的 replace_if_not_null，来实现部分列更新的方法。使用这个方法有一定的开发成本，开发者需要把宽表的下标凑齐，没有数据的位置需要显式的去补 null 值。

今天谈的 PK 模型的部分列更新功能，开发成本会更低，数据接入时只需要指定该数据流的相关列名即可。虽然 SR 在多表查询方面性能非常好，但是在一些场景下，用户还是期望大宽表带来的极速性能。

目前，如果想要实现这个效果，有几个常见方案。

1. 在上游数据流中插入一个 Join 模块或者算子，通常使用 Flink 等流式计算平台。用多流 Join，拼成整行数据。

   如果上游多个数据流的数据到达时间不一致，很难设计合适的 window 去在计算引擎里打宽数据，启用 mapState 之类的状态计算又过于定制，迭代效率又是个问题。

2. 用 TP 系统建宽表。上游模块以部分列更新方式写入 TP 系统，再通过 TP 系统，同步给 AP 系统。这样需要额外搭一套 TP 模块和同步模块。
3. 先分模块导入 AP 系统，AP 系统中通过 DML 定期做 Join，后置的定期去刷新大宽表，这样会牺牲一定实时性。

这三种方式都有一定的复杂度，如果 SR 能够直接支持部分列更新，将带来全新的思路，能很好的解决这个问题，简化多流 Join 的链路。

从 2.3 版本开始支持了部分列更新功能，实现方式还是以现有的 Insert+Delete 模式为基础，流程可以参考图中的例子。我要把第一列为 3 的那行最后那个值列 c 改为 y。需要先找到 3 所在的行，然后把跟本次更新无关的列带出来，标记这行为 Delete，然后再追加更新后新的行进去。

剩下的操作就和原来的 Full Row Upsert 类似了。由于采用 Delete+Insert 的方式，实现部份列更新，读写放大问题其实对这种用法造成了一定的限制，特别是对大宽表仅更新很少一部分列的情况。

比如有个表有 10000 列，我们只更新其中的一列。需要先读取其余的 9000 多列，再写入全部 10000 列。所以，我们目前推荐部份列更新，仅在列不是特别多的场景下使用（比如小于 500 列的情况下），并且尽量在固态盘上使用这个功能。为了部份解决这个问题，我们后面打算引入行存，这样能够解决一部分读放大的问题。