关于误删除elasticSearch 索引,怎么能快速找回?

背景

之前公司小王在工作中清理elasticSearch 索引,不小心使用脚本清空了最近使用的重要索引,导致开发无法准确的进行定位生产问题,造成了很大困扰。

当时我们的生产环境中是这样配置日志系统的:服务器 -> filebeat -> kakfa -> logstash -> elasticsearch -> kibana 页面。其中在缓存配置中我们配置了kafka集群。

es 数据丢失之后,我们考虑使用拉去kafka 集群中的存储的消息,让它重新消费一边,更换消费组,从头开始进行消费 配置 auto_offset_reset => "earliest" 设定。

具体实践如下:

logstash.conf配置

如下配置是我测试通过的

input {
kafka {
bootstrap_servers => ["192.168.39.23:9082,192.168.39.24:9082,192.168.39.25:9082"] topics_pattern => ["MY_REPORT_TOPIC_.*"]
client_id => "logstash-1"
group_id => "logstash_to_elastic2" # 这里的组要更新一下,原先是logstash_to_elastic
auto_offset_reset => "earliest" # 这里设定消费模式为从头开始消费,也就是从最初开始消费
consumer_threads => 1
decorate_events => true
codec => "json"
}
} output {
elasticsearch {
hosts => "http://192.168.39.27:9200"
index => "logtype-%{[logType]}"
document_id => "%{[@metadata][kafka][topic]}-%{[@metadata][kafka][partition]}-%{[@metadata][kafka][offset]}"
}
}

重点说下读取kafka的配置:

bootstrap_servers:kafka集群地址
topics_pattern:我这里是通配的前缀为”MY_REPORT_TOPIC_”的topic,注意后面跟的是.*,不要忘了这个点,正则的语法,通配所有字符用的
client_id:自定义的客户端id
group_id:自定义组id,两个消费者同时消费同一组会造成其中一个无法消费,但是只要创建多个partition就可以了,即组内消费者数量和分区数相同就都可以消费了,这样一个组内所有消费者能够分担负载
auto_offset_reset:
  • earliest

    当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
  • latest

    当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
  • none

    topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常

由于我有些历史数据要消费,所以设置为earliest从头读的

consumer_threads:一般最佳配置是同一个组内consumer个数(或线程数)等于topic的分区数,如果大于分区数会造logstash进程闲置,否则一个进程访问多个分区。如果有多个Logstash实例,那就让实例个数 * consumer_threads等于分区数即可

如何避免重复消费

数据写入到Kafka时没有重复,但后续流程可能因为网络抖动、传输失败等导致重试造成数据重复。如何解决呢?不需要依赖业务数据就可以去重。去重的原理也很简单,利用es document id即可。

对于es,如果写入数据时没有指定document id,就会随机生成一个uuid,如果指定了,就使用指定的值。对于需要去重的场景,我们指定document id即可。

在output elasticsearch中可以通过document_id字段指定document id,我们需要构造出一个”uuid”能惟一标识kafka中的一条数据,这个非常简单:<topic>+<partition>+<offset>,这三个值的组合就可以唯一标识kafka集群中的一条数据。

input kafka插件也已经帮我们把消息对应的元数据信息记录到了@metadata(Logstash的元数据字段,不会输出到output里面去)字段里面:

  • [@metadata][kafka][topic]:索引信息
  • [@metadata][kafka][consumer_group]:消费者组信息
  • [@metadata][kafka][partition]:分区信息
  • [@metadata][kafka][offset]:offset信息
  • [@metadata][kafka][key]:消息的key(如果有的话)
  • [@metadata][kafka][timestamp]:时间戳信息(消息创建的时间或者broker收到的时间)

因此就有了上面的配置:

document_id => "%{[@metadata][kafka][topic]}-%{[@metadata][kafka][partition]}-%{[@metadata][kafka][offset]}"
decorate_events:只有当decorate_events选项配置为true的时候,上面的[@metadata](https://github.com/metadata)才会记录那些元数据,否则不会记录。而该配置项的默认值是false,即不记录
上一篇:说说CSS学习中的瓶颈


下一篇:elasticsearch 索引清理脚本及常用命令