kafka 分区迁移

冷不防 2022-08-31 09:46 162阅读 0赞

**扩容kafka之后，针对扩容之前的topic进行重新平衡leader，Replicas，Isr**

[【kafka运维】你真的懂数据迁移吗？][kafka]

[分区副本重分配+注意事项+LogiKM简化迁移流程(1)][LogiKM_1]

[Kafka的灵魂伴侣Logi-KafkaManger(4)之运维管控–集群运维(数据迁移和集群在线升级)][Kafka_Logi-KafkaManger_4]

# 1.broker间分区迁移 #

**生成分配计划获取kafka推荐的重分配策略**

1.  生成分配计划  
    新建一个json文件 move-json-file.json 文件名可随便取  
    新建一个json文件，内容如下：（topic为要修改的topic）这里文件取名为  
    move-json-file.json  
    ![在这里插入图片描述][1c844df5af334791aacc908f3ab5c3dc.png]

{ 
    “topics”: [{ 
    “topic”: “test10”
    }],
    “version”: 1
    }

1.  使用kafka-reassign-partitions.sh命令生成一个分配计划  
    下面的命令行中broker-list参数即对应BrokerID  
    命令如下：

bin/kafka-reassign-partitions.sh --zookeeper zk01.td.com:2181 \
    –topics-to-move-json-file topics-to-move.json \
    –broker-list “0,1” \
    –generate

结果入下图，生成了两条信息，第一条为现在的分配情况，第二条为计划更改的内容  
![在这里插入图片描述][567dad29e87f403d81d09e7edf962d9a.png]  
把计划修改的结果复制，放在第二个json文件中，这里取名为reassignment-json-file.json  
![在这里插入图片描述][b824b1e4305d447793b5ab70640d6ccf.png]

3.根据推荐执行分配计划  
运行kafka-reassign-partition.sh命令根据上述执行计划生成的结果进行分配，命令如下：

bin/kafka-reassign-partitions.sh \
    –zookeeper zk01.td.com:2181 \
    –reassignment-json-file reassignment-json-file.json \
    –execute

![在这里插入图片描述][6f33b990a601458eaa0b9d86836314e9.png]  
\*\*  
迁移的过称的思想是，增加副本，同步数据，删除原来的副本  
迁移的过程是先把原来的分区日志文件标记为删除状态进行删除，同时在要迁移的位置新建文件日志，其实就是在要迁移的位置新建一个副本，同步完成之后，再把原来的位置的日志文件删掉\*\*

查看topic，已经重新分配

1.  验证迁移成功  
    迁移完成后，使用 --verify 来验证  
    如果在迁移过程中添加限流，使用 --verify 来验证，会把迁移过程中的限流也删除掉  
    ![在这里插入图片描述][3fe39eebd299439382be0a295689ad8f.png]

# 2.磁盘间、不同路径分区迁移 #

**1.1版本正式支持分区在不同路径间的迁移**

**为什么线上Kafka机器各个磁盘间的占用不均匀，经常出现“一边倒”的情形？这是因为Kafka只保证分区数量在各个磁盘上均匀分布，但它无法知晓每个分区实际占用空间，故很有可能出现某些分区消息数量巨大导致占用大量磁盘空间的情况。在1.1版本之前，用户对此毫无办法，因为1.1之前Kafka只支持分区数据在不同broker间的重分配，而无法做到在同一个broker下的不同磁盘间做重分配。1.1版本正式支持副本在不同路径间的迁移**

在一台Broker上用多个路径存放分区  
在kafka/conf/server.properties

log.dirs=kafka-logs-5,kafka-logs-6,kafka-logs-7,kafka-logs-8

注意同一个Broker上不同路径只会存放不同的分区，而不会将一个分区的多个副本存放在同一个Broker; 不然那副本就没有意义了(容灾)

1.  准备迁移文件  
    迁移的json文件有一个参数是log\_dirs; 默认请求不传的话 它是"log\_dirs": \[“any”\] （这个数组的数量要跟副本保持一致） 但是你想实现跨路径迁移,只需要在这里填入绝对路径就行了,例如下面

迁移的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"]
      }]
    }

1.  执行脚本

