关于误删除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，即不记录