1.概述

Kafka一些实时流数据业务场景被广泛使用。Kafka做数据分流。数据迁移是分布式应用场景中常见的问题。Kafka今天叶秋学长将详细介绍如何迁移集群数据。

2.内容

本博客介绍了两种迁移场景，即同一集群数据迁移和跨集群数据迁移。如下图所示：

2.1 同集群迁移

例如，在现有集群中增加了同一集群之间的数据迁移Broker节点需要在原集群中存在Topic将数据迁移到新的集群中，以缓解集群压力。

添加新节点Kafka集群很简单，只需为它们分配一个唯一的Broker ID，并在新服务器上启动Kafka。然而，这些新的服务器节点不会自动分配任何数据分区，因此除非将分区移动到新的节点，否则会创建新的节点Topic新节点以前不会执行任何操作。因此，通常会添加新的服务器节点Kafka在集群中，有必要将一些现有数据迁移到这些新节点。

迁移数据的过程是手动启动的，执行过程是完全自动化的。Kafka在后台服务中，Kafka将添加新服务器作为正在迁移的分区Follower，并允许新节点完全复制分区中的现有数据。当新服务器节点完全复制分区内容并添加同步副本时（ISR）其中一个现有副本将删除其分区的数据。

Kafka系统提供了分区重新分配的工具（kafka-reassign-partitions.sh），该工具可用于在场Broker分区之间迁移。理想情况下，一切都将得到保证Broker均匀分配数据和分区。分区重新分配工具不能自动分析Kafka集中的数据分布和迁移分区，实现均匀的负载平衡。因此，管理员在操作时，必须弄清楚应该迁移什么Topic或分区。

分区重新分配工具可在三种互斥模式下运行：

• --generate：在此模式下，给定Topic列表和Broker列表将生成候选人重新分配指定Topic所有分区迁移到新的Broker在给定的选项中，这个选项只提供了一种方便的方法Topic和目标Broker在列表中生成分区重新分配计划。
• --execute：在此模式下，该工具将根据用户提供的重新分配计划启动分区的重新分配。 （使用--reassignment-json-file选项)。自定义重新分配计划也可以由管理员手动制定--generate选项提供。
• --verify：在这种模式下，工具将验证最后一次--execute期间列出的所有分区的重新分配状态。状态可以是成功、失败或正在进行的状态。

2.1.1 实现迁移过程

可用于分区重新分配工具Topic从当前的Broker将节点迁移到新添加的节点中Broker中。这通常对扩大现有集群非常有用，因为它将使整个集群Topic移动到新的Broker变得更容易，而不是一次移动分区。当执行此操作时，用户需要提供现有操作Broker节点的Topic列表，到新节点Broker列表（源Broker到新Broker映射关系)。然后，工具是新的Broker均匀分配给指定Topic列表的所有分区。迁移过程中，Topic复制因子保持不变。

现有以下例子，将Topic为ke01，ke从02的所有分区Broker1中移动到新增的Broker2和Broker3.因为工具被接受。Topic作为输入列表JSON因此，文件需要明确迁移Topic并创建json文件如下：

> cat topic-to-move.json {"topics": [{"topic": "ke01"},             {"topic": "ke02"}], "version":1 }

准备好JSON然后使用分区重新分配工具生成候选人分配，命令如下：

> bin/kafka-reassign-partitions.sh --zookeeper dn1:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate

执行命名前，Topic（ke01、ke02）的分区如下图所示：

执行命令后，控制台出现以下信息：

该工具生成候选分配，将所有分区从Topic ke01，ke02移动到Broker1和Broker2。需要注意的是，分区移动还没有开始，它只是告诉你当前的分配和建议。保存当前分配，以防你想回滚。应保存新的赋值JSON文件（例如expand-cluster-reassignment.json）中，以使用--execute选项执行。JSON文件如下：

{"version":1,"partitions":[{"topic":"ke02","partition":0,"replicas":[2]},{"topic":"ke02","partition":1,"replicas":[1]},{"topic":"ke02","partition":2,"replicas":[2]},{"topic":"ke01","partition":0,"replicas":[2]},{"topic":"ke01","partition":1,"replicas":[1]},{"topic":"ke01","partition":2,"replicas":[2]}]}

执行命令如下：

> ./kafka-reassign-partitions.sh --zookeeper dn1:2181 --reassignment-json-file expand-cluster-reassignment.json --execute

最后，--verify选项可与工具一起使用，以检查分区的重新分配状态。需要注意的是，同样的expand-cluster-reassignment.json（与--execute应与选项一起使用)--verify执行命令如下：

> ./kafka-reassign-partitions.sh --zookeeper dn1:2181 --reassignment-json-file expand-cluster-reassignment.json --verify

执行结果如下图所示：

同时，我们可以通过Kafka Eagle工具来查看Topic分区情况。

2.2 跨集群迁移

我们指的是跨集群迁移Kafka复制数据镜像的过程在多个集群之间，以避免与单个集群中的节点混淆。 Kafka附带一个用于在场Kafka集群之间的镜像数据工具。该工具从源集群使用并生成到目标集群。该镜像的一个常见用例是在另一个数据中心提供副本。

此外，您还可以操作许多此类镜像过程来增加吞吐量和容错（如果一个过程终止，其他过程将占用额外的负载）。将来自源集群Topic读取数据，并将其写入目标集群中具有相同名称的主题。事实上，“镜像”数据只不过是一个Kafka将消费者与生产者联系起来。

源集群和目标集群是完全独立的实体，可以有不同数量的分区，偏移量也会不同。因此，镜像集群并不是真正的意图作为容错机制（因为消费者的位置会有所不同）；因此，建议在正常的集群中复制。然而，镜像过程将被保留和使用信息Key分区可以按Key保留顺序。

以下是一个跨集群的单个集群Topic实例，命令如下：

> ./kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist ke03

需要注意的是，consumer.properties文件配置源Kafka集群Broker地址，producer.properties文件配置目标Kafka集群地址。若需迁移多个Topic，可以使用 --whitelist 'A|B如果需要迁移一切Topic，可以使用 --whitelist '*'。

3.结果预览

实施跨集群迁移命令后，使用目标集群Kafka Eagle中查看Topic Size看源集群是否与源集群相关。Topic Size大小相等，或者使用SQL结果如下图所示:

4.总结

跨集群迁移数据的本质是，Kafka启动消费者读取源集群数据，并将消费后的数据写入目标集群，在迁移过程中，可启动多个例子，提供迁出吞吐量。

到目前为止，关注博主不迷路，叶秋学长带你上高速公路