bin/kafka-reassign-partitions.sh --zookeeper xxxxx --reassignment-json-file config/reassignment-json-file.json --execute --bootstrap-server
    xxxxx:9092 --replica-alter-log-dirs-throttle 10000

# 3.迁移的优化 #

## 临时缩小数据保存时间 ##

**数据量太大，需要将过期的.log数据保存时间临时缩小**  
减少迁移的数据量: 如果要迁移的Topic 有大量数据（Topic 默认保留7天的数据），可以在迁移之前临时动态地调整retention.ms 来减少数据量，比如下面命令改成1小时; Kafka 会主动purge 掉1小时之前的数据;

动态修改topic的日志保留时间：  
增删改 配置 --alter

**可以先去zk中查看对应的配置文件**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]

1.  添加配置  
    –alter

删除配置: --delete-config k1=v1,k2=v2
    添加/修改配置: --add-config k1,k2
    选择类型: --entity-type (topics/clients/users/brokers/broker-
    loggers)  # 如果针对topic级别的修改 --entity-type topics 
    类型名称: --entity-name    # 如果上面的选择类型(--entity-type topics)针对topic级别的修改 ,需要指定topic的名字 --entity-name topicName

Topic添加/修改动态配置  
–add-config

bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --add-config file.delete.delay.ms=3000,retention.ms=36000

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]

1.  数据过期删除后，开始执行迁移
2.  迁移完成前删除配置  
    Topic删除动态配置  
    –delete-config

sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --delete-config file.delete.delay.ms,retention.ms

## 迁移时限流 ##

迁移过程注意流量陡增对集群的影响 ，Kafka提供一个broker之间复制传输的流量限制，限制了副本从机器到另一台机器的带宽上限，这个限制主要是显示副本之间的数据同步

当重新平衡集群，引导新broker，添加或移除broker时候，这是很有用的。因为它限制了这些密集型的数据操作从而保障了对用户的影响、 例如我们上面的迁移操作加一个限流选项-- throttle 50000000 \#单位 B/s

**注意：限流也会限制其他没有参加迁移的分区副本之间的限流，需要计算好正常的副本间同步流量，不要小于此值**

执行迁移脚本时，增加限流

1.  跨broker的分区迁移限流

bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute -- throttle 50000000

在后面加上一个—throttle 50000000 参数, 那么执行移动分区的时候,会被限制流量在50000000 B/s 加上参数后你可以看到

1.  磁盘间、不同路径分区迁移限流

需要注意的是,如果你迁移的时候包含 副本跨路径迁移(同一个Broker多个路径)那么这个限流措施不会生效,你需要再加上 --replica-alter-log-dirs-throttle 这个限流参数,它限制的是同一个Broker不同路径直接迁移的限流;

如果你想在重新平衡期间修改限制，增加吞吐量，以便完成的更快。你可以重新运行execute命令，用相同的reassignment-json-file

1.  迁移完后后要删除限流设置

[kafka]: https://mp.weixin.qq.com/s/fQ03wpctV1dGnmk1r-xEWA
[LogiKM_1]: https://mp.weixin.qq.com/s/Qkht7jwqVqKwcf5-wJYIPQ
[Kafka_Logi-KafkaManger_4]: https://shirenchuang.blog.csdn.net/article/details/116119900
[1c844df5af334791aacc908f3ab5c3dc.png]: /images/20220829/f64bff194f2c42299320b92df41ad998.png
[567dad29e87f403d81d09e7edf962d9a.png]: /images/20220829/c07e1a7f6bad47cdbd2e3474c3e3d928.png
[b824b1e4305d447793b5ab70640d6ccf.png]: /images/20220829/19dc67ba044f47bab512933c1fffd9a0.png
[6f33b990a601458eaa0b9d86836314e9.png]: /images/20220829/f17992e220524fd19b0818fcd10d38b1.png
[3fe39eebd299439382be0a295689ad8f.png]: /images/20220829/efb3faade6df470a9fe55b32ca97bb1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20220829/9f064287e9ee433f87b48ea73c7927dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]: /images/20220829/fea7cf7d1ec34e2f93fa6d8b8ca92a15.png