Elastic Search 摄取附件插件块

Question

我正在使用 NEST (C#) 和 ingest attachment plugin 将成千上万的文档摄取到 Elastic 搜索实例中。不幸的是，过了一会儿，一切都静止了——即没有更多的文件被摄取。日志显示：

[2019-02-20T17:35:07,528][INFO ][o.e.m.j.JvmGcMonitorService] [BwAAiDl] [gc][7412] overhead, spent [326ms] collecting in the last [1s]

不确定这是否告诉任何人任何事情？顺便说一句，有没有更有效的方法来摄取许多文档（而不是使用数千个 REST 请求）？

我正在使用这种代码：

client.Index(new Document
{
    Id = Guid.NewGuid(),
    Path = somePath,
    Content = Convert.ToBase64String(File.ReadAllBytes(somePath))
}, i => i.Pipeline("attachments"));

定义管道：

client.PutPipeline("attachments", p => p
    .Description("Document attachment pipeline")
    .Processors(pr => pr
        .Attachment<Document>(a => a
        .Field(f => f.Content)
        .TargetField(f => f.Attachment)
        )
        .Remove<Document>(r => r
        .Field(f => f.Content)
        )
    )
);

Answer 1

日志表明在 Elasticsearch 服务器端执行垃圾收集花费了大量时间；这很可能是您看到的大型停止事件的原因。如果您在集群上启用了监控（理想情况下将此类数据导出到单独的集群），我会分析这些数据，看看它是否能阐明为什么会发生大型 GC。

有没有更有效的方法来摄取大量文档（而不是使用数千个 REST 请求）？

是的，您在单独的索引请求中为每个附件编制索引。根据每个附件的大小，base64 编码，您可能希望在一个批量请求中发送多个

// Your collection of documents
var documents = new[]
{
    new Document
    {
        Id = Guid.NewGuid(),
        Path = "path",
        Content = "content"
    },
    new Document
    {
        Id = Guid.NewGuid(),
        Path = "path",
        Content = "content" // base64 encoded bytes
    }
};

var client = new ElasticClient();

var bulkResponse = client.Bulk(b => b
    .Pipeline("attachments")
    .IndexMany(documents)
);

如果您正在从文件系统读取文档，您可能希望懒惰地枚举它们并发送批量请求。在这里，您也可以使用BulkAll 辅助方法。

首先有一些惰性枚举的文档集合

public static IEnumerable<Document> GetDocuments()
{
    var count = 0;
    while (count++ < 20)
    {
        yield return new Document
        {
            Id = Guid.NewGuid(),
            Path = "path",
            Content = "content" // base64 encoded bytes
        };
    }
}

然后配置BulkAll调用

var client = new ElasticClient();

// set up the observable configuration
var bulkAllObservable = client.BulkAll(GetDocuments(), ba => ba
    .Pipeline("attachments")
    .Size(10)
);

var waitHandle = new ManualResetEvent(false);

Exception exception = null;

// set up what to do in response to next bulk call, exception and completion
var bulkAllObserver = new BulkAllObserver(
    onNext: response => 
    {
        // perform some action e.g. incrementing counter
        // to indicate how many have been indexed
    },
    onError: e =>
    {
        exception = e;
        waitHandle.Set();
    },
    onCompleted: () =>
    {
        waitHandle.Set();
    });

// start the observable process
bulkAllObservable.Subscribe(bulkAllObserver);

// wait for indexing to finish, either forever,
// or set a max timeout as here.
waitHandle.WaitOne(TimeSpan.FromHours(1));

if (exception != null)
    throw exception;

大小决定了在每个请求中发送多少文档。对于集群的大小没有硬性规定，因为它可能取决于许多因素，包括摄取管道、文档的映射、文档的字节大小、集群硬件等。您可以配置observable 可重试未能被索引的文档，如果您看到 es_rejected_execution_exception，则您的集群可以同时处理的内容已受到限制。

另一个建议是文档 ID。我看到您正在为文档的 id 使用新的 Guid，这对我来说意味着您不在乎每个文档的值是什么。如果是这种情况，我建议不要发送 Id 值，而是允许 Elasticsearch 为每个文档生成一个 id。这很可能会导致improvement in performance （我相信自这篇文章以来，Elasticsearch 和 Lucene 中的实现已经略有变化，但重点仍然存在）。