117笔记问答在Debian上使用Apache Kafka进行实时数据处理涉及多个步骤，包括安装Kafka、配置Kafka集群、启动Kafka服务以及编写和运行数据处理应用程序。以下是一个详细的指南：
1. 安装Java
Kafka依赖于Java运行时环境（JRE），因此首先需要安装Java。
sudo apt update sudo apt install openjdk-11-jdk
验证Java安装：
java -version
2. 下载并安装Kafka
从Apache Kafka官方网站下载最新版本的Kafka。
wget https://downloads.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz tar -xzf kafka_2.13-3.2.0.tgz cd kafka_2.13-3.2.0
3. 配置Kafka
Kafka的配置文件位于config目录下。主要需要配置以下几个文件：
server.properties: Kafka服务器的主要配置文件。zookeeper.properties: Zookeeper的配置文件。
配置Zookeeper
编辑config/zookeeper.properties文件：
dataDir=/var/lib/zookeeper clientPort=2181 maxClientCnxns=0
创建Zookeeper数据目录并启动Zookeeper：
sudo mkdir -p /var/lib/zookeeper sudo chown -R $(whoami):$(whoami) /var/lib/zookeeper bin/zookeeper-server-start.sh config/zookeeper.properties
配置Kafka
编辑config/server.properties文件：
broker.id=1 listeners=PLAINTEXT://:9092 log.dirs=/var/lib/kafka-logs zookeeper.connect=localhost:2181
创建Kafka日志目录并启动Kafka服务器：
sudo mkdir -p /var/lib/kafka-logs sudo chown -R $(whoami):$(whoami) /var/lib/kafka-logs bin/kafka-server-start.sh config/server.properties
4. 创建Topic
创建一个Topic用于数据传输。
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1
5. 编写数据处理应用程序
可以使用Kafka Streams或Kafka Connect进行实时数据处理。以下是一个简单的Kafka Streams示例。
添加依赖
在项目的pom.xml中添加Kafka Streams依赖：
org.apache.kafka kafka-streams 3.2.0
编写Kafka Streams应用程序
创建一个Java类来编写Kafka Streams应用程序：
import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; public class KafkaStreamsExample { public static void main(String[] args) { StreamsBuilder builder = new StreamsBuilder(); KStream sourceStream = builder.stream("test-topic"); sourceStream.mapValues(value -> value.toUpperCase()).to("processed-topic"); KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig()); streams.start(); Runtime.getRuntime().addShutdownHook(new Thread(streams::close)); } private static Properties getStreamsConfig() { Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-example"); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); return props; } } 编译并运行应用程序
使用Maven编译并运行应用程序：
mvn clean package java -cp target/kafka-streams-example-1.0-SNAPSHOT.jar KafkaStreamsExample
6. 验证数据处理
生产一些数据到test-topic并验证处理后的数据是否出现在processed-topic。
bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092
输入一些文本并按回车键发送消息。
bin/kafka-console-consumer.sh --topic processed-topic --from-beginning --bootstrap-server localhost:9092
你应该能看到处理后的消息。
通过以上步骤，你可以在Debian上成功安装和配置Kafka，并使用Kafka Streams进行实时数据处理。