在JAVA中将Elasticsearch索引加载到Lucene API

19-01-09 banq
              

每隔一段时间,Elasticsearch中就会出现意外(或无意)崩溃。对于我的情况,在Elasticsearch的大量IO操作期间是硬件故障(让我们假设我没有任何副本或者我设法使所有集群崩溃)。经过一些研究,我发现它搞砸了许多索引的状态文件(已损坏!)。我想,如果Elasticsearch使用Lucene,我肯定可以加载我的数据并使用Lucene API重新索引它。

重要说明:要使本指南处理索引,必须在映射中启用_source字段的索引。如果您因任何原因禁用了_source字段,则此代码将不适用于您的索引。Elasticsearch建议始终启用它。

在我们开始之前,让我们引用两个重要的术语。更多细节可以在这里找到。

  • 索引:用于在Elasticsearch中存储数据的位置。
  • Shard分片或Lucene索引实例(!):每个索引由一个或多个分片组成。它是一个独立的搜索引擎,可以索引和处理Elasticsearch集群中数据子集的查询。

正如你在这里看到,我们的数据驻留在索引和这些分布到多个分片,后者实际上是一个Lucene索引 !

让我们看一下Elasticsearch数据的结构,并尝试找到这些Lucene索引。您可以配置数据路径elasticsearch.yml与关键path.data。

索引indices目录下有一个名为类似eCCAJ-x6SOuN6w7vqr4tGQ格式的文件夹。(这些项目的说明是完全独立的主题。为了简单起见,只需要知道状态文件是由Elasticsearch生成的,它是一个SMILE-encode文件,其中包含Elasticsearch元数据,如索引名称,节点ID等。您可以使用十六进制编辑器来检查文件)

让我们通过Curl我们的Elasticsearch实例来列出我们的索引。

$ curl elastic-host:9200/_cat/indices
yellow open bad_ip eCCAJ-x6SOuN6w7vqr4tGQ 5 1 4609 0 1.2mb 1.2mb

这是我的索引,其中索引了一些恶意IP数据。我们来看看那个目录。

$ ls my-elasticsearch/data/nodes/0/indices/eCCAJ-x6SOuN6w7vqr4tGQ
0      1      2      3      4      _state

这是我们索引的分片,或者是我们要加载的Lucene索引实例。

提示:如果此时Elasticsearch实例崩溃,您只需在状态文件上运行cat命令即可查看它们所属的索引。所有文件都不可读,但至少你应该看到你的索引名称。

重要说明#2:在继续执行代码之前,您应该确保您的Lucene索引没有损坏。为了检查和修复您的Lucene索引(或Elasticsearch分片),我强烈建议您使用CheckIndex工具。

现在,我们找到了Lucene索引,让我们做一些编码来拯救我们的数据。我将创建一个示例Maven项目。这是我们需要的依赖项:

<dependency>
    <groupId>org.apache.lucene</groupId>
    <artifactId>lucene-core</artifactId>
    <version>7.5.0</version>
</dependency>

<dependency>
    <groupId>org.apache.lucene</groupId>
    <artifactId>lucene-queryparser</artifactId>
    <version>7.5.0</version>
</dependency>

重要提示#3:您应该查看您的Elasticsearch实例的Lucene版本,并在此处使用相同的Lucene API版本。运行该命令curl elastic-host:9200 并查找lucene_version密钥。

为了简单起见,我将在main方法中编写所有代码并抛出异常。这取决于您如何构建代码。

import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;

public class Main {
    public static void main(String[] args) throws IOException {
        String luceneIndexPath = "my-elasticsearch/data/nodes/0/indices/eCCAJ-x6SOuN6w7vqr4tGQ/0/index";
        Directory index = FSDirectory.open(Paths.get(luceneIndexPath));

        IndexReader reader = DirectoryReader.open(index);

        System.out.println(reader.maxDoc());
        
        reader.close();
        index.close();
    }
}

运行它,你会看到你的分片文件的数量。就我而言,我的第0个碎片中有952个文档。如果列出所有主分片并对其进行总结,则它将等于您的Elasticsearch索引中的总文档数。

  • Directory类来自Lucene API,但它与Java i / o库没有太大区别。它只是为了简化不同来源的实现,例如存储在数据库中的索引。我的索引的实际实例是 MMapDirectory。如果您有不同的配置,可能会有所不同。(同样,这是另一个话题。)
  • IndexReader是一个用于访问索引的视点的抽象类。

我们可以使用我们的reader变量访问大量信息,例如段信息和文档本身。我们想要的是找到这些文件。让我们编写那部分代码。

for(int i = 0; i < reader.maxDoc(); i++){
    if(((DirectoryReader) reader).isCurrent()){
      Document document = reader.document(i);

      String source = document.getBinaryValue("_source").utf8ToString();
      System.out.println(source);
    }
}

我们所做的是一个简单的循环并获得该位置的相应文档。之后,我们只获取JSON所在的_source键。它以二进制形式存储,因此,首先我们需要获取二进制值,然后进行utf8ToString转换。