流式高频写入：Apache Hudi 非阻塞并发控制详解

介绍

在流式摄取场景中，有很多使用案例需要从多个流式源进行并发摄取。用户可以将所有上游源输入合并到一个下游表中，以收集记录，以便跨联合查询进行统一访问。另一种非常常见的情况是将多个流源连接在一起以补充记录的维度，以构建一个宽维表，其中每个源流都获取具有部分表架构字段的记录。对多流并发摄取的普遍而强烈的需求一直存在。自从 Hudi 支持流式摄取和处理以来，Hudi 社区已经从用户那里收集了大量反馈。

从 Hudi 1.0.0[1] 开始，我们很高兴地宣布为 Apache Hudi 推出一种新的通用并发模型——非阻塞并发控制 （NBCC），该模型针对流处理或高争用/频繁写入场景。与乐观并发控制相比，在乐观并发控制[2]中，如果有一丝争用，写入器会中止事务，这项创新允许对同一个 Hudi 表进行多次流式写入，而不会产生任何冲突解决的开销，同时保留流式系统中的事件时间排序[3]语义，以及异步表服务，例如压缩、归档和清理。

NBCC 无缝运行，无需任何新的基础设施或运营开销。在本博客的后续部分中，我们将简要介绍 Hudi 关于数据文件布局和时间生成的 TrueTime 语义的内部结构，这是讨论 NBCC 的先决条件。接下来，我们将深入研究 NBCC 的设计和工作流程，然后通过一个简单的 SQL 演示来展示与 NBCC 相关的配置选项。该博客将对 NBCC 的未来工作发表见解。

之前的设计

了解 Hudi 存储布局[4]以及它演变/管理数据版本非常重要。在 1.0.0 之前的旧版本中，Hudi 以 FileGroup 为单位组织数据文件。每个文件组包含多个 FileSlice。此文件组上的每次压缩都会生成一个新的文件切片。每个文件切片可以包含一个可选的基本文件（列式文件格式，如 Apache Parquet 或 ORC）和多个日志文件（Apache Avro 或 Parquet 中的行文件格式）。

基本文件名中的时间戳是写入它的 compaction 的 instant 时间，在 Hudi 的概念中也称为 “requested instant time”。日志文件名中的时间戳与当前文件切片基准即时时间的时间戳相同。具有相同即时时间的数据文件属于一个文件切片。实际上，文件组表示基本文件 （检查点） 后跟日志文件 （增量） 和基本文件 （检查点） 的线性有序序列。

日志文件中的即时时间命名约定在并发模式下成为哈希限制。每个日志文件都包含来自多个提交的增量更改。每个编写器都需要查询文件布局以获取基本即时时间，并在刷新记录之前找出完整的文件名。一个更严重的问题是，基本即时时间可以随着异步压缩的推进而变化。为了使日志编写者的基本即时时间具有确定性，Hudi 强制在写入提交和压缩计划之间设置计划序列：只有在没有正在进行的摄取到 Hudi 表中时，才能计划压缩。否则日志文件可能会使用错误的基本即时时间写入，这可能会导致数据丢失。这意味着压缩调度可能会阻止并发模式下的所有写入器。

NBCC 设计

为了解决这些痛点，从 1.0.0 开始，Hudi 引入了一种新的存储布局，该布局基于操作的请求时间和完成时间，将它们视为间隔。1.x Hudi 中的每个提交都有两个重要的时间概念[5]：即时时间（或请求时间）和完成时间。所有生成的时间戳都是全局单调递增的。Hudi 现在只使用请求的写入即时时间，而不是将基本即时时间放在日志文件名中。在文件切片期间， Hudi 会查询每个日志文件的完成时间以及即时时间，并且我们有一个新的文件切片规则：

日志文件属于最大基本请求时间小于（或等于）其完成时间的文件切片。[^1]

借助新文件布局的灵活性，消除了日志编写器查询基本即时时间的开销，并且可以在任何即时时间随时随地安排压缩。有关更多信息，请参阅 RFC-66[6]。

真实时间 API

为了保证时间戳生成的单调性，Hudi 从 1.x 版本开始引入了“TrueTime API[7]”。基本上，有两种方法可以使时间生成单调递增，这与 TrueTime 语义一致：

