Kafka 数据迁移在 Debian 上的步骤

Kafka数据迁移在Debian上的步骤

一、准备工作

安装JDK：Kafka依赖Java运行环境，需在Debian上安装OpenJDK（推荐8或11）。执行以下命令：
```
sudo apt update && sudo apt install -y openjdk-8-jdk
```

下载并解压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/

配置环境变量：编辑/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工具：

添加新Broker：编辑server.properties文件，为新Broker分配唯一ID（如broker.id=3），并配置listeners、log.dirs等参数，启动新Broker。

生成重新分配计划：创建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

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

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

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

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

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

1. 使用MirrorMaker（Kafka自带）

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

配置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=.*

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

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

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

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

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

部署Docker环境：安装Docker并启动Zookeeper、Kafka、Kafka Connect等服务（参考搜索结果中的docker-compose.yaml示例）。

配置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-"
  }
}

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

四、数据验证与切换

数据一致性检查：使用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  # 对比差异

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

五、注意事项

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