介绍

Apache Hadoop Ⓡ分布式文件系统 (HDFS) 是一种分布式文件系统，旨在以可靠且容错的方式跨多台机器存储大型文件。它是 Apache Hadoop 框架的一部分，也是 Uber 数据堆栈的主要组件之一。

Uber 拥有世界上最大的 HDFS 部署之一，数十个集群中有 EB 级的数据。在效率、服务可靠性和高性能之间取得平衡的同时，不断扩展我们的数据基础设施非常重要，但也很有挑战性。

图 1：Uber 的 HDFS 基础设施。

概述

HDFS 平衡器是通过在集群中均匀地重新分配数据来保持 DataNode 正常运行的关键组件。随着我们的 HDFS 集群中节点退役越来越频繁，HDFS 平衡器必须更有效地平衡数据以防止 DataNode 倾斜。节点退役要求来自区域退役、安全补丁自动集群周转以及 DataNode 共置等项目。

然而，开源 HDFS 附带的平衡器并不能立即满足这一要求。我们已经看到一个 DataNode 出现倾斜的问题（即与同一集群中的其他节点相比存储了更多数据），这会带来多种副作用：

导致主机 I/O 带宽过高，数据过多利用率高的节点运行缓慢的概率更高，节点故障和数据丢失的风险也更高集群具有较少的活跃和健康节点来为客户提供写入流量

下面是一个不平衡数据的示例：在我们最大的集群中，数千个节点的磁盘利用率接近 95%，该集群由数千个 DataNode 组成，容量达数百 PB，而平衡吞吐量无法有效地将数据移动到其他新添加的 DataNode。这种不平衡的数据分布是由热分层和 EC 转换[1] 产生的突发写入流量、区域退役/集群因安全补丁而交接导致的大量节点退役造成的。由于写入可靠性是第一要务，所有 DataNode 都使用可用容量加权算法来处理写入流量。写入流量越多，数据倾斜度就越大。

图 2：我们最大的集群之一包含大约数千个 DataNode，容量为数百 PB，DataNode 已经出现偏差。

因此，我们需要优化 HDFS 平衡器，将数据平衡从高使用率的 DataNode 增加到另一个占用率较低的 DataNode。

考虑到 Uber 的数据存储规模，单个集群中数据不平衡的节点将超过 20 PB，集群数量为 7-8 个。为了解决 Uber DataLake 中 HDFS DataNode 平衡问题，我们设计了一种新算法来增加 DataNode 之间形成的对的数量，这将在平衡数据的同时增加并行块移动。此外，我们根据利用率对 DataNode 进行排序，以便优化形成的数据节点对，并且不会发生递归平衡。

该算法将继续增加我们的平衡吞吐量，即每秒从占用率较高的数据节点移动到占用率较低的数据节点以进行平衡的数据大小。

建筑与设计

图 3：HDFS 平衡器架构。

初始化和设置：HDFS 平衡器作为 Hadoop 集群内的服务在主机上运行。要启动平衡过程，集群中需要有一个具有平衡器角色的节点。两个平衡器不能同时运行。请求集群信息：平衡器首先联系 NameNode，请求有关集群内数据分布的信息。它向 NameNode 发送请求，以获取有关数据块在 DataNode 之间的分布的详细信息。NameNode 会响应 DataNode 及其包含的块的列表，以及它们的存储容量和其他相关信息。地块选择与规划：根据从 NameNode 收到的信息，平衡器算法选择需要移动的块以实现更均衡的分布。平衡器在规划块移动时会考虑 DataNode 利用率、机架信息、线程和存储容量等因素。数据移动的协调：确定要移动哪些块后，平衡器会协调 DataNode 之间的实际数据移动。它通过心跳与 NameNode 通信有关移动的块。区块迁移：平衡器通过直接与源和目标 DataNode 通信来启动块迁移。它指示源DataNode将选定的块传输到目标DataNode，直接移动数据块。监测进度：在整个数据移动过程中，平衡器会持续监控进度。它会跟踪已成功传输的块数，并确保数据移动按照计划进行。完成和报告：平衡操作完成后，平衡器会在日志和指标中报告已传输的数据和剩余待传输的数据。它还可以提供有关平衡过程的统计数据和指标，包括移动的块数和所花费的时间。终止：在主机中，平衡器作为服务运行。因此，在集群达到平衡之前，它不会停止移动数据。初步优化

