kafka命令记录

基于kafka0.8.2

1.kafka创建topic

kafka-topics.sh --create --zookeeper $ZK_CONNECT --topic test1 --partitions 2 --replication-factor 2

2.kafka创建topic，手动指定partition数，副本数在哪些broker id上

kafka-topics.sh --create --zookeeper $ZK_CONNECT --topic testass --replica-assignment 0:1,1:2,2:0

3.查看topic信息

kafka-topics.sh --describe --zookeeper $ZK_CONNECT --topic testass

Topic:testass PartitionCount:3 ReplicationFactor:2 Configs:

Topic: testass Partition: 0 Leader: 0 Replicas: 0,1 Isr: 0,1

Topic: testass Partition: 1 Leader: 1 Replicas: 1,2 Isr: 1,2

Topic: testass Partition: 2 Leader: 2 Replicas: 2,0 Isr: 2,0

4.删除topic信息

kafka-topics.sh --delete --zookeeper $ZK_CONNECT --topic test1

5.kafka脚本只可以增加partition数，不可以减少分区数，不可以修改replication-factor

kafka-topics.sh --zookeeper $ZK_CONNECT --topic testass --alter --partitions 4（将partition数增加1）

6.kafka为topic增加配置

kafka-topics.sh --zookeeper $ZK_CONNECT --alter --topic testass --config x=y

7.kafka为topic减少配置

kafka-topics.sh --zookeeper $ZK_CONNECT --alter --topic testass --deleteConfig x

8.当kafka集群新增机器时，已经创建的topic不会自动进行负载均衡。

场景为自动迁徙数据到新增的机器，将迁徙topic foo1,foo2 到brokers 5,6

（1）创建要迁徙的文件topics-to-move.json

cat topics-to-move.json

{"topics": [{"topic": "foo1"},

{"topic": "foo2"}],

"version":1

}

（2）生成一个要迁徙的新的文件目录，执行如下命令

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --topics-to-move-json-file topics-to-move.json --broker-list "5,6" --generate

执行结果显示：

Current partition replica assignment

{"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]}]

}

Proposed partition reassignment configuration

{"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]}]

}

（3）将上一步生成的Proposed partition reassignment configuration 配置文件写入文件中，例如写入expand-cluster-reassignment.json中

（4）执行迁徙过程

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file expand-cluster-reassignment.json --execute

（5）验证迁徙过程

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file expand-cluster-reassignment.json --verify

9.自定义指定分区的迁徙

（1） 创建迁徙的配置文件

cat custom-reassignment.json

{"version":1,"partitions":[{"topic":"foo1","partition":0,"replicas":[5,6]},{"topic":"foo2","partition":1,"replicas":[2,3]}]}

这个配置文件的意思是将foo1的0分区迁徙到broker id为5,6的broker上，将foo2的1分区迁徙到broker id 为2,3的broker上。

（2）执行迁徙过程

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file custom-reassignment.json --execute

（3）执行验证

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file custom-reassignment.json --verify

10.增加分区的replication factor

（1） 创建迁徙的配置文件

cat increase-replication-factor.json

{"version":1,

"partitions":[{"topic":"foo","partition":0,"replicas":[5,6,7]}]}

假设foo原来只有broker 5一个replic，现在要增加broker 6,7上。

（2）执行迁徙过程

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file increase-replication-factor.json --execute

（3）执行验证

kafka-reassign-partitions.sh --zookeeper $ZK_CONNECT --reassignment-json-file increase-replication-factor.json --verify