kafka和flink的动态扩容

问题一、kafka和flink是否可以动态扩容?

答：可以的。

问题二、扩容之后修改的配置是什么？扩容方法？

Kafka修改:

1）拷贝线上配置到本地kafka目录。 2）需要注意的是server.properties里broker和ip的指定，必须要唯一。
3）server.properties中log.dirs必须要手动指定。此配置项是kafka接收日志的存储目录
4）zookeeper的指定可以是下面这种形式，手动定制zookeeper的目录，并且用逗号分隔ip
zookeeper.connect=1.1.1.1:2181,1.1.1.2:2181,1.1.1.3:2181
5）默认分区和备份数分别为10和3（num.partitions=10 /
default.replication.factor=3）当然这也可以在创建topic时手动指定。

Kafka扩容：

1.部署环境，保证所有机器上运行环境相同。
2.JMX_PORT=9997 bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 & 部署新节点，然后通过命令启动kafka，JMX_PORT是为了打开kafka的可管理端口（默认不启用）。
3.根据"配置修改"修改配置，然后启动集群，确保新节点为可用状态。

4.生成扩容使用的json文件： cat topics-to-move.json {“topics”: [{“topic”: “topic1”}], //加粗部分是要手动指定的topic名称 “version”:1 }

5.通过上一步写好的json文件，使用kafka命令生成数据迁移配置 bin/kafka-reassign-partitions.sh --topics-to-move-json-file topics-to-move.json --zookeeper 1.1.1.99:2181/kafka-0.8.1 --broker-list “0,1,2,3,4” --generate //加粗部分是指定数据迁移到那些broker。

6.将第一部分保存留作回退备份(即Proposed partition reassignment configuration上面的json串)，下面json串为扩容将要使用的到的配置，将其保存为expand-cluster-reassignment.json

7.执行扩容命令： bin/kafka-reassign-partitions.sh --zookeeper 1.1.1.99:2181/kafka-0.8.1 --reassignment-json-file expand-cluster-reassignment.json --execute
正常执行的话会生成同上图类似的json串，表示原始状态和目标状态

8.查询执行状态： bin/kafka-reassign-partitions.sh --zookeeper 1.1.1.99:2181/kafka-0.8.1 --reassignment-json-file expand-cluster-reassignment.json --verify 正常执行后会返回当前数据迁移的不用partion的

9.注：数据迁移一旦开始无法停止，也不要强行停止集群，这样会造成数据不一致，带来无法挽回的后果。 kafka数据迁移的原理是先拷贝数据到目标节点，然后再删除原节点的数据。这样的话如果集群原节点空间不足，不要继续指定其为迁移broker，这样将造成原节点空间用尽，例如原节点是broker为0，1，2，3，4就不要这样指定
–broker-list “0,1,2,3,4”，应该这样 --broker-list “5,6”。另外数据迁移也可以通过手工定制。

Flink：
1.添加机器是，必须修改内容：

每个JobManager的可用内存量（jobmanager.heap.mb），
每个TaskManager（taskmanager.heap.mb）的可用内存量，
每台计算机可用的CPU数（taskmanager.numberOfTaskSlots），
集群中的CPU总数（parallelism.default）和临时目录（taskmanager.tmp.dirs）
2.以下脚本在本地节点上启动JobManager，并通过SSH连接到slaves文件中列出的所有辅助节点，以在每个节点上启动TaskManager。
#bin/start-cluster.sh

3.可以使用bin/jobmanager.sh和bin/taskmanager.sh脚本将JobManager和TaskManager实例都添加到正在运行的集群中。

4.添加JobManager bin/jobmanager.sh ((start|start-foreground) cluster)|stop|stop-all 添加任务管理器 bin/taskmanager.sh
start|start-foreground|stop|stop-all 确保在要启动/停止相应实例的主机上调用这些脚本。

注：

最大并行度必须满足以下条件： 0<parallelism<=max parallelism <=2^15
可以通过设置最大并行度setMaxParallelism(int
maxparallelism)。默认情况下，首次启动作业时，Flink将根据并行度选择最大并行度： 128 ：对于所有并行度<= 128。
MIN(nextPowerOfTwo(parallelism + (parallelism / 2)), 2^15) ：对于所有并行度>128的情况。

问题三、脚本编写方法，方向：

Flink：
1.实际需要修改参数：集群中的CPU总数（parallelism.default）
注：要小于cpu总数一点
2.修改脚本里参数：slave ip

Kafka：
1）拷贝线上配置到本地kafka目录。
2）需要注意的是server.properties里broker和ip的指定，必须要唯一。
3）server.properties中log.dirs必须要手动指定。此配置项是kafka接收日志的存储目录
4）zookeeper的指定可以是下面这种形式，手动定制zookeeper的目录，并且用逗号分隔ip
zookeeper.connect=1.1.1.1:2181,1.1.1.2:2181,1.1.1.3:2181

增加broker数量

新添加的Kafka节点并不会自动地分配数据，所以无法分担集群的负载，除非我们新建一个topic。但是现在我们想手动将部分分区移到新添加的Kafka节点上，Kafka内部提供了相关的工具来重新分布某个topic的分区。在重新分布topic分区之前，我们先来看看现在topic的各个分区的分布位置：
./bin/kafka-topics.sh --topic iteblog --describe --zookeeper zookeeper:2181
Topic:iteblog PartitionCount:7 ReplicationFactor:2 Configs:

由于查看到的分区是4个，所以现在我们需要将原先分布在broker 1-4节点上的分区重新分布到broker 1-5节点上，借助kafka-reassign-partitions.sh工具生成reassign plan，不过我们先得按照要求定义一个文件，里面说明哪些topic需要重新分区，文件内容如下：

#bin/kafka-reassign-partitions.sh --zookeeper zookeeper:2181 --topics-to-move-json-file topics-to-move.json --broker-list “1,2,3,4,5” --generate

自动生成执行文件
#cat topics-to-move.json
{“topics”: [{“topic”: “iteblog”}],
“version”:1
}

这个文件也可以自己手动添加：
{
“version”: 1,
“partitions”: [
{
“topic”: “iteblog”,
“partition”: 0,
“replicas”: [
1,
2
]
},
{
“topic”: “iteblog”,
“partition”: 1,
“replicas”: [
2,
3
]
},
{
“topic”: “iteblog”,
“partition”: 2,
“replicas”: [
3,
4
]
},
{
“topic”: “iteblog”,
“partition”: 3,
“replicas”: [
4,
5
]
},
{
“topic”: “iteblog”,
“partition”: 4,
“replicas”: [
5,
1
]
},
{
“topic”: “iteblog”,
“partition”: 5,
“replicas”: [
1,
3
]
},
{
“topic”: “iteblog”,
“partition”: 6,
“replicas”: [
2,
4
]
}
]
}

#bin/kafka-reassign-partitions.sh --zookeeper zookeeper:2181 --reassignment-json-file result.json --execute

验证：
#bin/kafka-reassign-partitions.sh --zookeeper zookeeper:2181 --reassignment-json-file result.json --verify

最后再查一下分区状况
#./bin/kafka-topics.sh --topic iteblog --describe --zookeeper zookeeper:2181

Topic扩容：
#./bin/kafka-topics.sh --zookeeper 192.168.2.225:2183 --alter --partitions 15 --topic push-token-topic

#./bin/kafka-reassign-partitions.sh --zookeeper 192.168.2.225:2183
–reassignment-json-file partitions-extension-push-token-topic.json --execute

脚本partitions-extension-push-token-topic.json内容：
{
“partitions”:
[
{
“topic”: “push-token-topic”,
“partition”: 12,
“replicas”: [101,102]
},
{
“topic”: “push-token-topic”,
“partition”: 13,
“replicas”: [103,104]
},
{
“topic”: “push-token-topic”,
“partition”: 14,
“replicas”: [105,106]
}
],
“version”:1
}