由于我们的目标是提高吞吐量，以更快的速度平衡 DataNode，因此我们优化了 HDFS 平衡器，使其具有现有的 DataNode 属性，以提高吞吐量。尽管我们将平衡器的速度提高了 3 倍，但吞吐量仍然不够。我们有太多占用率高的节点，而现有算法中数据传输到的 DataNode 对的数量将大大减少。此外，我们无法通过平衡器线程提高每个节点的吞吐量，因为增加吞吐量会增加节点的速度并影响读/写流量。因此，我们需要增加 DataNode 对的数量，这最终会导致平衡吞吐量的增加。我们使用的 DataNode 和 Balancer 配置如下所述。根据您的情况，工作负载的配置可能会有所不同。

DataNode配置属性：

财产

默认

快速模式

dfs.DataNode.balance.max.concurrent.moves

5

250

dfs.DataNode.balance.bandwidthPerSec

1048576（1MB）

1073741824（1GB）

平衡器配置属性：

财产

默认

快速模式

dfs.DataNode.balance.max.concurrent.moves

5

250

dfs.balancer.moverThreads

1000

2000

dfs.balancer.最大移动大小

10737418240（10GB）

107374182400（100GB）

dfs.balancer.getBlocks.最小块大小

10485760（10MB）

104857600（100MB）

算法优化增加 DataNode 对以实现高吞吐量

更多的 DataNode 对意味着我们可以进行更多的并发块传输，因此一个关键的改进是构建更多的对。由于现有算法，高度倾斜的集群会形成较少的 DataNode 对。

图 4：现有算法。

在现有的 HDFS Balancer 算法中，集群平均利用率以上的 DataNode（即利用率高于平均水平和利用率过高的节点）的数量比利用率低于平均水平和利用率不足的节点的数量要高得多。因此，我们面临着节点稀缺的问题，无法将数据从利用率高的 DataNode 中移出，导致利用率高的 DataNode 无法快速停机。

在上图中，有 8 个 DataNode 高于平均利用率，4 个 DataNode 低于平均利用率，因此有 4 个目标可以移动数据。目的是修改 HDFS 算法，以便为 DataNode 形成更多配对，从而从高使用率 DataNode 获得更高的吞吐量，从而实现均匀利用率，并随着 DataNode 覆盖范围的扩大而快速降低使用率。

我们的想法是使用基于百分位数的算法来创建更多的 DataNode 对。

图 5：新算法。

在新算法中，我们根据百分位数创建了调整后的平均值，这将增加数据可以移动到的节点数。高于平均水平/利用率过高的数据节点将尝试接近整体集群利用率，而利用率不足/利用率低于平均水平的节点将尝试接近调整后的百分位数平均值。使用基于百分位数的算法，我们的目标是使调整后的平均值接近整体集群利用率。

我们将使用基于百分位数的算法来增加 DataNode 对。在高度倾斜的集群中，百分位数相当高。以上图为例，我们将百分位数设为 P60，我们的调整平均值现在为 86.7%。在这种情况下，过度利用/利用率高于平均水平的节点数量减少，利用率不足/利用率低于平均水平的节点数量增加。

现在，将有 5 个过度利用和高于平均利用率的节点，以及 7 个利用率不足和低于平均水平的节点，这将导致从 4 对形成最多 7 对。

我们有一个新的 Hadoop 配置属性，dfs.balancer.separate-percentile，

图 6：用于定义百分位数的新 Hadoop 配置。

默认值为 0.5，表示第 50 个百分点。如果我们使用 -dynamicBalancer 部署平衡器命令，则此百分位数算法将生效，调整后的平均值将以更高的吞吐量进入画面。

我们还可以使用此阈值进行动态平衡。例如，如果 DataNodes 的使用率超过 90%，我们将积极地平衡它们（即提高速度）。因此，我们将平衡前 20% 的 DataNodes，这将导致将 moverThreads 集中在前 20% 的高利用率源上，并且数据将从高利用率的 DataNodes 更快地移动，从而更快地降低使用率。

图 7：用于定义积极平衡的新 Hadoop 配置。

将数据移动到占用率较低的 DataNode

