改进Apache Hudi的标记机制

Hudi 支持在写入操作期间对存储上未提交的数据进行全自动清理。 Apache Hudi 表中的写入操作使用标记来有效地跟踪写入存储的数据文件。 在这篇博文中，我们深入探讨了现有的直接标记文件机制的设计，并解释了它在 AWS S3 等云存储上对于非常大的写入的性能问题。 我们展示了如何通过引入基于时间线服务器的标记来提高写入性能。

写操作期间需要标记

Hudi中的marker，比如文件名唯一的marker文件，是一个标签，表示存储中存在对应的数据文件，然后Hudi在故障和回滚场景中自动清理未提交的数据。 每个标记条目由三部分组成，数据文件名、标记扩展名 (.marker) 和创建文件的 I/O 操作（CREATE – inserts, MERGE – updates/deletes, or APPEND – either）。 例如，标记91245ce3-bb82-4f9f-969e-343364159174-0_140-579-0_20210820173605.parquet.marker.CREATE表示对应的文件数据为91245ce3-bb82-4f9f-969e-343364159174-0_140-579-0_20210820173605.parquet. 并且 I/O 类型是 CREATE。 Hudi 在文件系统中创建相应的数据文件之前创建一个标记，并在成功时删除与提交有关的所有标记。

标记对于有效地执行写客户端的不同操作很有用。 标记用作跟踪感兴趣的数据文件的一种方式，而不是通过列出表中的所有文件来扫描整个 Hudi 表。 两个重要的操作使用标记来方便有效地找到感兴趣的未提交数据文件：

• 删除重复/部分数据文件：在 Spark 中，Hudi 写入客户端将数据文件写入委托给多个执行程序。一个执行者可以使任务失败，留下部分数据文件被写入，在这种情况下，Spark 会重试该任务，直到它成功。当启用推测执行时，也可以多次成功尝试将相同的数据写入不同的文件，最终只有一次会交给 Spark 驱动程序进程进行提交。标记有助于有效地识别写入的部分数据文件，其中包含与后来成功试用的数据文件相比的重复数据，并且这些重复的数据文件在提交完成时被清除。如果没有这样的标记来跟踪每次提交的数据文件，我们必须列出文件系统中的所有文件，将其与时间线中看到的文件相关联，然后删除属于部分写入失败的文件。正如您可以想象的那样，在非常大的数据湖安装中，这将是非常昂贵的。
• 回滚失败的提交：写操作可能会在中间失败，留下一些数据文件写入存储中。 在这种情况下，标记条目会在提交失败时保留在存储中。 在下一次写入操作中，写入客户端在继续新的写入之前回滚失败的提交。 回滚是在标记的帮助下完成的，以识别作为失败提交的一部分写入的数据文件。

接下来，我们深入研究现有的标记机制，解释其性能问题，并演示新的基于时间线服务器的标记机制来解决问题。

现有的直接标记机制及其局限性

现有的标记机制只是简单地为每个数据文件创建一个新的标记文件，标记文件名如上所述。 标记文件没有任何内容，即为空。 每个标记文件都被写入同一目录层次结构中的存储，即commit instant和分区路径，位于 Hudi 表的基本路径下的临时文件夹 .hoodie/.temp 下。 例如，下图显示了向 Hudi 表写入数据时创建的标记文件和对应的数据文件的一个示例。 当获取或删除所有标记文件路径时，该机制首先列出临时文件夹.hoodie/.temp/下的所有路径，然后进行操作。

虽然扫描整个表以查找未提交的数据文件要高效得多，但随着要写入的数据文件数量的增加，要创建的标记文件的数量也会增加。 对于需要写入大量数据文件（例如 10K 或更多）的大型写入，这可能会为 AWS S3 等云存储造成性能瓶颈。 在 AWS S3 中，每个文件创建和删除调用都会触发一个 HTTP 请求，并且对存储桶中每个前缀每秒可以处理多少个请求有速率限制。 当要并发写入的数据文件数量和标记文件的数量很大时，标记文件操作可能会在写入操作期间占用不小的时间，有时大约为几分钟或更长时间。 在像 HDFS 这样的存储设备上，用户可能几乎不会注意到这一点，文件系统元数据被有效地缓存在内存中。

