centos kafka数据迁移

CentOS环境下Kafka数据迁移常见方法及步骤

一、存储目录迁移（适用于单集群扩容/磁盘更换）

当需要迁移Kafka的数据存储目录（log.dirs）或日志目录（logs）时，需通过以下步骤完成：

1. 关闭Kafka服务：在所有Broker节点上执行./kafka-server-stop.sh停止服务，避免迁移过程中数据损坏。
2. 迁移数据文件：将原数据目录（如/usr/local/kafka/data）和日志目录（如/usr/local/kafka/logs）的内容复制到新位置（如/mnt/kafka/data、/mnt/kafka/logs），使用cp -r命令确保文件完整性。
3. 修改配置文件：
   • 编辑server.properties，更新log.dirs参数为新路径（如log.dirs=/mnt/kafka/data）；
   • 修改kafka-run-class.sh（或kafka-server-start.sh），将日志输出目录指向新位置（如LOG_DIR=/mnt/kafka/logs）。
4. 启动Kafka服务：在所有节点上执行./kafka-server-start.sh -daemon server.properties，启动后通过kafka-topics.sh --list --bootstrap-server localhost:9092验证Topic是否正常。

二、使用MirrorMaker跨集群迁移（适用于灾备/多活/云迁移）

MirrorMaker是Kafka官方提供的跨集群数据同步工具，支持全量/增量迁移，适用于将数据从一个集群复制到另一个集群：

1. 准备配置文件：创建mirror-maker.properties，配置源集群和目标集群的连接信息及同步Topic（示例）：

   # 源集群配置
   consumer.bootstrap.servers=source-host:9092
   # 目标集群配置
   producer.bootstrap.servers=target-host:9092
   # 同步的Topic（支持正则，如test.*）
   topics=test
   # 消费者组ID
   group.id=mirror-maker-group
   

2. 启动MirrorMaker：执行./kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config mirror-maker.properties --producer.config mirror-maker.properties --whitelist test（--whitelist指定Topic，支持*通配符）。
3. 验证同步结果：在目标集群上使用kafka-console-consumer.sh消费同步后的Topic，确认数据与源集群一致。

三、使用kafka-reassign-partitions迁移分区（适用于集群内部调整）

当需要将Topic的分区从一个Broker迁移到另一个Broker（如集群扩容后平衡负载），可使用kafka-reassign-partitions.sh工具：

1. 生成迁移计划：创建topics-to-move.json（指定要迁移的Topic），执行./kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2,3" --generate，生成包含当前分区分配和建议新分配的JSON文件。
2. 执行迁移：使用生成的建议计划文件（reassignment.json），执行./kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file reassignment.json --execute。
3. 验证迁移：再次执行./kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file reassignment.json --verify，检查分区是否迁移成功（状态为completed）。

四、使用Kafka Linking零停机迁移（适用于AutoMQ集群）

若目标集群为AutoMQ，可使用Kafka Linking工具实现零停机迁移，确保业务不中断：

1. 创建Kafka Link：在新集群上执行automq cluster create kafka link <link-name> <source-topic> <target-topic>，建立源Topic到目标Topic的镜像链路。
2. 镜像Topic：启动Link后，AutoMQ会自动同步源集群的Topic数据到目标集群。
3. 迁移消费组：通过consumer groups promote命令将消费组从源集群切换到目标集群，消费者开始从新集群消费。
4. 停止同步并推广Topic：确认消费组迁移完成后，执行promote topic停止同步，并将目标Topic设为生产可用。

五、使用CloudCanal同步数据（适用于异构数据源）

CloudCanal是一款数据集成工具，支持Kafka与其他数据源（如MySQL、Elasticsearch）的双向同步，也可用于Kafka集群间迁移：

1. 配置数据源：登录CloudCanal控制台，添加源Kafka和目标Kafka的数据源（配置连接信息、权限）。
2. 创建同步任务：选择源Topic和目标Topic，配置同步参数（如消息格式为Avro/JSON、增量同步起始位置）。
3. 启动任务并监控：启动同步任务，通过CloudCanal的监控界面查看同步进度和数据一致性（如消息延迟、错误数）。

迁移注意事项

• 数据一致性：迁移前建议停止源集群的生产者（或使用MirrorMaker的--sync参数保证同步），避免数据丢失；迁移后验证目标集群的数据完整性（如消息数量、内容）。
• 性能监控：迁移过程中监控源集群和目标集群的性能（如CPU、磁盘IO、网络带宽），避免因迁移导致业务受影响。
• 备份策略：迁移前对源集群数据进行全量备份（如使用kafka-dump或kafka-console-consumer导出到文件），以便迁移失败时快速恢复。
• 客户端更新：迁移完成后，更新生产者和消费者的连接地址（如bootstrap.servers）指向目标集群，重启业务确保连接正常。