如何将嗅探到的数据包连续提供给kafka？

Question

目前我正在从本地 wlan 接口嗅探数据包，例如：

sudo tshark > sampleData.pcap

但是，我需要将此数据提供给 kafka。

目前，我有一个kafka生产者脚本producer.sh：

../bin/kafka-console-producer.sh --broker-list localhost:9092 --topic 'spark-kafka'

并像这样向 kafka 提供数据：

producer.sh < sampleData.pcap

在 sampleData.pcap 中我有预先捕获的 IP 数据包信息。

但是，我想将流程自动化，如下所示：

sudo tshark > http://localhost:9091
producer.sh < http://localhost:9091

这显然只是一个伪算法。我想要做的是，将嗅探数据发送到一个端口并让 kafka 不断读取它。我不希望 kafka 连续读取文件，因为这意味着来自单个文件的大量读/写操作会导致效率低下。

我在互联网上搜索并发现了 kafka-connect，但我找不到任何有用的文档来实现这样的东西。

实现这样的事情的最佳方式是什么？

谢谢！

Answer 1

使用 netcat

无需编写服务器，您可以使用 netcat（并告诉您的脚本监听标准输入）：

shell1> nc -l 8888 | ./producer.sh
shell2> sudo tshark -l | nc 127.1 8888

tshark 的-l 防止它过多地缓冲输出（在每个数据包后刷新）。

---

使用命名管道

您还可以使用命名管道将 tshark 输出传输到您的第二个进程：

shell1> mkfifo /tmp/tsharkpipe
shell1> tail -f -c +0 /tmp/tsharkpipe | ./producer.sh
shell2> sudo tshark -l > /tmp/tsharkpipe

Answer 2

如果你使用 Node，你可以使用 child_process 和 kafka_node 来做。像这样的：

var kafka = require('kafka-node');
var client = new kafka.Client('localhost:2181');
var producer = new kafka.Producer(client);

var spawn = require('child_process').spawn;
var tshark = spawn('sudo', ['/usr/sbin/tshark']);

tshark.stdout.on('data', (data) => {
  producer.send([
    {topic: 'spark-kafka', messages: [data.split("\n")]}
  ], (err,result) => { console.log("sent to kafka")});
});

Answer 3

我觉得你也可以

• 创建一个连接到 kafka 的微型服务器 ant 监听一个端口
• 使用 kafka-file 连接器并将所有数据附加到该文件。 http://kafka.apache.org/documentation.html#quickstart_kafkaconnect

Answer 4

另一种选择是使用 Apache NiFi。使用 NiFi，您可以执行命令并将输出传递给其他块以进行进一步处理。在这里，您可以让 NiFi 在本地主机上执行 tshark 命令，然后将输出传递给 Kafka。

有一个示例here 应该更详细地演示这种方法。