将kafka中创建的主题自动下沉到elasticsearch答案

【问题标题】：Auto sinking of topics being created in kafka to elasticsearch将kafka中创建的主题自动下沉到elasticsearch
【发布时间】：2019-04-13 00:48:40
【问题描述】：

我在 kafka（test1、test2、test3）中创建了主题，我想在创建时将它们下沉到弹性。我尝试了topics.regex，但它只为已经存在的主题创建索引。动态创建新主题时如何将其放入索引中？

这是我用于 kafka-sink 的连接器配置：

{
    "name": "elastic-sink-test-regex",
    "config": {
        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
        "tasks.max": "1",
        "topics.regex": "test[0-9]+",
        "type.name": "kafka-connect",
        "connection.url": "http://192.168.0.188:9200",
        "key.ignore": "true",
        "schema.ignore": "true",
        "schema.enable": "false",
        "batch.size": "100",
        "flush.timeout.ms": "100000",
        "max.buffered.records": "10000",
        "max.retries": "10",
        "retry.backoff.ms": "1000",
        "max.in.flight.requests": "3",
        "is.timebased.indexed": "False",
        "time.index": "at"
    }
}

【问题讨论】：

我相信topics.regex 会按计划阅读新主题，否则您必须物理重启连接器。只是任务，而不是机器。 .无论哪种情况，都不会立即获取新主题

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

【解决方案1】：

在此连接器重新启动（或发生计划的重新平衡）之前，接收器连接器不会读取新主题。您可以运行 Kafka Stream 从新主题中读取消息并将它们放入类似结果的主题中。 Sink 连接器从类似结果的主题中读取数据。

要保存匹配的“消息 - 主题”，您可以使用 Kafka 记录头。

确保它符合您的要求！

【讨论】：