如何通过 Kafka 将数据导入 Elasticsearch

作者：来自 Elastic Andre Luiz

将 Apache Kafka 与 Elasticsearch 集成的分步指南，以便使用 Python、Docker Compose 和 Kafka Connect 实现高效的数据提取、索引和可视化。

在本文中，我们将展示如何将 Apache Kafka 与 Elasticsearch 集成以进行数据提取和索引。我们将概述 Kafka、其生产者（producers）和消费者（consumers）的概念，并创建一个日志索引，其中将通过 Apache Kafka 接收和索引消息。该项目以 Python 实现，代码可在 GitHub 上找到。

先决条件

Docker 和 Docker Compose：确保你的机器上安装了 Docker 和 Docker Compose。
Python 3.x：运行生产者和消费者脚本。

Apache Kafka 简介

Apache Kafka 是一个分布式流媒体平台，具有高可扩展性和可用性以及容错能力。在 Kafka 中，数据管理通过主要组件进行：

Broker/代理：负责在生产者和消费者之间存储和分发消息。
Zookeeper：管理和协调 Kafka 代理，控制集群的状态、分区领导者和消费者信息。
Topics/主题：发布和存储数据以供使用的渠道。
Consumers 及 Producers/消费者和生产者：生产者向主题发送数据，而消费者则检索该数据。

这些组件共同构成了 Kafka 生态系统，为数据流提供了强大的框架。

项目结构

为了理解数据提取过程，我们将其分为几个阶段：

基础设施配置/Infrastructure Provisioning：设置 Docker 环境以支持 Kafka、Elasticsearch 和 Kibana。
创建生产者/Producer Creation：实现 Kafka 生产者，将数据发送到日志主题。
创建消费者/Consumer Creation：开发 Kafka 消费者以读取和索引 Elasticsearch 中的消息。
提取验证/Ingestion Validation：验证和确认已发送和已使用的数据。

使用 Docker Compose 进行基础设施配置

我们利用 Docker Compose 来配置和管理必要的服务。下面，你将找到 Docker Compose 代码，它设置了 Apache Kafka、Elasticsearch 和 Kibana 集成所需的每项服务，确保数据提取过程。

docker-compose.yml

version: "3"services:zookeeper:image: confluentinc/cp-zookeeper:latestcontainer_name: zookeeperenvironment:ZOOKEEPER_CLIENT_PORT: 2181kafka:image: confluentinc/cp-kafka:latestcontainer_name: kafkadepends_on:- zookeeperports:- "9092:9092"- "9094:9094"environment:KAFKA_BROKER_ID: 1KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST:${HOST_IP}:9092KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXTKAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1elasticsearch:image: docker.elastic.co/elasticsearch/elasticsearch:8.15.1container_name: elasticsearch-8.15.1environment:- node.name=elasticsearch- xpack.security.enabled=false- discovery.type=single-node- "ES_JAVA_OPTS=-Xms512m -Xmx512m"volumes:- ./elasticsearch:/usr/share/elasticsearch/dataports:- 9200:9200kibana:image: docker.elastic.co/kibana/kibana:8.15.1container_name: kibana-8.15.1ports:- 5601:5601environment:ELASTICSEARCH_URL: http://elasticsearch:9200ELASTICSEARCH_HOSTS: '["http://elasticsearch:9200"]'

你可以直接从 Elasticsearch Labs GitHub repo 访问该文件。

使用 Kafka 生产器发送数据

生产器负责将消息发送到日志主题。通过批量发送消息，可以提高网络使用效率，允许使用 batch_size 和 linger_ms 设置进行优化，这两个设置分别控制批次的数量和延迟。配置 acks='all' 可确保消息持久存储，这对于重要的日志数据至关重要。

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],  # Specifies the Kafka server to connectvalue_serializer=lambda x: json.dumps(x).encode('utf-8'),  # Serializes data as JSON and encodes it to UTF-8 before sendingbatch_size=16384,     # Sets the maximum batch size in bytes (here, 16 KB) for buffered messages before sendinglinger_ms=10,         # Sets the maximum delay (in milliseconds) before sending the batchacks='all'            # Specifies acknowledgment level; 'all' ensures message durability by waiting for all replicas to acknowledge
)def generate_log_message():levels = ["INFO", "WARNING", "ERROR", "DEBUG"]messages = ["User login successful","User login failed","Database connection established","Database connection failed","Service started","Service stopped","Payment processed","Payment failed"]log_entry = {"level": random.choice(levels),"message": random.choice(messages),"timestamp": time.time()}return log_entrydef send_log_batches(topic, num_batches=5, batch_size=10):for i in range(num_batches):logger.info(f"Sending batch {i + 1}/{num_batches}")for _ in range(batch_size):log_message = generate_log_message()producer.send(topic, value=log_message)producer.flush()if __name__ == "__main__":topic = "logs"send_log_batches(topic)producer.close()

