使用python连接kafka和elasticsearch

Question

我刚开始使用 Kafka 和 elasticsearch。我一直在尝试使用弹性搜索，但我遇到了一些问题。我已经长大了一个 docker compose 文件，其中包含构建环境所需的所有图像，然后使用 kafka 我已经将数据产品放入特定主题，然后我需要从 Kafka 的消费者数据中获取 pub/sub 系统将用于摄取的数据发送到 elasticsearch。 我使用 python 实现了所有这些。我已经看到在端口和localhost中出现了ip elasticsearch，而不是页面中的kibana出现了以下句子： kibana 服务器还没有准备好

消费者 python 类似于我从主题中获取数据的东西：

from kafka import KafkaConsumer
# Import sys module
import sys

# Import json module to serialize data
import json

# Initialize consumer variable and set property for JSON decode
consumer = KafkaConsumer ('JSONtopic',bootstrap_servers = ['localhost:9092'],
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
for message in consumer:
 print("Consumer records:\n")
 print(message)
 print("\nReading from JSON data\n")
 print("Name:",message[6]['name'])
 print("Email:",message[6]['email'])
 # Terminate the script
 sys.exit()

目标是使用 elasticsearch 进行分析，因此我需要将其用作后端，以便将数据可视化到 kibana 中。它也可以非常感谢一个教程来理解我应该做什么来链接这些信息。 （P.s. 数据从一个主题到另一个主题没有问题，但问题是获取这些信息并插入到弹性中，并有可能将这些信息可视化）

Answer 1

如果您要将数据从 Kafka 推送到 Elasticsearch，那么使用 Consumer API 执行此操作通常不是一个好主意，因为存在一些工具可以做得更好并处理更多功能。

例如：

• Kafka Connect（例如? https://rmoff.dev/kafka-elasticsearch-video）
• Lo​​gstash