基于时间线服务器的标记机制提高写入性能

为了解决由于上述 AWS S3 的速率限制而导致的性能瓶颈，我们引入了一种利用时间线服务器的新标记机制，该机制通过非平凡的文件 I/O 延迟优化了与标记相关的存储延迟。 Hudi 中的时间线服务器用作提供文件系统和时间线视图的集中位置。 如下图所示，新的基于时间线服务器的标记机制将标记创建和其他标记相关操作从各个执行者委托给时间线服务器进行集中处理。 时间线服务器对标记创建请求进行批处理，并定期将标记写入文件系统中的一组有界文件。 这样，即使数据文件数量巨大，实际文件操作的数量和与标记相关的延迟也可以显着减少，从而提高写入的性能。

为了提高处理标记创建请求的效率，我们设计了在时间线服务器上批量处理标记请求。 每个标记创建请求都在 Javalin 时间线服务器中异步处理，并在处理前排队。 对于每个批处理间隔，例如 20 毫秒，时间线服务器从队列中拉出待处理的标记创建请求，并以循环方式将所有标记写入下一个文件。 在时间线服务器内部，这种批处理是多线程的，旨在保证一致性和正确性。 批处理间隔和批处理并发都可以通过写入选项进行配置。

请注意，工作线程始终通过将请求中的标记名称与时间线服务器上维护的所有标记的内存副本进行比较来检查标记是否已经创建。 存储标记的基础文件仅在第一个标记请求（延迟加载）时读取。 请求的响应只有在新标记刷新到文件后才会发回，因此在时间线服务器发生故障的情况下，时间线服务器可以恢复已经创建的标记。 这些确保了存储和内存副本之间的一致性，并提高了处理标记请求的性能。

标记相关的写入选项

我们在 0.9.0 版本中引入了以下与标记相关的新写入选项，以配置标记机制。 请注意，0.9.0 版本的 HDFS 尚不支持基于时间线服务器的标记机制，我们计划在未来支持 HDFS 的基于时间线服务器的标记机制。

性能

我们通过使用 Amazon EMR 与 Spark 和 S3 批量插入大型数据集来评估直接和基于时间线服务器的标记机制的写入性能。 输入数据约为 100GB。 我们通过将最大 parquet 文件大小设置为 1MB 并将并行度设置为 240 来配置写入操作以同时生成大量数据文件。请注意，在生产中不太可能将最大 parquet 文件大小设置为 1MB，这样的设置是 仅用于评估有关标记机制的性能。 正如我们之前提到的，虽然直接标记机制的延迟对于写入的数据文件数量较少的增量写入是可以接受的，但对于产生更多数据文件的大批量插入/写入，它会显着增加。

如下所示，直接标记机制非常有效，当写入表的一部分时，例如 165K 数据文件中的 1K。 但是，当我们需要写入大量数据文件时，直接标记操作的时间并不重要。 与直接标记机制相比，基于时间线服务器的标记机制由于批处理生成的文件存储标记要少得多，从而导致标记相关的 I/O 操作的时间大大减少，从而实现写入完成时间减少 31% 相对于直接标记文件机制。

结论

我们发现，对于需要写入大量数据文件的大型写入，现有的直接标记文件机制可能会由于 AWS S3 等云存储上的文件创建和删除调用的速率限制而导致性能瓶颈。 为了解决这个问题，我们引入了一种利用时间线服务器的新标记机制，它将标记创建和其他与标记相关的操作从单个执行器委托给时间线服务器，并使用批处理来提高性能。 使用 Spark 和 S3 对 Amazon EMR 进行的性能评估表明，与标记相关的 I/O 延迟和总体写入时间都减少了。