Elasticsearch7.13用NEST .NET客户端查询超1万条日志的实现问题

1. 能否直接通过NEST代码获取PIT ID #

完全可以，你代码中已经写对了对应的调用方法：

var openPit = await _elasticClient.OpenPointInTimeAsync(indexName, d => d.KeepAlive("1m"));
var pitId = openPit.Id;

这段代码的效果和Kibana中执行POST请求获取PIT ID完全一致，不需要依赖Kibana CLI。

2. 排序字段相关问题 #

• 不需要必须使用文档ID作为排序依据，你可以直接使用timestamp作为search_after的排序字段。
• 注意：如果存在多条日志timestamp完全相同的情况，仅用timestamp排序可能导致分页时出现数据重复或者遗漏，推荐新增ES内置字段_shard_doc作为第二排序字段保证排序唯一性，该字段不需要你在EsSource实体类中定义。
• 你不需要额外调用接口获取所有文档ID，属于冗余操作，完全没有必要。

3. 现有代码修正说明 #

你的代码中存在几处可优化/错误的点：

• 你写的Enumerable.Range批量插入的循环逻辑完全多余，你的需求是查询数据不是插入数据，直接删除这段逻辑即可。
• searchAfter初始值不需要赋值为0，首次请求不需要传SearchAfter参数，或者传null即可。
• 每次查询后，searchAfter取上一次查询结果最后一条的Sort属性值即可，不需要自己从实体类取值拼接。
• 需要定义一个全局列表存储每批查询到的文档，最后返回该列表。

修正后的核心代码示例 #

[HttpGet("GetMonthlyLogs")]
public async Task<List<EsSource>> GetLogsByDate()
{
    string indexName = "customer-simulation-es-app-logs*";
    var connectionSettings = new ConnectionSettings(new Uri("http://localhost:9200"));
    connectionSettings.DefaultIndex(indexName);
    connectionSettings.EnableDebugMode();
    _elasticClient = new ElasticClient(connectionSettings);
    
    // 存储最终结果
    var result = new List<EsSource>();
    // 打开PIT
    var openPit = await _elasticClient.OpenPointInTimeAsync(indexName, d => d.KeepAlive("1m"));
    var pitId = openPit.Id;
    // 存储上一次的排序值
    List<object> lastSort = null;

    try
    {
        while (true)
        {
            var response = await _elasticClient.SearchAsync<EsSource>(s => s
                .TrackTotalHits(false)
                .Size(1000)
                .PointInTime(pitId, d => d.KeepAlive("1m"))
                .Source(src => src.Includes(i => i
                    .Fields(f => f.timestamp,
                        f => f.level,
                        f => f.messageTemplate,
                        f => f.message)))
                .Query(q => q
                    .Bool(b => b
                        .Should(
                            m => m.Match(ma => ma.Field(fa => fa.level).Query("Error")),
                            m => m.Match(ma => ma.Field(fa => fa.level).Query("Information")))
                        .Filter(f => f.DateRange(dr => dr
                            .Field("@timestamp")
                            .GreaterThanOrEquals("2021-07-14T00:00:00.000-05:00")
                            .LessThanOrEquals("2021-07-14T23:59:59.999-05:00")))
                        .MinimumShouldMatch(1)))
                // 按timestamp + _shard_doc排序保证唯一性
                .Sort(srt => 
                    srt.Ascending(f => f.timestamp)
                        .Ascending("_shard_doc"))
                // 首次请求lastSort为null，不会生效
                .SearchAfter(lastSort));

            if (!response.IsValid || response.Documents.Count == 0)
            {
                break;
            }
            // 添加当前批次结果
            result.AddRange(response.Documents);
            // 更新排序值为当前批次最后一条的排序值
            lastSort = response.Hits.Last().Sort.ToList();
        }
    }
    finally
    {
        // 关闭PIT
        await _elasticClient.ClosePointInTimeAsync(d => d.Id(pitId));
    }

    return result;
}

内容的提问来源于stack exchange，提问作者MarkCo