前言：Kafka的备份的单元是partition，也就是每个partition都都会有leader partiton和follow partiton。其中leader partition是用来进行和producer进行写交互，follow从leader副本进行拉数据进行同步，从而保证数据的冗余，防止数据丢失的目的。如图:

Kafka 数据备份类型：

• 全量备份
• 增量备份

全量备份是将整个 Kafka 的数据复制到一个不同的地方。
增量备份是在全量备份后仅仅备份增量的数据。

一、全量备份

# 指定备份的主题
BACKUP_TOPIC=test #指定备份的数据目录
BACKUP_DIR=/tmp/backup #创建备份目录
mkdir -p $BACKUP_DIR #备份主题数据
kafka-console-consumer.sh --bootstrap-server localhost:9092  --topic $BACKUP_TOPIC --from-beginning > $BACKUP_DIR/$BACKUP_TOPIC.txt。

以下是该命令的详细解释：

kafka-console-consumer.sh：这是 Kafka 提供的命令行消费者工具的可执行文件路径。
--bootstrap-server localhost:9092：指定 Kafka 的引导服务器地址和端口，这里是 localhost 的 9092 端口。
--topic $BACKUP_TOPIC：指定要消费的 Kafka 主题，这里是变量 $BACKUP_TOPIC 指定的主题。
--from-beginning：指定从主题的起始位置开始消费消息，而不是从最新的消息开始。
>：这是一个重定向符号，将输出导入到指定的文件中。
$BACKUP_DIR/$BACKUP_TOPIC.txt：指定输出文件的路径和名称，这里是变量 $BACKUP_DIR 指定的目录下创建名为 $BACKUP_TOPIC.txt 的文件。
通过执行这个命令，Kafka 将从指定的主题中读取消息，并将它们写入到指定的文本文件中，从而实现备份的目的。即使用kafka-console-consumer.sh，工具将主题$BACKUP_TOPIC的数据备份到$BACKUP_DIR目录下的$BACKUP_TOPIC.txt文件中。

二、增量备份

增量备份需要借助第三方工具，例如 Kafka 的 MirrorMaker 等实现 。
下面是 MirrorMaker 的用法示例：

# 指定源和目的地址
SOURCE_HOST=localhost:9092
DESTINATION_HOST=backup-host:9092。
# 创建 MirrorMaker 配置文件
cat > /tmp/mirror-maker.properties <<EOF
consumer.bootstrap.servers=$SOURCE_HOST
producer.bootstrap.servers=$DESTINATION_HOST
EOF。
# 运行 MirrorMaker
kafka-run-class.sh kafka.tools.MirrorMaker \
  --consumer.config /tmp/mirror-maker.properties \
--producer.config /tmp/mirror-maker.properties \
--whitelist $BACKUP_TOPIC。

上述代码中创建一个 MirrorMaker 配置文件将源端的数据同步到目标端–whitelist参数指定备份的主题。

下面介绍Kafka数据恢复

三、全量恢复

# 指定恢复的主题
RESTORE_TOPIC=test 
# 指定备份文件路径
BACKUP_FILE=/tmp/backup/$RESTORE_TOPIC.txt。
# 恢复主题数据
kafka-console-producer.sh \
--broker-list localhost:9092 \
--topic $RESTORE_TOPIC  \
--new-producer \
< $BACKUP_FILE

上述代码将$BACKUP_FILE文件中的数据恢复到$RESTORE_TOPIC主题中。
注意：该脚本也是同步操作，恢复时间较长时建议使用异步操作。

kafka-console-producer.sh \
--broker-list localhost:9092 \
--topic $RESTORE_TOPIC  \
--new-producer  \
--async \
< $BACKUP_FILE。

四、增量恢复

增量恢复需要使用 MirrorMaker 来实现，下面是 MirrorMaker 的用法示例：
# 创建MirrorMaker 配置文件
cat > /tmp/mirror-maker.properties <<EOF
consumer.bootstrap.servers=backup-host:9092producer.bootstrap.servers=localhost:9092 
EOF
# 运行MirrorMaker
kafka-run-class.sh kafka.tools.MirrorMaker \
--consumer.config /tmp/mirror-maker.properties \
--producer.config /tmp/mirror-maker.properties \ 
--whitelist $RESTORE_TOPIC  （或者是*，如： --whitelist “.*”)

上述代码中创建一个 MirrorMaker 配置文件将备份端的数据同步到目标端$RESTORE_TOPIC主题中。
注意：增量恢复会将备份端数据的变化同步到目标端，因此恢复时必须先将备份端数据同步完整。
实例：
consumer.properties文件：

#cat consumer.properties
bootstrap.servers=192.168.3.194:9092
group.id=mirrormaker
auto.offset.reset=earliest

producer.properties文件：
#cat producer.properties
bootstrap.servers=192.168.2.123:9092

运行mirror-maker
cat run-mirrormaker.sh
kafka-run-class.sh kafka.tools.MirrorMaker \
 --consumer.config ./consumer.properties \
--producer.config ./producer.properties \
--offset.commit.interval.ms 5000 --num.streams 2 \
--whitelist "test0428"

参数说明：
–consumer.config 消费者的配置文件（要消费的集群）
–producer.config 指定生产配置文件（要发送到的目标集群）
–whitelist 要同步的topic白名单，可以匹配正则，也可以指定具体topic
–offset.commit.interval.ms 消费端提交offset时间间隔
–num.streams MirrorMaker 要创建多少个 KafkaConsumer 实例

运行后会提示:
WARNING: The default partition assignment strategy of the mirror maker will change from ‘range’ to ‘roundrobin’ in an upcoming release (so that better load balancing can be achieved). If you prefer to make this switch in advance of that release add the following to the corresponding config: ‘partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor’
主要是说MirrorMaker使用的消费策略是 Range ，以后可能改成 “轮训策略” ，我们可以手动指定“轮询策略”。

在Consuemr.properties中设置：

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

此时我们已经运行成功，直接去查看是否可以消费成功即可。
注意：同步前最后将目标集群的Topic创建好，否则会使用Broker默认配置。
使用一个脚本来做备份和恢复：
简单的脚本，用于备份和恢复kafka数据：

#!/bin/bash
function backup_topic() {
    local topic=$1
    local backup_dir=$2

    echo "Starting backup for topic: $topic"

    mkdir -p $backup_dir
    kafka-console-consumer.sh \
        --bootstrap-server localhost:9092 \
        --topic $topic \
        --from-beginning \
        > $backup_dir/$topic.txt

    echo "Backup completed for topic: $topic"
}
function restore_topic() {
    local topic=$1
    local backup_file=$2

    echo "Starting restore for topic: $topic"
    kafka-console-producer.sh \
        --broker-list localhost:9092 \
        --topic $topic \
        --new-producer \
        < $backup_file
    echo "Restore completed for topic: $topic"
}
backup_topic example-topic /tmp/backup
restore_topic example-topic /tmp/backup/example-topic.txt

上述代码中定义了两个函数 backup_topic 和 restore_topic，分别用于备份和恢复 Kafka主题数据。
在这个脚本中备份的主题是 example-topic，备份数据存储的目录是 /tmp/backup。
要恢复数据，请调用 restore_topic 函数，并通过参数指定要恢复的主题和备份文件的路径。在脚本的最后示例恢复了 example-topic 主题的备份数据。