Kafka生产者读取数据文件答案

【问题标题】：Kafka producer to read data filesKafka生产者读取数据文件
【发布时间】：2016-05-24 12:04:13
【问题描述】：

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

启动 zookeeper：

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 生产者而不是控制台消费者。任何帮助将不胜感激。谢谢！

【问题讨论】：

【解决方案1】：

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

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

【讨论】：

【解决方案2】：

如果总是有一个文件，你可以只使用tail命令，然后将它传递给kafka控制台生产者。

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

【讨论】：

【解决方案3】：

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

我们有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

【讨论】：

能否举出config/connect-file-source.properties和config/connect-standalone.properties的内容示例
这里有更详细的解释bigdatums.net/2017/06/20/writing-file-content-to-kafka-topic

【解决方案4】：

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

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

【讨论】：

我正在尝试这个答案，但它给出了错误：没有找到文件然后我试图给出 C:\data\messages.txt 之类的实际路径，但出现了同样的错误。然后我在路径中尝试了 ..\ ，这意味着父文件夹，但在那里我感到困惑，所以我在那里使用选项卡来查看那里的文件。 Hurrrrraaaayyyyyy ！有效。它无法找到该文件，因为它正在同一位置进行搜索。即我已经给出了路径 c:\data\message.txt 它试图将 c 搜索到当前位置。所以我需要使用父文件夹commant ..\ 来移动它

【解决方案5】：

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

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

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

从标准输入读取消息，通过快速压缩生成“syslog”主题

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

【讨论】：

【解决方案6】：

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

【讨论】：

【解决方案7】：

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

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

【讨论】：