大家好，我是考100分的小小码 ，祝大家学习进步，加薪顺利呀。今天说一说hudi clustering 数据聚集（一）「建议收藏」,希望您对编程的造诣更进一步.
概要
数据湖的业务场景主要包括对数据库、日志、文件的分析，而管理数据湖有两点比较重要：写入的吞吐量和查询性能，这里主要说明以下问题：
1、为了获得更好的写入吞吐量，通常把数据直接写入文件中，这种情况下会产生很多小的数据文件。虽然小文件的使用可以增加写入的并行度，且能够并行读取文件以提高读取速度，但会出现一个数据量很小，需要从多个小文件中读取数据，增加了很多IO。
2、数据按照进入数据湖的方式写入到文件中，在同一个文件上，数据局部性不是最佳的。 数据之间，与传入批次相关，相近的批次的数据会相关联，而不是与经常要查询的数据相关联。所以小文件的大小和缺乏数据局部性会降低查询性能。
3、此外，许多文件系统（包括 hdfs），当有很多小文件时，性能会下降。
hudi clustering
hudi支持clustering功能，在不影响查询性能的情况下提高写入吞吐量。该功能可以以不同方式重写数据：
1、数据先写入小文件，在满足某些条件后（例如经过的时间、小文件数量、commit次数等），将小文件拼接成大文件。
2、通过对不同列上的数据进行排序，来更改磁盘上的数据布局，已提高数据间的相关性，可以提高查询性能。
实现
（用户可以将小文件的限制 hoodie.parquet.small.file.limit 配置为 0，这样可以强制将数据进入新的文件组。）
cow表的timeline
在上面的示例流程图中，显示了随时间（t5 到 t9）的分区状态。 主要有以下步骤：
- 在 t5，表中的一个分区有 5 个文件组 f0、f1、f2、f3、f4，分别在 t0、t1、t2、t3、t4时刻被创建。 假设每个文件组为 100MB。 所以分区中的总数据为 500MB。
- 在 t6 请求 clustering 操作。 与压缩类似，我们在带有“ClusteringPlan”的元数据中创建了一个“t6.clustering.requested”文件，其中包含跨所有分区的集群操作涉及的所有文件组。例如：{ partitionPath: {“datestr”}, oldfileGroups: [ {fileId: “f0”, time: “t0”}, { fileId: “f1”, time: “t1”}, … ], newFileGroups: [“c1”, “c2”] }
- 假设clustering后的最大文件大小配置为 250MB。 集群会将分区中的所有数据重新分配到两个文件组中：c1、c2。 此时这些文件组是“虚假”的，在 t8 clustering 完成之前，对查询不可见。
- 请注意，文件组中的记录可以拆分为多个文件组。 在此示例中，来自 f4 文件组的一些记录同时转到了新文件组 c1、c2。
- 当集群正在进行时（t6 到 t8），任何涉及到这些文件组的更新插入都会被拒绝。
- 在写入新的数据文件 c1-t6.parquet 和 c2-t6.parquet 后，如果配置了全局索引，我们会在记录级索引中为所有具有新位置的键添加条目。 新的索引条目对其他写入将不可见，因为还没有关联的提交。
- 最后，我们创建一个提交元数据文件“t6.commit”，其中包含由此次提交修改的文件组（f0、f1、f2、f3、f4）。
- 注：文件组（f0 到 f4）不会立即从磁盘中删除。 cleaner 会在归档 t6.commit 之前清理这些文件。 并且，clustering 还会更新所有视图和源数据文件。
mor表的时间线
这种方法同样支持mor表，且过程与cow 表非常相似。
clustering 的为 parquet 格式文件。
Clustering 操作步骤
总体来说，需要两步：
- clustering 调度：创建 clustering 计划
- 执行 clustering：执行计划。创建新的文件，并替换旧的文件。
clustering 调度
- 识别符合集群条件的文件
- 过滤特定分区（根据配置优先考虑最新分区或旧分区）
- 任何大小 > targetFileSize 的文件都不符合条件
- 任何有待定压缩/clustering计划的文件都不符合条件
- 任何具有日志文件的文件组都不符合集群条件（该限制以后可能会被取消）
- 根据特定条件对符合聚类条件的文件进行分组。 每个组的数据大小预计是“targetFileSize”的倍数。 分组是作为计划中定义的“策略”的一部分完成的：
- 根据记录键范围对文件进行分组。因为键值范围存储在parquet footer中，这个可用于某些查询/更新。
- 根据提交时间对文件进行分组。
- 对自定义列，且具有重叠值的文件进行分组（指定列进行排序）
- 分组随机文件
- 我们可以限制组大小以提高并行性
- 根据特定条件过滤组（类似于 CompactionStrategy 中的 orderAndFilter）
- 最后，clustering计划被保存到timeline中。
执行 clustering
- 读取clustering计划，查看“clusteringGroups”的数量（用于并行性）。
- 创建 inflight状态的 clustering 文件
- 对于每组：
- 使用 strategyParams 实例化适当的策略类（例如：sortColumns）
- 策略类定义了分区器，我们可以用它来创建桶并写入数据。
- 创建 replacecommit：
- operationType 设置为“clustering”。
- 扩展元数据，并存储附加字段以跟踪重要信息（策略类可以返回这些额外的元数据信息）
- 用于合并文件的策略
- 跟踪替换文件
【参考】
https://hudi.apache.org/docs/next/configurations/#hoodieclusteringplanstrategyclass
https://cwiki.apache.org/confluence/display/HUDI/RFC+-+19+Clustering+data+for+freshness+and+query+performance
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
转载请注明出处: https://daima100.com/5725.html