• • 一个全局锁，用于使用互斥锁保护时间生成，并等待分布式主机上估计的最大允许 clock skew ;
• • 全局同步的时间生成服务，例如 Google Spanner Time Service，服务本身可以保证单调性。

Hudi 现在在第一个解决方案中实现了“TrueTime”语义，支持可配置的最长等待时间。

LSM 时间线

新的文件布局需要从 Instant Time 开始进行高效查询以获取完成时间。Hudi 从 1.x 开始重新实现了存档的时间线，新的存档时间线数据文件被组织成一个 LSM 树[8]，以支持快速的时间范围过滤查询，并对其进行即时时间跳跃。

借助强大的新文件布局，实现非阻塞并发控制非常简单。该函数是通过 Flink 的 MOR 表上的简单 bucket 索引实现的。存储桶索引可确保多个工作负载的固定记录键到文件组的映射。日志编写器将记录写入 avro 日志，压缩表服务将负责解决冲突。因为每个日志文件名都包含即时时间，每条记录都包含事件时间排序字段，所以 Hudi 阅读器可以按自然顺序（处理时间序列）或事件时间顺序合并记录。

并发模式应配置为 NON_BLOCKING_CONCURRENCY_CONTROL ，可以在一个作业上启用表服务，并为其他作业禁用它。

Flink SQL 演示

这是一个演示，展示了 2 个摄取到同一下游表中的管道，两个 sink 表视图共享相同的表路径。

-- NB-CC demo

-- The source table
CREATE TABLE sourceT (
  uuid varchar(20),
  name varchar(10),
  age int,
  ts timestamp(3),
  `partition` as 'par1'
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '200'
);

-- table view for writer1
create table t1(
  uuid varchar(20),
  name varchar(10),
  age int,
  ts timestamp(3),
  `partition` varchar(20)
)
with (
  'connector' = 'hudi',
  'path' = '/Users/chenyuzhao/workspace/hudi-demo/t1',
  'table.type' = 'MERGE_ON_READ',
  'index.type' = 'BUCKET',
  'hoodie.write.concurrency.mode' = 'NON_BLOCKING_CONCURRENCY_CONTROL',
  'write.tasks' = '2'
);

insert into t1/*+options('metadata.enabled'='true')*/ select * from sourceT;

-- table view for writer2
-- compaction and cleaning are disabled because writer1 has taken care of it.
create table t1_2(
  uuid varchar(20),
  name varchar(10),
  age int,
  ts timestamp(3),
  `partition` varchar(20)
)
with (
  'connector' = 'hudi',
  'path' = '/Users/chenyuzhao/workspace/hudi-demo/t1',
  'table.type' = 'MERGE_ON_READ',
  'index.type' = 'BUCKET',
  'hoodie.write.concurrency.mode' = 'NON_BLOCKING_CONCURRENCY_CONTROL',
  'write.tasks' = '2',
  'compaction.schedule.enabled' = 'false',
  'compaction.async.enabled' = 'false',
  'clean.async.enabled' = 'false'
);

-- executes the ingestion workloads
insert into t1 select * from sourceT;
insert into t1_2 select * from sourceT;

未来路线图

虽然非阻塞并发控制对于流式处理用户来说是一项非常强大的功能，但它是解决多个写入器冲突的通用解决方案，以下是一些改进 Hudi 核心功能的计划：

• • NBCC 对元数据表的支持
• • NBCC 用于聚簇
• • NBCC 用于其他索引类型

引用链接

[1] Hudi 1.0.0: https://hudi.apache.org/releases/release-1.0.0 [2] 乐观并发控制相比，在乐观并发控制: https://hudi.apache.org/blog/2021/12/16/lakehouse-concurrency-control-are-we-too-optimistic/ [3] 事件时间排序: https://www.oreilly.com/radar/the-world-beyond-batch-streaming-101/ [4] 存储布局: https://hudi.apache.org/docs/next/storage_layouts [5] 重要的时间概念: https://hudi.apache.org/docs/next/timeline [6] RFC-66: https://github.com/apache/hudi/blob/master/rfc/rfc-66/rfc-66.md [7] TrueTime API: https://hudi.apache.org/docs/next/timeline#timeline-components [8] 成一个 LSM 树: https://hudi.apache.org/docs/next/timeline#lsm-timeline-history