当启动 producer 的时候，会批量的向 topic 发送消息，如下图：

INFO:kafka.conn:Set configuration …
INFO:log_producer:Sending batch 1/5 
INFO:log_producer:Sending batch 2/5
INFO:log_producer:Sending batch 3/5
INFO:log_producer:Sending batch 4/5

使用 Kafka Consumer 消费和索引数据

Consumer 旨在高效处理消息，消费来自日志主题的批次并将其索引到 Elasticsearch 中。使用 auto_offset_reset='latest'，可确保 Consumer 开始处理最新消息，忽略较旧的消息，max_poll_records=10 将批次限制为 10 条消息。使用 fetch_max_wait_ms=2000，Consumer 最多等待 2 秒以积累足够的消息，然后再处理批次。

在其主循环中，Consumer 消费日志消息、处理并将每个批次索引到 Elasticsearch 中，确保持续的数据摄取。

consumer = KafkaConsumer('logs',                               bootstrap_servers=['localhost:9092'],auto_offset_reset='latest',            # Ensures reading from the latest offset if the group has no offset storedenable_auto_commit=True,               # Automatically commits the offset after processinggroup_id='log_consumer_group',         # Specifies the consumer group to manage offset trackingmax_poll_records=10,                   # Maximum number of messages per batchfetch_max_wait_ms=2000                 # Maximum wait time to form a batch (in ms)
)def create_bulk_actions(logs):for log in logs:yield {"_index": "logs","_source": {'level': log['level'],'message': log['message'],'timestamp': log['timestamp']}}if __name__ == "__main__":try:print("Starting message processing…")while True:messages = consumer.poll(timeout_ms=1000)  # Poll receive messages# process each batch messagesfor _, records in messages.items():logs = [json.loads(record.value) for record in records]bulk_actions = create_bulk_actions(logs)response = helpers.bulk(es, bulk_actions)print(f"Indexed {response[0]} logs.")except Exception as e:print(f"Erro: {e}")finally:consumer.close()print(f"Finish")

在 Kibana 中可视化数据

借助 Kibana，我们可以探索和验证从 Kafka 提取并在 Elasticsearch 中编入索引的数据。通过访问 Kibana 中的开发工具，你可以查看已编入索引的消息并确认数据符合预期。例如，如果我们的 Kafka 生产者发送了 5 个批次，每个批次 10 条消息，我们应该在索引中看到总共 50 条记录。

要验证数据，你可以在 Dev Tools 部分使用以下查询：

GET /logs/_search
{"query": {"match_all": {}}
}

相应：

此外，Kibana 还提供了创建可视化和仪表板的功能，可帮助使分析更加直观和具有交互性。下面，你可以看到我们创建的一些仪表板和可视化示例，它们以各种格式展示了数据，增强了我们对所处理信息的理解。

使用 Kafka Connect 进行数据提取

Kafka Connect 是一种旨在促进数据源和目标（接收器）之间的集成的服务，例如数据库或文件系统。它使用预定义的连接器来自动处理数据移动。在我们的例子中，Elasticsearch 充当数据接收器。

使用 Kafka Connect，我们可以简化数据提取过程，无需手动将数据提取工作流实施到 Elasticsearch 中。借助适当的连接器，Kafka Connect 允许将发送到 Kafka 主题的数据直接在 Elasticsearch 中编入索引，只需进行最少的设置，无需额外编码。

使用 Kafka Connect

要实现 Kafka Connect，我们将 kafka-connect 服务添加到我们的 Docker Compose 设置中。此配置的一个关键部分是安装 Elasticsearch 连接器，它将处理数据索引。

配置服务并创建 Kafka Connect 容器后，将需要一个 Elasticsearch 连接器的配置文件。此文件定义基本参数，例如：

connection.url：Elasticsearch 的连接 URL。
topics：连接器将监视的 Kafka 主题（在本例中为 “logs”）。
type.name：Elasticsearch 中的文档类型（通常为 _doc）。
value.converter：将 Kafka 消息转换为 JSON 格式。
value.converter.schemas.enable：指定是否应包含架构。
schema.ignore 和 key.ignore：在索引期间忽略 Kafka 架构和键的设置。

以下是在 Kafka Connect 中创建 Elasticsearch 连接器的 curl 命令：

curl --location '{{url}}/connectors' \
--header 'Content-Type: application/json' \
--data '{"name": "elasticsearch-sink-connector","config": {"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","topics": "logs","connection.url": "http://elasticsearch:9200","type.name": "_doc","value.converter": "org.apache.kafka.connect.json.JsonConverter","value.converter.schemas.enable": "false","schema.ignore": "true","key.ignore": "true"}
}'

通过此配置，Kafka Connect 将自动开始提取发送到 “logs” 主题的数据并在 Elasticsearch 中对其进行索引。这种方法允许完全自动化的数据提取和索引，而无需额外的编码，从而简化整个集成过程。