域名网站打开慢,python18+21,南充房产网南充房产信息网,建站 小语种 连接文章目录 一、同集群迁移二、跨集群迁移 Kafka两种迁移场景#xff0c;分别是同集群数据迁移、跨集群数据迁移。 一、同集群迁移 应用场景#xff1a; broker 迁移 主要使用的场景是broker 上线,下线,或者扩容等.基于同一套zookeeper的操作。
实践#xff1a; 将需要新添加… 文章目录 一、同集群迁移二、跨集群迁移 Kafka两种迁移场景分别是同集群数据迁移、跨集群数据迁移。 一、同集群迁移 应用场景 broker 迁移 主要使用的场景是broker 上线,下线,或者扩容等.基于同一套zookeeper的操作。
实践 将需要新添加的broker 列表一并添加到kafka的集群中。启动新的kafka指定同一套zkKafka由之前的三节点扩容至四节点。 数据迁移
查询信息四个分区分布在三台机器上 新建json文件
cat topic-to-move.json{
topics: [
{topic: test-topic}
],
version:1
}获取重新分配方案
kafka-reassign-partitions.sh --zookeeper node1:2181,node2:2181,node3:2181 --topics-to-move-json-file topics-to-move.json --broker-list 150,151,155,159 –generate##通过kafka-reassign-partitions.sh 获取重新分配方案,–broker-lsit 的参数 “150,151,155,159是指集群中每个broker的id由于我们是需要将所有topic均匀分配到扩完结点的4台机器上所以要指定。同理当业务改变为将原来的所有数据从旧节点0,5,9迁移到新节点1实现数据平滑迁移这时的参数应4” ,执行后会出现以下内容: 复制新的方案到一个json文件 assignplan.json 文件名不重要文件格式也不一定要以json为 结尾只要保证内容是json即可 ./kafka-reassign-partitions.sh --zookeeper node1:2181 --reassignment-json-file assignplan.json --execute完成后查看topic信息四个分区分布在四台机器上
二、跨集群迁移 应用场景 主要用于kafka 集群的变更. 将数据同步等操作. 相当于是实现了双打.等各项数据消费端在零误差的对接好了后,可以停掉就集群
数据迁移: MirrorMaker
修改kafka配置 consumer.properties内容为原始集群信息
#config/consumer.properties 在网上看到有在此配置zookeeper的应该是之前的老版本。kafka_2.11-2.4.1中不需要
bootstrap.serverskafka-cluster1:9092,kafka-cluster1:9093 # source-cluster的broker list
group.idtest-consumer-group1 # 自定义一个消费者的group id
auto.offset.reset # latest当各分区下有已提交的offset时从提交的offset开始消费无提交的offset时消费新产生的该分区下的数据 earliest当各分区下有已提交的offset时从提交的offset开始消费无提交的offset时从头开始消费 nonetopic各分区都存在已提交的offset时从offset后开始消费只要有一个分区不存在已提交的offset则抛出异常producer.properties内容为新集群信息 #config/producer.properties 在网上看到有在此配置zookeeper的应该是之前的老版本。kafka_2.11-2.4.1中不需要 bootstrap.serverskafka-cluster2:9092,kafka-cluster2:9093 # destination-cluster的broker list compression.typenone # 数据压缩方式none, gzip, snappy, lz4, zstd partitioner.class # 指定分区程序路径默认为随机分区 request.timeout.ms # 请求超时时间 max.block.ms # KafkaProducer.send and KafkaProducer.partitionsFor 阻塞时间 linger.ms # 等待指定时间后批量发送 max.request.size # 发送消息最大字节数 batch.size # 单次批量处理的字节数 buffer.memory # 指定等待发送消息的缓冲区大小
执行操作
./kafka-mirror-maker.sh --consumer.config ../config/consumer.properties --producer.config ../config/producer.properties --whitelist test说明
–num.streams: 指定流就是指定消费者所有消费者公用一个生产者。–whitelist: 表明需要同步的白名单可以使用”|”来连接多个topic还可以使用正则表达式。可设置黑名单。
ZK迁移
zk迁移就比较简单了起新节点加入zk集群稳定后关停旧节点。新增broker加入集群将所有topic分区只分配给新broker执行分配任务后kafka将旧broker的分区数据复制到新broker新broker成为各分区的leader随后kafka删除旧broker上的分区数据整个过程中客户端应用正常生产消费消息执行结束后使用新的消费者组从头消费可以获取到全部历史消息。停止旧broker后正在运行的客户端应用正常生产消费消息新建客户端连接旧broker失败连接新broker正常
