kafka集群扩容后的数据均衡

Myth丶恋晨 2023-01-19 13:00 20阅读 0赞

[https://www.hyperxu.com/2019/12/01/kafka-1/][https_www.hyperxu.com_2019_12_01_kafka-1]

生产环境的`kafka`集群扩容，是一个比较常见的需求和操作。然而`kafka`在新增节点后并不会像`elasticsearch`那样感知到新节点加入后，自动将数据`reblance`到整个新集群中，因此这个过程需要我们手动分配。

## 分区重分配方案 ##

扩容后的数据均衡，其本质就是对`topic`进行分区重分配，数据迁移的过程。  
在执行分区重分配的过程中，对集群的影响主要有两点：

1.  分区重分配主要是对`topic`数据进行`Broker`间的迁移，因此会占用集群的带宽资源；
2.  分区重分配会改变分区`Leader`所在的`Broker`，因此会影响客户端。

针对以上两点，第一点可以在晚间业务低峰时操作，必要时还可以和业务沟通，临时缩短数据保存时间，加快迁移，减少带宽影响时间。  
针对第二点，有两个方案：

1.  整个分配方案分成两个步骤：  
    1）手动生成分配方案，原有分区`Leader`位置不改变，只对副本进行分区重分配；  
    2）等待数据迁移完成后，再手动更改分区分配方案，目的是均衡`Leader`。
2.  直接用`Kafka`官方提供的分区重新分配工具生成分区重分配方案，直接执行分区重分配。

## 重分配方案分析 ##

### 方案一 ###

方案一理论对客户端影响最小，把整个分配方案分成了两个步骤，也就是将对集群的带宽资源与客户端的影响分开了，对过程可控性很高。但问题是，如果集群中的某些`topic`，比如有64个分区，3副本，共192个副本，就需要在保持原有分区`Leader`位置不变的情况下，手动均衡其余副本，这个人工步骤过度繁杂，稍微有一点偏差，就会造成副本不均衡。

### 方案二 ###

针对方案二我特意去看了分区重分配的源码，并对其过程进一步分析。发现分区重分配的步骤是，将分区原有的副本与新分配的副本合并成一个新的副本集合，新分配的副本努力追上`Leader`的`offset`，最终加入`ISR`。待全部副本都加入`ISR`之后，就会进行分区`Leader`选举，选举完后删除原有副本。这里注意，由于是最后选举完成才删除原副本，所以重分配的过程中，日志存储量是会大幅增加的。具体细节我后续单独写一篇文章叙述。

根据以上分析，意味着在数据进行重分配过程中，`Leader`并没有发生变动，所以客户端不会阻塞，数据迁移完成后进行`Leader`选举时发生变更，生产者会及时拉取最新的元数据，并重新进行消息发送，影响并不大。

## 重分配步骤 ##

其实官方文档关于集群扩容讲解很详细：Expanding your cluster ,整个过程分为三个步骤：获取kafka给出的建议分配方案、按照给出的分配方案执行分配、查看分配的进度以及状态。这三个步骤对应了kafka脚本提供的三个`partition reassigment`工具。

<table> 
 <tbody> 
 <tr> 
 <td> <pre>
&nbsp;</pre> 1 2 3 </td> 
 <td> <pre>
&nbsp;</pre> --generate: 在此模式下，给定一个 topic 列表和一个 broker 列表，该工具会生成一个候选重新分配，以将指定的 topic 的所有分区移动到新的broker。此选项仅提供了一种便捷的方式，可以根据 tpoc 和目标 broker 列表生成分区重新分配计划。 --execute: 在此模式下，该工具基于用户提供的重新分配计划启动分区重新分配。（使用--reassignment-json-file选项）。这可以是由管理员制作的自定义重新分配计划，也可以是使用--generate选项提供的自定义重新分配计划。 --verify: 在此模式下，该工具将验证最近用 --execute 模式执行间的所有分区的重新分配状态。状态可以是成功完成，失败或正在进行。 </td> 
 </tr> 
 </tbody> 
</table>

1. 生成需要执行分区重分配的`topic`列表json文件：
 
 <table> 
 <tbody> 
 <tr> 
 <td> <pre>
 &nbsp;</pre> 1 2 3 4 5 </td> 
 <td> <pre>
 &nbsp;</pre> &gt; cat topics-to-move.json {"topics": [{"topic": "foo1"}, {"topic": "foo2"}], "version":1 } </td> 
 </tr> 
 </tbody> 
 </table>
2. 使用`kafka-reassign-partitions.sh`脚本获取分配方案：
 
 <table> 
 <tbody> 
 <tr> 
 <td> <pre>
 &nbsp;</pre> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 </td> 
 <td> <pre>
 &nbsp;</pre> &gt; bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topics-to-move.json --broker-list "5,6" --generate 当前分区副本分配 {"version":1, "partitions":[{"topic":"foo1","partition":2,"replicas":[1,2]}, {"topic":"foo1","partition":0,"replicas":[3,4]}, {"topic":"foo2","partition":2,"replicas":[1,2]}, {"topic":"foo2","partition":0,"replicas":[3,4]}, {"topic":"foo1","partition":1,"replicas":[2,3]}, {"topic":"foo2","partition":1,"replicas":[2,3]}] } 建议的分区重新分配配置 {"version":1, "partitions":[{"topic":"foo1","partition":2,"replicas":[5,6]}, {"topic":"foo1","partition":0,"replicas":[5,6]}, {"topic":"foo2","partition":2,"replicas":[5,6]}, {"topic":"foo2","partition":0,"replicas":[5,6]}, {"topic":"foo1","partition":1,"replicas":[5,6]}, {"topic":"foo2","partition":1,"replicas":[5,6]}] } </td> 
 </tr> 
 </tbody> 
 </table>
 
 保存当前分区副本分配情况，用作回滚操作用。保存建议的分区重新分配配置到`expand-cluster-reassignment.json`用于执行迁移。
3. 执行重分配，并验证。
 
 <table> 
 <tbody> 
 <tr> 
 <td> <pre>
 &nbsp;</pre> 1 2 3 4 5 6 7 8 9 10 </td> 
 <td> <pre>
 &nbsp;</pre> &gt; bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --execute &gt; bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --verify Status of partition reassignment: Reassignment of partition [foo1,0] completed successfully Reassignment of partition [foo1,1] is in progress Reassignment of partition [foo1,2] is in progress Reassignment of partition [foo2,0] completed successfully Reassignment of partition [foo2,1] completed successfully Reassignment of partition [foo2,2] completed successfully </td> 
 </tr> 
 </tbody> 
 </table>
 
 `is still in progress`表示还在处理中，全部迁移成功后每个`partition`都会显示`completed successfully`。注意如果`topic`数据量大，这个过程可能会很长，在此期间不要进行敏感操作，可能会导致数据不一致。

## 自定义重分配 ##

分区重新分配工具还可以将分区的副本移动到指定的一组`broker`。只需自定义修改分配配置文件，后续步骤同上。

例如，以下示例将`topic foo1`的分区0移到`broker`5,6中和将`topic foo2`的分区1移到`broker`2,3中：

json

<table> 
 <tbody> 
 <tr> 
 <td> <pre>
&nbsp;</pre> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 </td> 
 <td> <pre>
&nbsp;</pre> &gt; cat custom-reassignment.json { "version": 1, "partitions": [ { "topic": "foo1", "partition": 0, "replicas": [ 5, 6 ] }, { "topic": "foo2", "partition": 1, "replicas": [ 2, 3 ] } ] } </td> 
 </tr> 
 </tbody> 
</table>

[https_www.hyperxu.com_2019_12_01_kafka-1]: https://www.hyperxu.com/2019/12/01/kafka-1/