概要

数据湖的业务场景主要包括对数据库、日志、文件的分析，而管理数据湖有两点比较重要：写入的吞吐量和查询性能，这里主要说明以下问题：

    1、为了获得更好的写入吞吐量，通常把数据直接写入文件中，这种情况下会产生很多小的数据文件。虽然小文件的使用可以增加写入的并行度，且能够并行读取文件以提高读取速度，但会出现一个数据量很小，需要从多个小文件中读取数据，增加了很多IO。 

    2、数据按照进入数据湖的方式写入到文件中，在同一个文件上，数据局部性不是最佳的。 数据之间，与传入批次相关，相近的批次的数据会相关联，而不是与经常要查询的数据相关联。所以小文件的大小和缺乏数据局部性会降低查询性能。

   3、此外，许多文件系统（包括 hdfs），当有很多小文件时，性能会下降。

hudi clustering

hudi支持clustering功能，在不影响查询性能的情况下提高写入吞吐量。该功能可以以不同方式重写数据：

数据先写入小文件，在满足某些条件后（例如经过的时间、小文件数量、commit次数等），将小文件拼接成大文件。

通过对不同列上的数据进行排序，来更改磁盘上的数据布局，已提高数据间的相关性，可以提高查询性能。

实现

（用户可以将小文件的限制 hoodie.parquet.small.file.limit 配置为 0，这样可以强制将数据进入新的文件组。）

cow表的timeline

在上面的示例流程图中，显示了随时间（t5 到 t9）的分区状态。 主要有以下步骤：

1. 在 t5，表中的一个分区有 5 个文件组 f0、f1、f2、f3、f4，分别在 t0、t1、t2、t3、t4时刻被创建。 假设每个文件组为 100MB。 所以分区中的总数据为 500MB。
2. 在 t6 请求 clustering 操作。 与压缩类似，我们在带有“ClusteringPlan”的元数据中创建了一个“t6.clustering.requested”文件，其中包含跨所有分区的集群操作涉及的所有文件组。例如：{ partitionPath: {“datestr”}, oldfileGroups: [ {fileId: “f0”, time: “t0”}, { fileId: “f1”, time: “t1”}, ... ], newFileGroups: [“c1”, “c2”] }
3. 假设clustering后的最大文件大小配置为 250MB。 集群会将分区中的所有数据重新分配到两个文件组中：c1、c2。 此时这些文件组是“虚假”的，在 t8 clustering 完成之前，对查询不可见。
4. 请注意，文件组中的记录可以拆分为多个文件组。 在此示例中，来自 f4 文件组的一些记录同时转到了新文件组 c1、c2。
5. 当集群正在进行时（t6 到 t8），任何涉及到这些文件组的更新插入都会被拒绝。
6. 在写入新的数据文件 c1-t6.parquet 和 c2-t6.parquet 后，如果配置了全局索引，我们会在记录级索引中为所有具有新位置的键添加条目。 新的索引条目对其他写入将不可见，因为还没有关联的提交。
7. 最后，我们创建一个提交元数据文件“t6.commit”，其中包含由此次提交修改的文件组（f0、f1、f2、f3、f4）。
8. 注：文件组（f0 到 f4）不会立即从磁盘中删除。 cleaner 会在归档 t6.commit 之前清理这些文件。 并且，clustering 还会更新所有视图和源数据文件。

mor表的时间线

这种方法同样支持mor表，且过程与cow 表非常相似。

clustering 的为 parquet 格式文件。

Clustering 操作步骤

总体来说，需要两步：

1. clustering 调度：创建 clustering 计划
2. 执行 clustering：执行计划。创建新的文件，并替换旧的文件。

clustering 调度

1. 识别符合集群条件的文件
   1. 过滤特定分区（根据配置优先考虑最新分区或旧分区）
   2. 任何大小 > targetFileSize 的文件都不符合条件
   3. 任何有待定压缩/clustering计划的文件都不符合条件
   4. 任何具有日志文件的文件组都不符合集群条件（该限制以后可能会被取消）
2. 根据特定条件对符合聚类条件的文件进行分组。 每个组的数据大小预计是“targetFileSize”的倍数。 分组是作为计划中定义的“策略”的一部分完成的：
   1. 根据记录键范围对文件进行分组。因为键值范围存储在parquet footer中，这个可用于某些查询/更新。
   2. 根据提交时间对文件进行分组。
   3. 对自定义列，且具有重叠值的文件进行分组（指定列进行排序）
   4. 分组随机文件
   5. 我们可以限制组大小以提高并行性
3. 根据特定条件过滤组（类似于 CompactionStrategy 中的 orderAndFilter）
4. 最后，clustering计划被保存到timeline中。

执行 clustering

1. 读取clustering计划，查看“clusteringGroups”的数量（用于并行性）。
2. 创建 inflight状态的 clustering 文件
3. 对于每组：
   1. 使用 strategyParams 实例化适当的策略类（例如：sortColumns）
   2. 策略类定义了分区器，我们可以用它来创建桶并写入数据。
4. 创建 replacecommit：
   1. operationType 设置为“clustering”。
   2. 扩展元数据，并存储附加字段以跟踪重要信息（策略类可以返回这些额外的元数据信息）
      1. 用于合并文件的策略
      2. 跟踪替换文件

【参考】

https://hudi.apache.org/docs/next/configurations/#hoodieclusteringplanstrategyclass

https://cwiki.apache.org/confluence/display/HUDI/RFC+-+19+Clustering+data+for+freshness+and+query+performance