kafka集群扩容后的数据迁移

最近我们生产环境的kafka集群有增加节点的需求，然而kafka在新增节点后并不会像elasticsearch那样感知到新节点加入后自动将数据reblance到新集群中，因此这个过程需要我们手动分配。一番折腾之后，实现了增加kafka集群节点并将原有数据均匀分配到扩容后的集群。下面结合一个例子谈一下整个过程。

假定当前的cluster中只有（101，102，103）三个kafka节点，有一个名为think_tank的topic，该topic有2个replica，均匀分布在三个节点上.

我们要做的是在cluster中新增两个节点（记为104，105）后，将的数据均匀分到新集群中的5个节点上。

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

结合例子具体说明：

脚本的参数是以json文件的形式传入的，首先要新建一个json文件并设置需要分配哪些topic，think_tank-to-move.json:

使用/bin目录中提供的 kafka-reassign-partitions.sh 的脚本请求获取生成分配方案：

--broker-lsit 的参数 "101,102,103,104,105"是指集群中每个broker的id，由于我们是需要将所有topic均匀分配到扩完结点的5台机器上，所以要指定。同理，当业务改变为将原来的所有数据从旧节点（01,102,103）迁移到新节点（104，105）实现数据平滑迁移，这时的参数应"104，105".

脚本执行后返回的结果如下：

可以看出当前正在运行的方案中，think_tank的replica都是分布在101，102，103这3个节点，新给出的建议方案中replica均匀分布在扩容后的5个节点中。

将上一个步骤中生成的建议方案复制到新建的think_tank_reassignment.json中：

使用脚本执行：

脚本执行，返回内容：

如上，成功开始执行分配数据，同时提示你如果有需要将之前的分配方案备份便于回滚到原方案。

查看脚本的方法如下，注意这次的json文件要和执行步骤中的json是同一个文件：

返回结果：

is still in progress表示还在处理中，全部迁移成功后每个partition都会显示 completed successfully.注意如果topic数据量大，这个过程可能会时间长一些， 不要轻易重启节点！ 可能会导致数据不一致！！！

这个partion reassignment工具同样可以按需手动地将某个特定的topic指定到特定的broker上，所要做的就是按照步骤一给定的格式关联partition到borker即可，如，将think_tank的partition0指定到101、102两节点上：

另外，如果有增加replica的个数的需求，同样可以使用这个脚本，可以翻一下官网文档。

一点儿感触，在确定问题所在后，官方的文档应该作为我们优先考虑的一个重要资料源，网上的资料由于时间较早、版本不同的原因，解决方式可能需要细微的改动才能达到目的，这些坑在官方的一手资料上其实是可以规避的。

欢迎拍砖，欢迎交流~

在一个kafka的cluster集群中，如果一个节点的数据损坏了，例如本地磁盘坏了，导致单个kafka无法使用，或者无法启动了；首先我们想到的是尽快修复和恢复本地数据，然后重新加入cluster。

但是如果存在无法修复的情况，那么怎么办呢，基于是一个集群环境，kafka能够自动从集群中其他节点恢复数据。

在${log.dirs}目录下面通常有这些文件：

---