由于自动化（即自动将数据从 DataNode 移至其他 DataNode 以进行维护），大型集群中的 DataNode 经常发生退役，其中退役节点的数据被移动到其他节点，从而增加了这些节点上的占用百分比。新出现的节点由于没有优先权而慢慢地达到平衡。另外，例如，如果平均利用率为 83%，阈值为 3%，则 90% 的 DataNode 将其部分数据移动到 79% 的节点，该节点变为 81%。现在，如果新客户端在 81% 时转储数据，则变为 87%，这可能需要进一步平衡此节点，从而分配调度程序和移动程序线程。

图 8：旧算法——形成对。

图 9：旧算法——出现了新的过度利用节点。

图 10：新算法——首选优化。

我们的增强功能是通过按升序对利用率不足的节点或低于平均水平的节点中的节点进行排序，优先考虑占用率较小的 DataNode，首先平衡来自过度利用的节点的数据，然后按降序排列来自利用率高于平均水平的节点的数据，以便在平衡时防止递归平衡。

更好的可观察性

我们没有关于同一节点组、同一机架和任何其他机架之间过度利用和利用不足、过度利用和利用不足和利用不足之间形成的 DataNode 对的指标，也没有其他相关指标。因此，我们无法校准这些对之间的流量分布。为了找出可以增加 DataNode 对以增加吞吐量的地方，我们创建了一个新的仪表板。

最后，我们添加了 10 多个指标来跟踪算法变化的性能，这将帮助我们更好地校准平衡器的自定义算法。

图 11：我们的指标仪表板的快照。

结果

通过优化平衡算法，我们将吞吐量提高了 5 倍以上，没有利用率高于 90% 的数据节点，并且总体上降低了数据节点的使用率。此外，现在无需部署手动平衡器（仅使用某些硬编码节点来平衡数据），因为我们对算法的优化已经解决了这个问题。

作为我们新算法的一部分 –

提高吞吐量 – 我们将吞吐量提高了 5 倍以上。关闭高度使用的数据节点 – 我们将利用率超过 90% 的数据节点降至 0。数据节点利用率大致相同 – 降低数据节点的总体使用率，使其容量大致相同。我们最大的集群中的所有数据节点的利用率都低于 85%。更好地管理容量 – 我们的 HDFS 集群利用率从 65-66% 增加到 85% 左右，我们遇到了容量瓶颈。虽然集群利用率比以往更高，但现在我们不再有高度占用的数据节点。

图 12：由于算法的改变，DataNode 处于相似的水平，并且我们最大的集群的利用率低于 85％。

图 13：反映 DataNode 倾斜减少的面板。

图 14：平衡器算法改变之前 - 使用率超过 90% 的数据节点为 50.8%。

图 15：平衡器算法改变后，使用率超过 90% 的数据节点低于 0。

图 16：我们的一个集群，集群利用率较低，约为 65%。

图 17：对于上述同一集群，集群利用率增加到 83% 左右。

图 18：由于算法改变，吞吐量增加了 3 倍以上。

结论

在 HDFS 集群中，数据可能会在不同 DataNode 之间出现偏差，并可能导致节点上的 I/O 过高，从而导致节点速度变慢或宕机，进而造成数据丢失。新算法将有助于更快地平衡 DataNode，以实现更高的效率、服务可靠性和高性能，同时防止出现更高的速度缓慢概率、更高的节点故障风险和数据丢失。

在 Uber，我们将这一变化部署到多个集群，以提高平衡吞吐量。我们正在为我们的优化提出一个开源补丁。Uber HDFS 团队继续致力于解决类似的数据分布问题——考虑到我们的规模，即使是微小的改进也能带来巨大的收益。

[1] Uber 将具有不同访问温度的数据保存到专用集群中，以提高可靠性和成本效益。我们应用暖分层将数据从热集群移动到温集群，并采用 EC 转换将数据移动到具有擦除编码功能的集群，从而节省了 50% 的容量。

“Apache®、Apache Hadoop® 和 Hadoop® 是 Apache 软件基金会在美国和/或其他国家/地区的注册商标或商标。使用这些商标并不代表 Apache 软件基金会对其的认可。”

作者:

Atul Kaushik

Atul Kaushik is a Software Engineer II with the Data storage team at Uber. He has been working on optimizations related to DataNode balancing and developing HDFS Quota solutions for customers at Uber.

Yangjun Zhang

Yangjun is a Staff Software Engineer with the Data storage team at Uber. He has been working on the reliability, efficiency, and modernization improvement for the HDFS dataplane.

出处:https://www.uber.com/en-JP/blog/balancing-hdfs-datanodes-in-the-uber-datalake/