系列 | kafka 知其所以然【二】快速入门

大数据技能圈 2023-10-10

kafka 快速入门

获取Kafka

下载最新的 Kafka 版本并解压：

$ tar -xzf kafka_2.13-3.5.0.tgz
$ cd kafka_2.13-3.5.0

启动KAFKA环境

注意：您的本地环境必须安装 Java 8+。

Kafka与ZooKeeper

运行以下命令以便以正确的顺序启动所有服务：

# Start the ZooKeeper service
$ bin/zookeeper-server-start.sh config/zookeeper.properties

打开另一个终端会话并运行：

# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties

所有服务成功启动后，您将拥有一个正在运行并可供使用的基本 Kafka 环境。

Kafka与 Kraft

生成集群 UUID

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

设置日志目录格式

$ bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

启动Kafka服务器

$ bin/kafka-server-start.sh config/kraft/server.properties

一旦 Kafka 服务器成功启动，您将拥有一个正在运行并可供使用的基本 Kafka 环境。

创建一个主题来存储您的事件

在编写第一个事件之前，您必须创建一个主题。打开另一个终端会话并运行：

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

所有 Kafka 的命令行工具都有附加选项：运行kafka-topics.sh不带任何参数的命令来显示使用信息。例如，它还可以向您显示新主题的分区计数等详细信息：

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events        TopicId: NPmZHyhbR9y00wMglMH2sg PartitionCount: 1       ReplicationFactor: 1  Configs:
    Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

将一些事件写入主题

运行控制台生产者客户端以将一些事件写入您的主题。默认情况下，您输入的每一行都会导致将一个单独的事件写入主题。

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event

Ctrl-C您可以随时停止生产者客户端。

阅读事件

打开另一个终端会话并运行控制台消费者客户端来读取您刚刚创建的事件：

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

Ctrl-C您可以随时停止消费者客户端。

由于事件持久存储在 Kafka 中，因此它们可以被任意数量的消费者读取任意多次。您可以通过打开另一个终端会话并再次重新运行之前的命令来轻松验证这一点。

使用 KAFKA CONNECT 将数据作为事件流导入/导出

首先，确保添加connect-file-3.5.0.jar到plugin.pathConnect 工作线程配置中的属性。出于本快速入门的目的，我们将使用相对路径并将连接器的包视为 uber jar，它在从安装目录运行快速入门命令时起作用。但是，值得注意的是，对于生产部署，使用绝对路径始终是更好的选择。有关如何设置此配置的详细说明，请参阅plugin.path 。

编辑config/connect-standalone.properties文件，添加或更改与plugin.path以下内容匹配的配置属性，然后保存文件：

> echo "plugin.path=libs/connect-file-3.5.0.jar"

然后，首先创建一些种子数据进行测试：

> echo -e "foo\nbar" > test.txt

或者在 Windows 上：

> echo foo> test.txt
> echo bar>> test.txt

接下来，我们将启动两个以独立模式运行的连接器，这意味着它们在单个本地专用进程中运行。我们提供三个配置文件作为参数。第一个始终是 Kafka Connect 进程的配置，包含常见配置，例如要连接的 Kafka 代理和数据的序列化格式。其余配置文件各自指定要创建的连接器。这些文件包括唯一的连接器名称、要实例化的连接器类以及连接器所需的任何其他配置。

> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

这些示例配置文件包含在 Kafka 中，使用您之前启动的默认本地集群配置并创建两个连接器：第一个是源连接器，用于从输入文件读取行并将每行生成到 Kafka 主题，第二个是接收器连接器从 Kafka 主题读取消息并将每条消息生成为输出文件中的一行。

在启动过程中，您将看到许多日志消息，其中包括一些指示连接器正在实例化的日志消息。一旦 Kafka Connect 进程启动，源连接器应开始从test.txt主题读取行并将其生成到主题connect-test，接收器连接器应开始从主题读取消息connect-test 并将其写入文件test.sink.txt。我们可以通过检查输出文件的内容来验证数据是否已通过整个管道传递：

> more test.sink.txt
foo
bar

请注意，数据存储在 Kafka topic 中connect-test，因此我们还可以运行控制台消费者来查看主题中的数据（或使用自定义消费者代码来处理它）：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
...

连接器继续处理数据，因此我们可以将数据添加到文件并查看它在管道中移动：

> echo Another line>> test.txt

您应该看到该行出现在控制台使用者输出和接收器文件中。

使用 KAFKA STREAMS 处理您的事件

将数据作为事件存储在 Kafka 中后，您可以使用适用于 Java/Scala 的Kafka Streams客户端库处理数据。它允许您实现关键任务实时应用程序和微服务，其中输入和/或输出数据存储在 Kafka 主题中。Kafka Streams 将客户端编写和部署标准 Java 和 Scala 应用程序的简单性与 Kafka 服务器端集群技术的优点相结合，使这些应用程序具有高度可扩展性、弹性、容错性和分布式性。该库支持一次性处理、有状态操作和聚合、窗口、连接、基于事件时间的处理等等。

为了让您初步体验一下，以下是如何实现流行的WordCount算法：

KStream<String, String> textLines = builder.stream("quickstart-events");


KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((keyIgnored, word) -> word)
            .count();


wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

终止KAFKA环境

现在您已经完成了快速入门，可以随意拆除 Kafka 环境，或者继续尝试。

Ctrl-C如果您还没有这样做，请使用停止生产者和消费者客户端。
使用停止 Kafka 代理Ctrl-C。
最后，如果遵循 Kafka with ZooKeeper 部分，请使用停止 ZooKeeper 服务器Ctrl-C。

如果您还想删除本地 Kafka 环境的任何数据，包括您在此过程中创建的任何事件，请运行以下命令：

$ rm -rf /tmp/kafka-logs /tmp/zookeeper /tmp/kraft-combined-logs

恭喜！

您已成功完成 Apache Kafka 快速入门。

文章转载自大数据技能圈，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

系列 | kafka 知其所以然【二】快速入门

评论