Elastic.Ingest.Elasticsearch 0.7.0

Elastic.Ingest.Elasticsearch

Elastic.Channels实现了BufferedChannelBase，允许数据推送到索引或数据流中

DataStreamChannel<TEvent>

这是一种专门用于将带有时间戳的数据写入Elasticsearch数据流的通道。例如，给定以下文档。

public class TimeSeriesDocument
{
    [JsonPropertyName("@timestamp")]
    public DateTimeOffset Timestamp { get; set; }

    [JsonPropertyName("message")]
    public string Message { get; set; }
}

可以创建一个通道将数据推送到logs-dotnet-default数据流。

var dataStream = new DataStreamName("logs", "dotnet");
var bufferOptions = new BufferOptions { }
var options = new DataStreamChannelOptions<TimeSeriesDocument>(transport)
{
  DataStream = dataStream,
  BufferOptions = bufferOptions
};
var channel = new DataStreamChannel<TimeSeriesDocument>(options);

注意：有关Elastic数据流命名约定的更多信息，请参见此处：https://elastic.ac.cn/blog/an-introduction-to-the-elastic-data-stream-naming-scheme

现在我们可以使用DataStreamChannel将数据推送到Elasticsearch。

var doc = new TimeSeriesDocument 
{ 
    Timestamp = DateTimeOffset.Now, 
    Message = "Hello World!", 
}
channel.TryWrite(doc);

目标数据流初始化

可选地，可以使用以下方式初始化目标数据流。

await channel.BootstrapElasticsearchAsync(BootstrapMethod.Failure, "7-days-default"); 

这将尝试设置目标数据流，并使用7-days-default ILM策略。如果失败（因为提供了BootstrapMethod.Failure），则引发异常。

将根据目标数据流的部分类型和数据集创建索引模板以及相关的组件模板。

IndexChannel<TEvent>

一个专门用于将目录数据写入Elastic索引的通道。目录数据通常是具有某种标识符的。

给定以下最小化文档

public class CatalogDocument
{
    [JsonPropertyName("id")]
    public string Id { get; set; }

    [JsonPropertyName("title")]
    public string Title { get; set; }

    [JsonPropertyName("created")]
    public DateTimeOffset Created { get; set; }
}

我们可以创建一个IndexChannel<>来推送CatalogDocument实例。

var options = new IndexChannelOptions<CatalogDocument>(transport)
{
    IndexFormat = "catalog-data-{0:yyyy.MM.dd}",
    BulkOperationIdLookup = c => c.Id,
    TimestampLookup = c => c.Created,
};
var channel = new IndexChannel<CatalogDocument>(options);

现在我们可以使用以下方式推送数据

var doc = new CatalogDocument 
{ 
    Created = date, 
    Title = "Hello World!", 
    Id = "hello-world" 
}
channel.TryWrite(doc);

这将向catalog-data-2023.01.1推送数据，因为TimestampLookup给出了对IndexFormat的Created值。

IndexFormat也可以简单地是一个固定字符串，用于写入Elasticsearch别名/索引。

BulkOperationIdLookup决定了文档应该使用create还是index操作推送到Elasticsearch。

引导目标索引

可选地，可以使用以下方式对目标索引进行引导。

await channel.BootstrapElasticsearchAsync(BootstrapMethod.Failure, "7-days-default"); 

这将尝试设置目标数据流，并使用7-days-default ILM策略。如果失败（因为提供了BootstrapMethod.Failure），则引发异常。

将根据名称使用IndexFormat创建具有配套组件模板的索引模板。