使用 Kafka-Connect 将消息路由到不同的 ElasticSearchIndex答案

【问题标题】：Route message to different ElasticSearchIndex using Kafka-Connect使用 Kafka-Connect 将消息路由到不同的 ElasticSearchIndex
【发布时间】：2020-06-17 06:39:00
【问题描述】：

我有一个 kafka 主题，其中向其发布了多种类型的消息（都是 JSON），其标头属性为：- eventType = "abc" 或 eventType = "xyz" 或 eventType = "def" 等。

我的要求是根据在 kafka 中发送的 eventType 标头字段将这些消息从 Kafka 主题发送到不同的 elasticSearch 索引。例如如果 eventType = "abc"，将数据推送到 elasticsearch index = "abc123" if eventType = "xyz"，推送数据到elasticsearch index = "xyz123"

我打算使用 kafka connect 进行此操作。我需要帮助如何使用 kafka-connect 实现这一目标？

我正在使用 kafka 1.1.0 并计划使用 confluentinc-kafka-connect-elasticsearch--plugin。

这是我的弹性搜索接收器连接配置文件：-

name=elasticsearch-sink
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=test-elasticsearch-sink
topic.index.map=test-1:elastic_purchase_index1
key.ignore=true
connection.url=<url>
type.name=purchasev1
key.ignore=true
schema.ignore=true

【问题讨论】：

你研究过RegexRouter吗？或者，使用 ksqlDB 或 Kafka Streams 将事件过滤到新主题
正则表达式路由器将用于过滤事件并应用转换，例如替换字段并放入新主题。这将导致从一个我不想要的源主题放入多个主题。我认为 Kafka 流可以做到这一点（消耗事件然后发送到相应的弹性搜索索引）但想通过 kafka-connect 检查是否可能？
您不会使用 Kafka Streams 将数据发送到 Elastic... 参考 - topic.index.map docs.confluent.io/current/connect/kafka-connect-elasticsearch/…

标签： elasticsearch apache-kafka apache-kafka-connect

【解决方案1】：

基于 eventType 头域

使用ExtractTopic$Value + RegexRouter 将消息路由到各个索引。

然后在 ElasticSearch 接收器中使用 topics.regex 来定义要读取的主题的模式

【讨论】：

因为我只会从单个主题获取数据，所以不需要在弹性搜索接收器连接器配置中使用 topic.regex。
ExtractTopic$Value 用于提取主题内容中的值，但它如何提取标题属性？我的要求是从单个主题中获取所有消息，并根据标题键映射到弹性搜索中的不同索引。你能在我的接收器配置中提供示例配置来测试吗？我不知道如何解决它。感谢您的帮助。
您需要将一个主题取出到 Connect 记录中的多个主题名称（不是实际的 Kafka 主题）。我不知道您的完整 JSON 是什么样子，所以我无法提供示例。正如我所指出的，您需要提取值，然后将其路由到索引（该索引始终基于每个 Connect 记录的“主题名称”元数据字段）。因此，更新名称，您将获得不同的索引。如果遇到问题，请尝试几种组合并更新问题
如何使用 kafka connect 解析 kafka 标头字段？在 kafka 连接文档中找不到。有效负载的 JSON 无关紧要，因为路由到不同的索引将基于标头键而与有效负载无关。我正在检查您对 Connect 记录的建议
我认为你必须在 Github 上找到或编写自己的 Transform 来获取 Kafka 消息头。