kafka-connect-elasticsearch:如何根据Kafka topic的header中的某个值删除文档
kafka-connect-elasticsearch是一个用于将Kafka中的数据流式传输到Elasticsearch的连接器。它可以根据Kafka topic的header中的某个值来删除文档。
具体实现这个功能的步骤如下：
- 配置Kafka Connect Elasticsearch连接器：在Kafka Connect配置文件中，配置kafka-connect-elasticsearch连接器，并指定Elasticsearch集群的地址和其他相关参数。
- 创建Kafka topic：使用Kafka命令行工具或者Kafka API创建一个包含需要处理的数据的Kafka topic。
- 发送带有header的消息到Kafka topic：使用Kafka生产者API发送消息到Kafka topic，并在消息的header中设置需要用来删除文档的值。
- 配置Kafka Connect Elasticsearch连接器的转换器：在连接器的配置中，指定转换器，以便将Kafka消息转换为Elasticsearch文档。
- 配置Elasticsearch索引和文档ID的生成规则：在连接器的配置中，指定如何生成Elasticsearch索引和文档ID，可以使用Kafka消息的其他字段或者自定义逻辑。
- 配置删除策略：在连接器的配置中，指定删除策略为根据Kafka topic的header中的某个值来删除文档。
- 启动Kafka Connect Elasticsearch连接器：启动Kafka Connect Elasticsearch连接器，它将自动从Kafka topic中读取消息，并将其转发到Elasticsearch。
- 监控和调试：使用Kafka Connect和Elasticsearch的监控工具来监控连接器的运行状态，并进行必要的调试和故障排除。
kafka-connect-elasticsearch的优势在于它提供了一个简单而强大的方式将Kafka中的数据流式传输到Elasticsearch，使得数据的实时索引和搜索变得更加容易。它可以广泛应用于日志分析、实时监控、数据仓库等场景。
腾讯云提供了一系列与Kafka和Elasticsearch相关的产品和服务，可以帮助用户快速搭建和管理Kafka和Elasticsearch集群。具体推荐的产品和产品介绍链接如下：
- 云消息队列CMQ：腾讯云的消息队列服务，可以用于替代Kafka作为消息中间件。产品介绍链接：https://cloud.tencent.com/product/cmq
- 云数据库TencentDB for Elasticsearch：腾讯云的托管Elasticsearch服务，提供了高可用、高性能的Elasticsearch集群。产品介绍链接：https://cloud.tencent.com/product/es
请注意，以上推荐的产品和链接仅供参考，具体选择应根据实际需求和情况进行。
相关·内容
我有一个处理流，看起来像这样： 来自S3的文件记录->来自发布到->kafka主题的源应用程序的自定义处理->Kafka连接->弹性搜索 这适用于创建/更新的场景。但是，我们希望处理文件的删除场景。我们的应用程序发布一个删除操作的事件，并将其设置为Kafka消息中标头值的一部分。我们希望删除<e
浏览 23提问于2021-04-07得票数 0
回答已采纳
type.name=logplugin.path=/home/kafka-connect-elasticsearch/
我去运行剧本..。/kafka-connect-elasticsearch-3.3.0-development/share/java/
浏览 1提问于2017-09-14得票数 0
1回答
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {}/*public void listenBatch(List<String> messages, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic)th
浏览 2提问于2019-12-05得票数 6
我已经在header中保存了topic，并尝试在Kafka URL中访问它，如下所示： from(...).to("kafka:${header.topic}?brokers=localhost:9092") 不幸的是，它显示Kafka URL无效： org.apache.kafka.common.errors.InvalidTopicException: Inval
浏览 44提问于2021-10-15得票数 0
回答已采纳
1回答
我对使用Prometheus指标非常陌生，有人要求我根据过去五分钟内某个值是否增加了10%来编写警报。 简而言之，当我的Kafka主题死信数在5分钟内增加了10%时，我必须报告。我可以使用以下查询来计算主题/子主题的当前死信数： count by(topic) (kafka_burrow_topic_partition_offset{topic=~".+my-subscription-name.+"})
浏览 76提问于2021-09-02得票数 2
回答已采纳
如何使用kafka从Kafka消息中检索传入的标头，以便将其存储为mongodb Sink连接器到MongoDB的附加数据字段。我想在某个地方有一个提取头的函数。例卡夫卡值
/
浏览 10提问于2020-10-11得票数 5
1回答
火花结构化流确认消息
、、、
我使用从卡夫卡主题(例如topic1)中读取，并使用接收器写入另一个主题(主题1-结果)。我可以看到，在使用Sink编写到另一个主题之后，这些消息并没有从Topic1中删除。// Subscribe to 1 topic .readStream .option("kafka.bootstrap.servers", &q
浏览 2提问于2020-02-11得票数 0
回答已采纳
1回答
这是我的配置 curl --location --request PUT 'http://localhost:8083/connectors/customer_sink_1/config' \"reporter.error.topic.replication.factor":"1"
}' 我从windows上的apache kafka
浏览 47提问于2021-10-19得票数 0
回答已采纳
在我们的事务处理系统中实现了kafka流处理。该解决方案实现如下：
Kafka producer将事件发布到kafka topic，流处理器处理输入事件并进行聚合操作。由于第一个topic中没有实现消费者，如何从第一个topic中移除已处理的消息。
浏览 0提问于2017-08-14得票数 1
我是新来的春天卡夫卡。我有一个微服务，它发送消息与kafka密钥，这是一个用户定义的对象。假设消息是由密钥myKey发送的。我想在监听程序中创建一个新的扩展密钥，如下所示：
@KafkaListener(groupId = B
浏览 1提问于2019-09-26得票数 2
2回答
就我而言，那是在/tmp/kafka-logs/<topic-name>。然后只需使用ls -l来查看大小(以字节为单位)。
但是，使用：$ bin/kafka-topics.sh -并检查其标题：$ b
浏览 6提问于2017-08-08得票数 2
回答已采纳
我想为我的pyspark应用程序使用kafka自定义分割器，从kafka推送到另一个kafka主题。使用pyspark处理将数据从源转换到宿。我希望能够根据data/message中的某个键控制应该将数据推送到哪个分区。在中，我找不到此类用例的任何参考资料或示例。我正在使用python处理和pyspark，被用作kafka客户端，但它也缺乏自定义分区程序的文档</
浏览 1提问于2021-11-01得票数 1
2回答
我想在我们的微服务中实现发件箱模式，使用Mongo连接器，在发件箱( MongoDB集合)中，我使用以下字段存储主题数据：kafka_topic、kafka_key、kafka_value、kafka_header_keys、kafka_header_values。但是，我应该如何配置mongo连接器来动态地从kafka_topic字段中选择主题
浏览 3提问于2020-12-25得票数 2
1回答
我得到的异常风暴拓扑。
java.lang.NoSuchMethodError: org.apache.kafka.common.network.NetworkSend.at kafka.network.BlockingChannel.send(BlockingChannel.scala:112) ~[stormjar.jar:?]at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsum
浏览 0提问于2018-08-27得票数 0
回答已采纳
/usr/local/confluent/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties目前，我只在一个EC2中运行我所有的设置，现在如果我必须增加5个EC2来使连接器更加并行并加快速度，我如何才能使这个连接理解5多个EC2已经被添加，并且它必须共享工作负载？如何确认所有5个EC2都在同一个连接器下
浏览 7提问于2020-02-01得票数 1
根据文档安装和配置Suricata 5.0.2之后。我尝试通过添加以下内容来更改suricata.yaml中的一些配置： enabled: yes kafkaxxx-kafka-online005:9092, xxx-kafka-online00
浏览 8提问于2020-02-18得票数 1
3回答
我的服务器机器上运行着单节点kafka。首先从一些java应用程序日志文件中读取数据，注入相同的kafka。它工作得很好，我可以在控制台上使用"bin/ kafka - console -consumer.sh --zookeeper localhost:2181 --topic test --from- see“命令查看kafka中的数据。但是另一个logstash实例，它从kafka<
浏览 2提问于2016-08-16得票数 1
1回答
:9092 --topic zipkin --from-beginning)的kafka-console-console. the时，我没有看到任何zipkin跟踪。<scope>test</scope> </dependency>这些是我在根据spring-cloud-stream-bin
浏览 0提问于2018-02-12得票数 1
回答已采纳
我收藏了ElasticSearch &Kibana7.6.2input { bootstrap_servers => "******" decorate_events => true
add_field => { "[topic_name]" =>"%{[@metadata][<em
浏览 0提问于2020-04-07得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
