hudiclustering数据聚集(一)

database

概要

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

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

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

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

hudi clustering

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

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

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

实现

(用户可以将小文件的限制 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

以上是 hudiclustering数据聚集(一) 的全部内容, 来源链接: utcz.com/z/536068.html

回到顶部