apache-kafka - Kafka生产者读取数据文件

Question

我正在尝试在循环中加载数据文件（以检查统计信息）而不是 Kafka 中的标准输入。下载 Kafka 后，我执行了以下步骤：

启动动物园管理员：

bin/zookeeper-server-start.sh config/zookeeper.properties

启动服务器：

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

创建了一个名为“test”的主题：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

运行制片人：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 
Test1
Test2

消费者聆听：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
Test1
Test2

我想将一个数据文件传递给消费者可以直接看到的生产者，而不是标准输入。或者是否有任何可以读取数据文件的 kafka 生产者而不是控制台消费者。任何帮助将不胜感激。谢谢！

score 7 · Accepted Answer

您可以通过 cat 读取数据文件并将其通过管道传输到 kafka-console-producer.sh。

cat ${datafile} | ${kafka_home}/bin/kafka-console-producer.sh --broker-list ${brokerlist} --topic test

score 4 · Accepted Answer

如果总是有一个文件，您可以使用 tail 命令，然后将其通过管道传输到 kafka 控制台生产者。

但是如果在满足某些条件时会创建一个新文件，您可能需要使用 apache.commons.io.monitor 来监控创建的新文件，然后重复上述操作。

score 2 · Accepted Answer

Kafka 有这个内置的文件流连接器，用于将文件的内容通过管道传输到生产者（文件源），或将文件内容定向到另一个目的地（文件接收器）。

我们必须bin/connect-standalone.sh从可以在config/connect-file-source.properties和中配置的文件中读取config/connect-standalone.properties。

所以命令将是：

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

score 1 · Accepted Answer

如果您使用的是 Linux 或 Mac，最简单的方法是：

kafka-console-producer --broker-list localhost:9092 --topic test < messages.txt

score 0 · Accepted Answer

kafka-console-produce.sh \
  --broker-list localhost:9092 \
  --topic my_topic \
  --new-producer < my_file.txt

score 0 · Accepted Answer

下面的命令当然是最简单的方法。

kafka-console-producer --broker-list localhost:9092 --topic test < message.txt

但有时它无法找到该文件。例子：

C:\kafka_2.11-2.4.0\bin\windows>kafka-console-producer.bat --broker-list localhost:9092 --topic jason-input < C:\data\message.txt

您给出了实际路径，但它无法在当前位置找到 C，因此它会给出错误：找不到文件。我们会认为我们已经给出了实际路径，所以它会转到根目录，它会从那里开始路径，但它会在当前位置找到 C(root)。

解决方案是将 ..\ 放入移动到父文件夹的路径中。例如。你正在执行命令

C:\kafka_2.11-2.4.0\bin\windows>kafka-console-producer.bat --broker-list localhost:9092 --topic jason-input < ..\..\..\data\message.txt

截至目前，我进入了 windows 文件夹。..\ 将当前目录移动到 bin 文件夹，再次 ..\ 将当前目录移动到 kafka.. 文件夹，再次 ..\ 将移动到 C:。所以现在我的道路开始了。数据，然后是 message.txt

score 0 · Accepted Answer

您也可以尝试使用kafkacat实用程序。Github 上的自述文件提供了示例

如果您能分享哪种工具最适合您，那就太好了:)

来自 KafkaCat 自述文件的详细信息：

从标准输入读取消息，使用快速压缩生成“系统日志”主题

$ tail -f /var/log/syslog | kafkacat -b mybroker -t syslog -z snappy

7 回答 7