使用 NEST Elasticsearch .NET 客户端对文档进行索引
引言
您可以通过多种方式使用 NEST Elasticsearch .NET 客户端将文档索引到 Elasticsearch 中。
本篇博客文章将演示一些简单的方法，从一次索引单个文档，到使用 BulkObservable 辅助工具的更高级方法。
单个文档
在 NEST 中，文档被建模为 POCO（普通旧式 CLR 对象），示例如下：
public class Person
{
public int Id { get; set; }
public 字符串 FirstName { get; set; }
public 字符串 LastName { get; set; }
}
此对象的实例（代表 Elasticsearch 中的单个文档）随后可以使用几种不同的方法进行索引。让我们以以下实例为例：
var person = new Person
{
Id = 1,
FirstName = "Martijn",
LastName = "Laarman"
};
IndexDocument<T> 和 IndexDocumentAsync<T> 方法提供了一种使用默认参数索引 T 类型单个文档的简单方法。可以检查此方法调用的结果，以确定索引操作是否成功。
// 返回 IIndexResponse 对象的同步方法
var indexResponse = client.IndexDocument(person);
// 返回可等待的 Task<IIndexResponse> 的异步方法
var indexResponseAsync = await client.IndexDocumentAsync(person);
// 检查同步操作的结果
if (!indexResponse.IsValid)
{
// If the request isn't valid, we can take action here
}
IsValid 属性可用于检查响应在功能上是否有效。这是一个 NEST 抽象，提供了一个单一的检查点，用于确认请求是否出现了问题。
如果您在索引文档时需要设置其他参数，可以使用流式语法或对象初始化器语法。这将使您能够更精细地控制索引过程。在下面的示例中，我们将把文档索引到名为“people”的索引中。
// 流式语法
var fluentIndexResponse = client.Index(person, i => i.Index("people"));
// 对象初始化器语法
var initializerIndexResponse = client.Index(new IndexRequest<Person>(person, "people"));
索引多个文档的一种简单方法是创建一个循环，在每次迭代中索引单个文档；然而，这是一种效率极低的方法，无法很好地扩展以处理大型文档集合。
多个文档
批量 API 可用于索引多个文档。首先，让我们创建一个要索引的文档集合：
var people = new []
{
new Person
{
Id = 1,
FirstName = "Martijn",
LastName = "Laarman"
},
new Person
{
Id = 2,
FirstName = "Stuart",
LastName = "Cam"
},
new Person
{
Id = 3,
FirstName = "Russ",
LastName = "Cam"
}
// snip
};
可以使用 IndexMany 和 IndexManyAsync 方法分别同步或异步索引多个文档。这些方法是 NEST 客户端特有的，封装了对客户端 Bulk 方法和批量 API 的调用，为索引大量文档提供了便捷的快捷方式。
请注意，这些方法会在单个 HTTP 请求中索引所有文档，因此对于非常大的文档集合，您需要将集合分区为许多较小的批量，并发出多个 Bulk 调用。当您发现自己需要这样做时，请考虑改用稍后在文章中介绍的 BulkAllObservable<T> 辅助程序。
// 返回 IBulkResponse 的同步方法
var indexManyResponse = client.IndexMany(people);
if (indexManyResponse.Errors)
{
// 可以检查响应中的错误
foreach (var itemWithError in indexManyResponse.ItemsWithErrors)
{
// 如果存在错误，可以枚举并检查它们
Console.WriteLine("未能索引文档 {0}：{1}",
itemWithError.Id, itemWithError.Error);
}
}
// 或者，可以异步索引文档
var indexManyAsyncResponse = await client.IndexManyAsync(people);
如果您需要对大量文档的索引进行更细粒度的控制，可以使用 Bulk 和 BulkAsync 方法，并利用描述符来自定义批量调用。
与上述 IndexMany 方法一样，文档会在单个 HTTP 请求中发送到 _bulk 终端。这意味着需要考虑 HTTP 请求的总体大小。对于索引大量文档，您可能需要使用 BulkAllObservable<T> 辅助程序。
// 返回一个可检查错误的 IBulkResponse
var bulkIndexResponse = client.Bulk(b => b
.Index("people")
.IndexMany(people)
);
// 异步版本
var asyncBulkIndexResponse = await client.BulkAsync(b => b
.Index("people")
.IndexMany(people)
);
BulkAllObservable<T> 辅助程序
使用 BulkAllObservable<T> 辅助方法，您可以专注于索引文档集合这一总体目标，而无需担心重试、退避或批量处理机制。
可以使用 BulkAll 方法和 BlockingSubscribeExtensions Wait() 扩展方法对多个文档进行索引。此辅助程序提供了在索引失败时自动重试/退避的功能，并可控制单个 HTTP 请求中索引的文档数量。
在以下示例中，每个请求都会索引 1,000 个文档，这些文档是从原始输入中批量处理的。如果文档数量庞大，可能会导致产生许多 HTTP 请求，每个请求包含 1,000 个文档（最后一个请求包含的文档数量可能会更少，具体取决于总数）。
该辅助程序会延迟枚举 IEnumerable<T> 集合，使您可以轻松索引大量文档，例如从分页数据库记录中具体化的文档。
var bulkAllObservable = client.BulkAll(people, b => b
.Index("people")
// 重试之间的等待时间
.BackOffTime("30s")
// 如果发生故障，尝试重试的次数
.BackOffRetries(2)
// 批量操作完成后刷新索引
.RefreshOnCompleted()
// 要进行的并发批量请求数
.MaxDegreeOfParallelism(Environment.ProcessorCount)
// 每个批量请求的项目数
.Size(1000)
)
// 执行索引，最多等待 15 分钟。
// 虽然 BulkAll 调用是异步的，但这属于阻塞操作
.Wait(TimeSpan.FromMinutes(15), next =>
{
// do something on each response e.g. write number of batches indexed to console
});
BulkAllObservable<T> 辅助程序公开了许多高级功能。
- BufferToBulk 允许在批量请求发送到服务器之前，对其内部的各个操作进行自定义。
- RetryDocumentPredicate 支持对是否应重试索引失败的文档进行精细化控制。
- DroppedDocumentCallback：如果文档在重试后仍未被索引，则会调用此委托。
client.BulkAll(people, b => b
.BufferToBulk((descriptor, list) =>
{
// 在批量请求分发之前
// 自定义各个操作
foreach (var item in list)
{
// index each document into either even-index or odd-index
descriptor.Index<Person>(bi => bi
.Index(item.Id % 2 == 0 ? "even-index" : "odd-index")
.Document(item)
);
}
})
.RetryDocumentPredicate((item, person) =>
{
// decide if a document should be retried in the event of a failure
return item.Error.Index == "even-index" && person.FirstName == "Martijn";
})
.DroppedDocumentCallback((item, person) =>
{
// 如果文档无法被索引，则调用此委托
Console.WriteLine($"Unable to index: {item} {person}");
})
);
摄取节点
由于 Elasticsearch 会自动将摄取请求重新路由到摄取节点，因此您无需指定或配置任何路由信息。但是，如果您正在进行大量的摄取工作并拥有专用的摄取节点，那么直接将索引请求发送到这些节点是合理的，这样可以避免集群中出现任何额外的跳转。
实现这一目标的最简单方法是创建一个专用的“索引”客户端实例，并将其用于索引请求。
// 摄取节点列表
var pool = new StaticConnectionPool(new []
{
new Uri("http://ingestnode1:9200"),
new Uri("http://ingestnode2:9200"),
new Uri("http://ingestnode3:9200")
});
var settings = new ConnectionSettings(pool);
var indexingClient = new ElasticClient(settings);
在复杂的集群配置中，使用监听连接池配合节点谓词来过滤出具有摄取功能的节点会更容易。这使您可以自定义集群，而无需重新配置客户端。
// 集群节点列表
var pool = new SniffingConnectionPool(new []
{
new Uri("http://node1:9200"),
new Uri("http://node2:9200"),
new Uri("http://node3:9200")
});
// 仅选择具有摄取功能的节点的谓词
var settings = new ConnectionSettings(pool).NodePredicate(n => n.IngestEnabled);
var indexingClient = new ElasticClient(settings);
采集管道
让我们修改 Person 类型以包含一些额外信息：
public class Person
{
public int Id { get; set; }
public 字符串 FirstName { get; set; }
public 字符串 LastName { get; set; }
public 字符串 IpAddress { get; set; }
public GeoIp GeoIp { get; set; }
}
public class GeoIp
{
public 字符串 CityName { get; set; }
public 字符串 ContinentName { get; set; }
public 字符串 CountryIsoCode { get; set; }
public GeoLocation Location { get; set; }
public 字符串 RegionName { get; set; }
}
我们可以创建一个摄取管道，在值被索引之前对其进行处理。假设我们的应用程序始终要求姓氏大写，并将首字母索引到它们自己的字段中。我们还有一个 IP 地址，希望将其转换为人类可读的位置信息。
我们可以通过创建自定义映射和创建摄取管道来实现此需求。然后，无需进行任何进一步更改即可使用新的 Person 类型。
首先，我们将创建索引和自定义映射：
client.CreateIndex("people", c => c
.Mappings(ms => ms
.Map<Person>(p => p
//从类型自动创建映射
.AutoMap()
//覆盖 AutoMap() 推断出的任何映射
.Properties(props => props
// 创建一个额外的字段来存储首字母
.Keyword(t => t.Name("initials"))
//将字段映射为 IP 地址类型
.Ip(t => t.Name(dv => dv.IpAddress))
// 将 GeoIp 映射为对象
.Object<GeoIp>(t => t.Name(dv => dv.GeoIp))
)
)
)
);
接下来，我们将创建一个摄取管道，利用 6.7 版本中捆绑的摄取-geoip 插件。
client.PutPipeline("person-pipeline", p => p
.Processors(ps => ps
//将姓氏转换为大写
.Uppercase<Person>(s => s
.Field(t => t.LastName)
)
// 使用 Painless 脚本填充新字段
.Script(s => s
.Lang("Painless")
.Source("ctx.initials = ctx.firstName.substring(0,1) + ctx.lastName.substring(0,1)")
)
// 使用摄取-geoip 插件根据提供的 IP 地址丰富 GeoIp 对象
.GeoIp<Person>(s => s
.Field(i => i.IpAddress)
.TargetField(i => i.GeoIp)
)
)
);
现在，让我们使用这个新的索引和摄取管道来索引一个 Person 实例。
var person = new Person
{
Id = 1,
FirstName = "Martijn",
LastName = "Laarman",
IpAddress = "139.130.4.5"
};
// 使用创建的管道索引文档
var indexResponse = client.Index(person, p => p
.Index("people")
.Pipeline("person-pipeline")
);
现在搜索显示了带有丰富值的索引文档。{
"took": 5,
"timed_out": false,
"_shards":{
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
} {分片},
"hits": {
"total": 1,
"max_score": 1,
"hits": [
{
"_index": "people",
"_type": "person",
"_id": "1",
"_score": 1,
"_source": {
"firstName": "Martijn",
"lastName": "LAARMAN",
"initials": "ML",
"geoIp": {
"continent_name": "Oceania",
"region_iso_code": "AU-NSW",
"city_name": "Sydney",
"country_iso_code": "AU",
"region_name": "New South Wales",
"location":{
"lon": 151.2167,
"lat": -33.7333
}
},
"ipAddress": "139.130.4.5",
"id": 1
}
}
]
}}
当指定管道时，在索引过程中会增加文档扩充的开销；在上述示例中，即执行大写转换和 Painless 脚本的开销。
对于大型批量请求，明智的做法是增加默认的索引超时时间，以避免异常。
client.Bulk(b => b
.Index("people")
.管道("person-管道")
//增加 Elasticsearch 服务器端的超时时间
.Timeout("5m")
.IndexMany<Person>(people)
.RequestConfiguration(rc => rc
// 在中止请求之前，增加客户端上的 HTTP 请求超时时间
.RequestTimeout(TimeSpan.FromMinutes(5))
)
);
总结
在本篇博客文章中，我们介绍了从索引单份文档的简单情况，到使用摄取管道批量索引多份文档的各种场景。
欢迎在您自己的集群中试用，或者快速部署一个 14 天免费试用的 Elasticsearch Service，该服务运行于 Elastic Cloud 之上。如果您遇到任何问题或有任何疑问，请随时访问 Discuss 论坛与我们联系。
有关使用 NEST Elasticsearch .NET 客户端进行索引的完整文档，请参阅我们的文档。