如何使用 python 将数据从 Kafka 发送到 ElasticSearch

Question

我正在使用 Python 语言。我有 csv 文件，我需要转换为 json 并发送到 kafka，然后发送到 ElasticSearch。

我能够将 Csv 转换为 Json 并发送给 kafka 消费者。如何从 Kafka Consumer 获取数据到 ElasticSearch

Answer 1

您不需要编写任何 Python。 Kafka 本身提供了一个流集成 API，称为 Kafka Connect。要使用它，您只需使用 JSON 文件对其进行配置。

• 使用https://github.com/jcustenborder/kafka-connect-spooldir 将 CSV 文件提取到 Kafka 中
  • 使用示例：https://www.confluent.io/blog/ksql-in-action-enriching-csv-events-with-data-from-rdbms-into-AWS/
• 将数据从 Kafka 主题流式传输到 Elasticsearch https://www.confluent.io/connector/kafka-connect-elasticsearch/
  • 使用示例：https://www.confluent.io/blog/blogthe-simplest-useful-kafka-connect-data-pipeline-in-the-world-or-thereabouts-part-2

免责声明：我为 Confluent 工作，并撰写了上述文章。

Answer 2

您可以使用带有 Kafka 输入插件的 Logstash 将数据发送到 Elasticsearch。它易于配置和运行。 https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html