Fork me on GitHub

KAFKA扩容节点和分区迁移-CDH

字数统计: 1.4k字   |   阅读时长≈ 6分 
1
2
由于服务器老化原因，需要迁移CDH KAFKA集群中的一台服务器，上面运行有KAFKA服务，假设broker id为536
假设老化服务器IP地址是10.0.0.200

CDH新增KAFKA节点

1
2
3
(1)通过CDH 管理界面在KAFKA集群上新增一个KAFKA角色,并启动KAFKA服务,假设broker id为577 。
(2)停用并删除集群中10.0.0.200上的KAFKA角色
(3)查看现有KAFKA集群节点的 Kafka broker id，假设为(535,577)

CDH新增KAFKA节点后，若出现启动错误

1
2
3
请检查如下配置：
     （1）JDK版本是否与其他KAFKA节点版本一致，不一致则需要升级
      (2) 若启动日志扔报错，需要检查CDH管理页面，新增kafka节点的配置是否与其他正常节点一致，尤其是参数Java Heap Size of Broker in Megabytes这两个参数，如果设置为0，则会导致启动KAFKA服务失败。

查看Topic

在有权限的服务器上，查看所有的Topic列表

1
bin/kafka-topics.sh --zookeeper 10.0.0.133:2181 --describe --list

查看具体某个Topic分区情况

1
bin/kafka-topics.sh --zookeeper 10.0.0.133:2181 --describe --topic time
1
2
3
4
5
Topic:time	PartitionCount:2	ReplicationFactor:2	Configs:
Topic: time	Partition: 0	Leader: 535	Replicas: 535,536	Isr: 535,536
Topic: time	Partition: 1	Leader: 536	Replicas: 536,535	Isr: 536,535

可以看到，虽然新KAFKA节点已经加入了集群，但已有的Topic分区不会自动复制到新KAFKA节点，故需要进行分区迁移

创建要迁移的Topic列表

1
2
3
[在有权限执行查看topic命令的机器上操作]

新建文件topics-to-move.json，其中, 包含要迁移到Topic列表。这里只迁移了一个Topic，也可以是多个Topic。
1
2
3
4
5
6
7
8
9
10
mkdir -p partner
vim partner/topics-to-move.json

内容：
{
  "topics": [
       {"topic": "time"}
     ],
  "version": 1
}

生成Topic分区分配表

1
使用kafka-reassign-partitions命令生成分区分配表，其中需要指定topics-to-move.json文件和迁移目标节点的broker id
1
bin/kafka-reassign-partitions.sh --zookeeper 10.0.0.133:2181 --topics-to-move-json-file ./partner/topics-to-move.json --broker-list "535,577" --generate
1
2
3
4
5
6
7
结果如下：
Current partition replica assignment
{"version":1,"partitions":[{"topic":"time","partition":1,"replicas":[535,536]},{"topic":"ktime_log","partition":0,"replicas":[536,535]}]}


Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"time","partition":1,"replicas":[535,577]},{"topic":"ktime_log","partition":0,"replicas":[577,535]}]}
1
2
3
4
5
注：
   (1)将Current partition replica assignment部分的内容保存到rollback-cluster-reassignment.json，用于回滚操作。
   (2)将Proposed partition reassignment configuration 的内容保存到expand-cluster-reassignment.json，用于执行迁移操作。
   
   保存完毕，可以手工编辑expand-cluster-reassignment.json 文件来更改replica 和partition 配置。

执行迁移操作

1
bin/kafka-reassign-partitions.sh --zookeeper 10.0.0.133:2181 --reassignment-json-file partner/expand-cluster-reassignment.json --execute
1
2
3
4
5
6
7
8
注：迁移操作会将指定Topic 的数据文件移动到新的节点目录下，这个过程可能需要等待很长时间，视Topic 的数据量而定。

执行结果如下：
Current partition replica assignment
{"version":1,"partitions":[{"topic":"time","partition":1,"replicas":[535,536]},{"topic":"time","partition":0,"replicas":[536,535]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions {"version":1,"partitions":[{"topic":"time","partition":1,"replicas":[535,577]},{"topic":"time","partition":0,"replicas":[577,535]}]}

验证迁移是否成功

1
2
3
4
5
6
bin/kafka-reassign-partitions.sh --zookeeper 10.0.0.133:2181 --reassignment-json-file partner/expand-cluster-reassignment.json --verify

执行结果如下：
Status of partition reassignment:
Reassignment of partition [time,1] completed successfully
Reassignment of partition [time,0] completed successfully
1
2
3
4
5
6
7
迁移状态有两种：
in progress 表示正在迁移
completed successlly 表示已经成功完成迁移。

如果命令出错或者遇到异常，分配状态会变成“failed” ；如果是json配置文件的原因，命令行就会打印出error日志；如果是kafka内部错误可以查看kafka的server.log 日志文件等

注：在此过程中，可以在各个Kafka 的节点上使用iftop 工具实时监控网络带宽。如果数据量较大，会占用大量带宽来完成数据同步。

查看topic详情-再次确认迁移

1
bin/kafka-topics.sh --zookeeper 10.0.0.133:2181 --describe --topic time
1
2
3
4
5
Topic:time	PartitionCount:2 ReplicationFactor:2	Configs:
Topic: time	Partition: 0	Leader: 535	Replicas: 535,577	Isr: 535,577
Topic: time	Partition: 1	Leader: 577	Replicas: 577,535	Isr: 535,577

可以看到，原已有的topic 分区数据已经到了新KAFKA节点上。

排错

1
2
3
4
如果某个KFAKA节点已经down掉，而Kafka分区唯一副本恰好在这个节点上，那么执行如上的热迁移分区命令，会一直卡在“in progress表示正在迁移“状态，而不会报错。

而在此情况下，直接进行下一个分区热迁移，会报如下错误：
kafka.common.AdminCommandFailedException: Partition reassignment currently in progress for Map(). Aborting operation

解决方法：

1
问题原因是：KAFKA节点down，分区热迁移导致zk上自动创建了/admin/reassign_partitions，这个残留文件，会导致正常的reassigin都没办法进行，于是只能命令行登陆zk，直接删除该节点

zookeeper连接server

1
bin/zkCli.sh -server 10.0.0.133:2181
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper, admin, consumers, config, controller, brokers, controller_epoch]
[zk: localhost:2181(CONNECTED) 1] ls /admin/reassign_partitions
[]
[zk: localhost:2181(CONNECTED) 2] get /admin/reassign_partitions
{"version":1,"partitions":[]}
cZxid = 0xd00008216
ctime = Mon Oct 26 14:47:30 CST 2015
mZxid = 0xd00008216
mtime = Mon Oct 26 14:47:30 CST 2015
pZxid = 0xd00008216
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 29
numChildren = 0
[zk: localhost:2181(CONNECTED) 3] rmr /admin/reassign_partitions

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true


不论我写了什么...
其实都是错的......