本文讲解以下内容【kafka运维】replicascaling,datamigration,replicaredistribution,replicacross-pathmigrationscriptparameter参数说明示例--zookeeperconnectionzk--zookeeperlocalhost:2181,localhost:2182--topics-to-move-json-file指定json文件，文件内容为topic配置--topics-to-move-json-fileconfig/move-json-file.jsonJson文件格式如下：--generate尝试给出一个副本重新分配策略，这个命令实际上并不执行--broker-list指定一个特定的BrokerList尝试给出一个分配策略，与--generate一起使用--broker-list0,1,2,3--reassignment-json-file指定要重新赋值的json文件，与--execute一起使用时json文件格式如下。例如：--execute开始执行重新赋值任务，与--reassignment-json-file一起使用使用--verify验证任务是否执行成功。如果使用--throttlecurrentlimit，该命令也会移除当前限制；这个命令很重要，如果不取消限流，正常副本之间的同步会受到影响。--throttle迁移过程中当前Brokers间的进程传输速率，单位bytes/sec--throttle500000--replica-alter-log-dirs-throttlebroker内部副本跨路径迁移数据限流功能，限制数据copyfrom一个目录转到另一个目录带宽上限单位字节/秒--replica-alter-log-dirs-throttle100000--disable-rack-aware关闭机架感知能力，分配时不参考机架信息--bootstrap-server如果跨路径迁移副本，则必须使用此参数。1.脚本使用介绍该脚本是Kafka提供的用于重新分配分区的脚本工具；1.1生成推荐配置脚本的关键参数--generate在分区副本重新分发之前，最好使用以下方法获取合理的分发文件；编写move-json-file.json文件；这个文件是告诉你要重新分配哪些topic的计算{"topics":[{"topic":"test_create_topic1"}],"version":1}然后执行下面的脚本，--broker-list"0,1,2,3"这个参数就是你要分配的Brokers；shbin/kafka-reassign-partitions.sh--zookeeperxxx:2181--topics-to-move-json-fileconfig/move-json-file.json--broker-list"0,1,2,3"--generate将打印Currentpartitionreplicaassignment//当前副本分配{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[3],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[2],"log_dirs":["any"]}]}Proposedpartitionreassignmentconfiguration//预期的重分配方式{"version":1,"partitions":[{"topic":"test_create_topic1",“分区”：2，“副本”：[2]，“log_dirs”：[“任何”]}，{“topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]}需要注意的是此时分区移动还没有开始，只是告诉你当前的分配和建议保存当前分配，以防您想要将其回滚1.2。执行Json文件关键参数--execute将上面想要的reassignment方法文件保存在json文件中reassignment-json-file.json{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]}然后执行shbin/kafka-reassign-partitions.sh--zookeeperxxxxx:2181--reassignment-json-fileconfig/reassignment-json-file.json--execute迁移过程注意流量突然增加集群影响Kafka对broker之间的复制传输提供流量限制，这限制从一台机器到另一台机器的副本带宽。这在重新平衡集群、引导新代理以及添加或删除代理时很有用。.因为它限制了这些密集的数据操作，以确保对用户的影响，例如我们在上面的迁移操作中添加一个限流选项--throttle50000000>shbin/kafka-reassign-partitions.sh--zookeeperxxxxx:2181--reassignment-json-fileconfig/reassignment-json-file.json--execute--throttle50000000在最后添加一个--throttle50000000参数，那么在执行移动分区时，流量将被限制在50000000B/s添加后参数，可以看到Thethrottlelimitwassetto50000000B/sSuccessfullystartedreassignmentofpartitions。需要注意的是，如果你的迁移包含副本跨路径迁移（同一个Broker多路径），那么这个限流措施将不会生效，你需要添加|--replica-alter-log-dirs-throttle这个当前limiting参数，限制同一个Broker不同路径直接迁移的当前限制；如果想在rebalancing的时候修改limit，增加throughput，这样可以更快的完成。您可以使用相同的reassignment-json-file1.3重新运行执行命令。验证关键参数--verify该选项用于检查分区重新分配的状态，--throttle流量限制也会被移除；否则可能会导致周期性复制操作的流量也受限。shbin/kafka-reassign-partitions.sh--zookeeperxxxx:2181--reassignment-json-fileconfig/reassignment-json-file.json--verifyimage注意：当你输入的BrokerId不存在时，复制操作会失败，但不影响他人；例如图2，副本伸缩Kafka没有提供专门的脚本来支持副本伸缩，不像kafka-topic.sh脚本那样可以扩展分区；认为要扩容和缩容副本，唯一救国的方法就是使用kafka-reassign-partitions.sh重新分配副本2.1副本扩容假设我们目前的情况是3分区1副本，为了提供可用性,我想增加副本数到2;2.1.1计算副本分配方式我们在步骤1.1中使用--generate获取当前分配情况，得到如下json{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]}我们想把replica中的所有partition都变成2，那么我们只需要修改"replicas"中的值：[]，就是Broker列表，第一个就是Leader；所以我们根据自己想要的分布规则进行修改把json文件改成如下{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0,1],"log_dirs":["any","any"]}]}注意log_dirs中的数字应该是与匹配的副本数相同；或者直接删除log_dirs选项；这个log_dirs是副本跨路径迁移时的绝对路径2.1.2Execute--execute如果要在rebalancing的时候修改limit，增加throughput，这样可以更快的完成可以重新执行execute命令，使用相同的reassignment-json-file：2.1.2Verification--verify图片完成后，副本数会增加；2.2CopyshrinkingCopyshrinking和expansion是一个意思；当副本数量减少，超过之前的数量时，多出来的副本会被删除；例如，我刚刚添加了一个新副本，想将其恢复到一个副本执行以下json文件{"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,“副本”：[2]，“log_dirs”：[“任何”]}，{“主题”：“test_create_topic1”，“分区”：1，“副本”：[1]，“log_dirs”：[“任何”]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]}执行后可以看到其他replicas被标记为删除；他们将在一段时间后被清理干净。虽然我们通过这种方式实现了副本的扩缩容，但是副本的分配需要我们自己来控制。做负载均衡等；没有Kafka自动帮我们分配肯定更合理；那有没有什么好的方法可以帮我们合理分配Json文件呢？PS：之前我们分析过【kafka源码】】创建Topic时，如何分配partitions和replicas？然后我们用同样的规则来分配这样一个分配过程。这样不行吗？--generate本质上就是调用这个方法，AdminUtils.assignReplicasToBrokers(brokerMetadatas,assignment.size,replicas.size)具体实现操作请参考【Kafka思维】最小成本扩容缩容副本设计方案写一个项目实现类似的方法。如果觉得麻烦，可以直接使用LogIKM新增的replica功能，直接帮你搞定；（以后会实现）3.分区扩容Kafka的分区扩容是通过kafka-topis.sh脚本实现的；如果不支持缩容和分区扩容，请看【kafka源码】TopicCommand的alter源码分析（分区扩容）4.分区迁移和分区迁移同上。请参阅第1.1、1.2、1.3节；5.跨路径迁移副本，但无法获知每个分区实际占用的空间，所以很可能部分分区消息较多，占用磁盘空间较大。在1.1版本之前，用户是没有办法做到这一点的，因为1.1之前的Kafka只支持分区数据在不同broker之间的重分配，而不能在同一个broker下的不同磁盘之间进行重分配。1.1版本正式支持副本在不同路径之间的迁移。如何使用多个路径在Broker上存储分区？你只需要在配置中连接多个文件夹##############################LogBasics################################Acommaseparatedlistofdirectoriesunderwhichtostorelogfileslog.dirs=kafka-logs-5,kafka-logs-6,kafka-logs-7,kafka-logs-8注意同一个Broker上的不同路径只会存储不同的分区，不会在同一个Broker中存储副本；否则则副本无意义（容灾）如何跨路径迁移？迁移后的json文件有个参数log_dirs；如果默认请求不通过，则为"log_dirs":["any"]（这个数组的个数必须和Copy保持一致）但是如果要实现跨路径迁移，只需要填写这里的绝对路径，例如下面迁移的json文件的例子{"version":1,"partitions":[{"topic":"test_create_topic4","partition":2,"replicas":[0],"log_dirs":["/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-5"]},{"topic":"test_create_topic4","partition":1,"replicas":[0],"log_dirs":["/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-6"]}]}然后执行脚本shbin/kafka-reassign-partitions.sh--zookeeperxxxxx--重新分配-json-fileconfig/reassignment-json-file.json--execute--bootstrap-serverxxxxx:9092--replica-alter-log-dirs-throttle10000注意--bootstrap-server在跨路径迁移的情况下，必须传入该参数如果需要限流的话，加上参数|--replica-alter-log-dirs-throttle；与--throttle的区别在于--replica-alter-log-dirs-throttle限制了Broker中不同路径的迁移流量；源码源码分析可参考文章【kafka源码】ReassignPartitionsCommand源码分析（复制缩放、数据迁移、分区重分配、复制跨路径迁移）视频链接：https://mp.weixin.qq。com/s/fQ03wpctV1dGnmk1r-xEWA本文转载自微信公众号“石珍珍的杂货店”，您可以通过以下二维码关注转载文章，请联系石珍珍的杂货店公众号。
