使用 ElasticLowLevelClient 客户端在 Elasticssearch 中进行批量索引

Posted 2023-03-22

【中文标题】使用 ElasticLowLevelClient 客户端在 Elasticssearch 中进行批量索引

【英文标题】：Bulk Indexing in Elasticssearch using the ElasticLowLevelClient client

【发布时间】：2018-07-23 02:27:33

【问题描述】：

我正在使用 ElasticLowLevelClient 客户端来索引 elasticsearch 数据，因为我无权访问 POCO 对象，因此需要将其作为原始字符串进行索引。我可以通过调用成功索引单个对象：

client.Index<object>(indexName, message.MessageType, message.Id, 
    new Elasticsearch.Net.PostData<object>(message.MessageJson));

如何使用 ElasticLowLevelClient 客户端批量插入索引？批量插入 API 都需要我没有的索引文档的 POCO，例如：

 ElasticsearchResponse<T> Bulk<T>(string index, PostData<object> body,
      Func<BulkRequestParameters, BulkRequestParameters> requestParameters = null)

我可以为每个对象并行调用 API，但这似乎效率低下。

【参考方案1】：

低级客户端泛型类型参数是预期响应的类型。

如果您使用在高级客户端上公开的低级客户端，通过.LowLevel 属性，您可以发送批量请求，其中您的文档是 JSON 字符串，如下 5.x 中所示

var client = new ElasticClient(settings);


var messages = new [] 

    new Message 
     
        Id = "1", 
        MessageType = "foo", 
        MessageJson = "\"name\":\"message 1\",\"content\":\"foo\"" 
    ,  
    new Message 
     
        Id = "2", 
        MessageType = "bar", 
        MessageJson = "\"name\":\"message 2\",\"content\":\"bar\"" 
       
;

var indexName = "my-index";

var bulkRequest = messages.SelectMany(m => 
    new[]
    
        client.Serializer.SerializeToString(new
            
                index = new
                
                    _index = indexName,
                    _type = m.MessageType,
                    _id = m.Id
                
            , SerializationFormatting.None),
        m.MessageJson
    );

var bulkResponse = client.LowLevel.Bulk<BulkResponse>(string.Join("\n", bulkRequest) + "\n");

发送以下批量请求

POST http://localhost:9200/_bulk
"index":"_index":"my-index","_type":"foo","_id":"1"
"name":"message 1","content":"foo"
"index":"_index":"my-index","_type":"bar","_id":"2"
"name":"message 2","content":"bar"

几个重点

我们需要自己构建批量请求以使用低级批量 API 调用。由于我们的文档已经是字符串，因此构建字符串请求是有意义的。 我们序列化一个匿名类型，每个批量项的操作和元数据没有缩进。 MessageJson 不能包含任何换行符，因为这会破坏批量 API；换行符是正文中 json 对象的分隔符。 因为我们使用的是暴露在高级客户端上的低级客户端，所以我们仍然可以利用高级请求、响应和序列化程序。批量请求返回一个BulkResponse，您可以像往常一样使用高级客户端发送批量请求时使用它。

【讨论】：

太棒了，我刚试了一下，效果很好！感谢您非常详细的回复，以及描述与批量索引相关的过程和要点的附加要点。