Kafka 数据迁移在 Debian 上的步骤

Kafka数据迁移在Debian上的步骤

一、准备工作

1. 安装JDK：Kafka依赖Java运行环境，需在Debian上安装OpenJDK（推荐8或11）。执行以下命令：

   sudo apt update &
       &
        sudo apt install -y openjdk-8-jdk
   

2. 下载并解压Kafka：从Apache Kafka官网下载所需版本（如3.5.2），使用tar命令解压：

   wget https://downloads.apache.org/kafka/3.5.2/kafka_2.12-3.5.2.tgz
   tar -xzf kafka_2.12-3.5.2.tgz -C /opt/
   

3. 配置环境变量：编辑/etc/profile文件，添加Kafka路径：

   echo 'export KAFKA_HOME=/opt/kafka_2.12-3.5.2' >
       >
        /etc/profile
   echo 'export PATH=$PATH:$KAFKA_HOME/bin' >
       >
    /etc/profile
   source /etc/profile  # 生效配置
   

二、同集群数据迁移（分区调整）

若需在同一个Kafka集群内迁移分区（如新增Broker节点），可使用kafka-reassign-partitions.sh工具：

1. 添加新Broker：编辑server.properties文件，为新Broker分配唯一ID（如broker.id=3），并配置listeners、log.dirs等参数，启动新Broker。
2. 生成重新分配计划：创建JSON文件（如reassign.json），指定待迁移的分区及目标Broker，执行命令生成计划：

   cat reassign.json
   # 示例内容：{
   "version":1,"partitions":[{
   "topic":"test_topic","partition":0,"replicas":[1,2,3]}
   ]}
       
   kafka-reassign-partitions.sh --zookeeper localhost:2181 --generate --topics-to-move-json-file reassign.json --broker-list "1,2,3" >
    reassign-plan.json
   

3. 执行重新分配：根据生成的reassign-plan.json执行迁移：

   kafka-reassign-partitions.sh --zookeeper localhost:2181 --execute --reassignment-json-file reassign-plan.json
   

4. 验证完成：检查分区状态，确认数据迁移成功：

   kafka-reassign-partitions.sh --zookeeper localhost:2181 --verify --reassignment-json-file reassign-plan.json
   

三、跨集群数据迁移（常用工具）

1. 使用MirrorMaker（Kafka自带）

MirrorMaker可实现跨集群数据镜像，适用于大规模数据同步：

1. 配置MirrorMaker：在目标集群上创建mirror-maker.properties文件，配置源集群（bootstrap.servers=source:9092）和目标集群（target.bootstrap.servers=target:9092）地址：

   # 源集群配置
   bootstrap.servers=source-kafka:9092
   # 目标集群配置
   target.bootstrap.servers=target-kafka:9092
   # 消费者组
   group.id=mirror-maker-group
   # 主题白名单（可选）
   topics=.*
   

2. 启动MirrorMaker：执行命令开始同步：

   kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist ".*"
   

3. 监控进度：通过Kafka自带的kafka-consumer-groups.sh工具查看消费进度，确保数据一致性。

2. 使用Debezium+Kafka Connect（实时同步）

适用于需要实时同步数据库变更或复杂数据流的场景：

1. 部署Docker环境：安装Docker并启动Zookeeper、Kafka、Kafka Connect等服务（参考搜索结果中的docker-compose.yaml示例）。
2. 配置Source Connector：根据数据源类型（如MySQL、PostgreSQL）创建Connector配置文件（如mysql-source.json），指定数据库连接信息和同步主题：

   {
   
     "name": "mysql-source",
     "config": {
   
       "connector.class": "io.debezium.connector.mysql.MySqlConnector",
       "tasks.max": "1",
       "database.hostname": "mysql-host",
       "database.port": "3306",
       "database.user": "user",
       "database.password": "password",
       "database.server.id": "184054",
       "database.server.name": "mysql-server",
       "table.include.list": "db.table1,db.table2",
       "topic.prefix": "mysql-"
     }
   
   }
       
   

3. 部署Sink Connector：配置Sink Connector将Kafka数据写入目标数据库（如PostgreSQL），类似Source Connector的配置方式。
4. 启动同步：通过Debezium UI或API启动Connector，实时同步数据。

四、数据验证与切换

1. 数据一致性检查：使用kafka-console-consumer.sh从源集群和目标集群消费相同主题的数据，对比内容是否一致：

   # 源集群消费
   kafka-console-consumer.sh --bootstrap-server source:9092 --topic test_topic --from-beginning | tee source-data.txt
   # 目标集群消费
   kafka-console-consumer.sh --bootstrap-server target:9092 --topic test_topic --from-beginning | tee target-data.txt
   diff source-data.txt target-data.txt  # 对比差异
   

2. 客户端切换：确认数据一致后，更新生产者和消费者的bootstrap.servers配置，指向目标集群地址，重启客户端应用。
3. 监控业务运行：迁移后持续监控目标集群的性能指标（如吞吐量、延迟）和业务日志，确保无异常。

五、注意事项

• 数据备份：迁移前务必备份源集群的所有数据（可通过kafka-dump-log.sh工具导出日志段）。
• 版本兼容：确保源集群和目标集群的Kafka版本兼容（建议使用相同主版本）。
• 性能优化：根据数据量调整MirrorMaker或Debezium的批量大小、线程数等参数，避免性能瓶颈。
• 监控报警：设置监控系统（如Prometheus+Granafa）监控迁移过程中的关键指标，及时处理异常。

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！