详解 Elasticsearch Index Sorting 原理

﹏ヽ暗。殇╰゛Y 2022-12-12 04:52 227阅读 0赞

### Elasticsearch Index Sorting 原理 ###

Elasticsearch 是一款搜索引擎，它使用倒排索引来通过分词去检索数据，倒排索引里面的数据（`docID`）是有顺序的，默认是写入顺序，在大部分情况下，当检索数据时，都需要遍历倒排索引里的所有`docID`，然后对当前`document`的数据做解析。

当我希望对检索的数据按某个`Field做排序`，且仅仅取结果集的`TopN`时，还是需要遍历倒排索引里的所有数据。因为每个document的Field是无序的，只有遍历了所有的document时才知道哪些是TopN的。

这样就有很大的性能问题，毕竟当倒排索引的数据量很大时，遍历处理过程需要耗费很长的时间，比如`取指定地区订单数额最大的10笔订单`，那么可能就要遍历指定地区的所有订单，有没有什么能有提高性能的方式？ 有！！！ 这就是 `Index Sorting`

Index Sorting，顾名思义，就是对索引里的数据做排序，不仅仅是倒排索引，Segment里的所有数据都是，包括正向索引，docValues，倒排索引等等。使这些数据结构里的数据不再按照原来的写入顺序排序，而是按照指定的Field，ASC/DESC 形式来排序，这样当我们`检索时的排序方式契合Index Sorting的形式`，且`不统计命中总数`，`不计算最大评分` ，仅仅`取TopN的数据`时，就能触发 `Early Termination` 机制，不再需要遍历所有的document，`仅仅按顺序处理，当符合要求的数据达到N时，直接结束。`

### Index Sorting使用方式 ###

Index Sorting需要在创建索引时在Mappings和Settings里指定，具体使用方式如下：

#### 单Field排序 ####

PUT events
    { 
        "settings" : { 
            "index" : { 
                "sort.field" : "timestamp",
                "sort.order" : "desc" 
            }
        },
        "mappings": { 
            "properties": { 
                "timestamp": { 
                    "type": "date"
                }
            }
        }
    }

#### 多Field排序 ####

PUT events
    { 
        "settings" : { 
            "index" : { 
                "sort.field" : ["timestamp","age"],
                "sort.order" : ["desc","desc"] 
            }
        },
        "mappings": { 
            "properties": { 
                "timestamp": { 
                    "type": "date"
                },
                "age":{ 
                    "type": "integer"
                }
            }
        }
    }

当通过配置 `index.sort.field`，`index.sort.order` 两个选项后，Elasticsearch在Refresh时，会`对当前内存新写入的数据根据这两个配置项作重新排序`，然后生成Segment。这几个Field构成的排序机制就类似于Mysql的`聚集索引`，且可以是`复合聚集索引`。当搜索的数据是`按照主键索引排序`，且`不是全表查询`，`不统计总数`时，就能发挥主键索引的所用，能极大的提高检索效率。

#### 检索请求体 ####

GET /events/_search
    { 
        "size": 10,
        "sort": [ 
            {  "timestamp": "desc" }
        ],
        "track_total_hits": false
    }

### Elasticsearch 源码简析 ###

#### Index Sorting的Settings ####

public final class IndexSortConfig { 
 /** * The list of field names */
 public static final Setting<List<String>> INDEX_SORT_FIELD_SETTING =
 Setting.listSetting("index.sort.field", Collections.emptyList(),
 Function.identity(), Setting.Property.IndexScope, Setting.Property.Final);
 
 /** * The {@link SortOrder} for each specified sort field (ie. asc or desc). */
 public static final Setting<List<SortOrder>> INDEX_SORT_ORDER_SETTING =
 Setting.listSetting("index.sort.order", Collections.emptyList(),
 IndexSortConfig::parseOrderMode, Setting.Property.IndexScope, Setting.Property.Final);
 
 
 /** * The {@link MultiValueMode} for each specified sort field (ie. max or min). */
 public static final Setting<List<MultiValueMode>> INDEX_SORT_MODE_SETTING =
 Setting.listSetting("index.sort.mode", Collections.emptyList(),
 IndexSortConfig::parseMultiValueMode, Setting.Property.IndexScope, Setting.Property.Final);
 
 /** * The missing value for each specified sort field (ie. _first or _last) */
 public static final Setting<List<String>> INDEX_SORT_MISSING_SETTING =
 Setting.listSetting("index.sort.missing", Collections.emptyList(),
 IndexSortConfig::validateMissingValue, Setting.Property.IndexScope, Setting.Property.Final);
 }

上述4个Settings对应着 Index Sorting的4个配置：

*  **index.sort.field**：对应排序的Field，可指定一个或者多个Field，优先按第一个排序，相同的情况下在按后续的Field排序
 *  **index.sort.order**：对应Field的排序规则，只能时ASC或者时DESC
 *  **index.sort.mode**：当对应的Field是数组时，取Max或者Min的值作为排序的基准
 *  **index.sort.missing**：当对应的Field为null时，排第一个还是最后一个

#### 通过settings构建Field Sort配置 ####

// visible for tests
 final FieldSortSpec[] sortSpecs;
 
 public IndexSortConfig(IndexSettings indexSettings) { 
 final Settings settings = indexSettings.getSettings();
 // 提取Field配置
 List<String> fields = INDEX_SORT_FIELD_SETTING.get(settings);
 this.sortSpecs = fields.stream()
 .map((name) -> new FieldSortSpec(name))
 .toArray(FieldSortSpec[]::new);
 		// 提取Order配置
 if (INDEX_SORT_ORDER_SETTING.exists(settings)) { 
 List<SortOrder> orders = INDEX_SORT_ORDER_SETTING.get(settings);
 ......
 for (int i = 0; i < sortSpecs.length; i++) { 
 sortSpecs[i].order = orders.get(i);
 }
 }
 		// 提取Mode配置
 if (INDEX_SORT_MODE_SETTING.exists(settings)) { 
 List<MultiValueMode> modes = INDEX_SORT_MODE_SETTING.get(settings);
 ......
 for (int i = 0; i < sortSpecs.length; i++) { 
 sortSpecs[i].mode = modes.get(i);
 }
 }
 		// 提取Missing配置
 if (INDEX_SORT_MISSING_SETTING.exists(settings)) { 
 List<String> missingValues = INDEX_SORT_MISSING_SETTING.get(settings);
 			......
 for (int i = 0; i < sortSpecs.length; i++) { 
 sortSpecs[i].missingValue = missingValues.get(i);
 }
 }
 }

#### 提取ES配置，生成Lucene的Sort ####

public Sort buildIndexSort(Function<String, MappedFieldType> fieldTypeLookup,
 Function<MappedFieldType, IndexFieldData<?>> fieldDataLookup) { 
 if (hasIndexSort() == false) { 
 return null;
 }
 
 final SortField[] sortFields = new SortField[sortSpecs.length];
 for (int i = 0; i < sortSpecs.length; i++) { 
 FieldSortSpec sortSpec = sortSpecs[i];
 final MappedFieldType ft = fieldTypeLookup.apply(sortSpec.field);
 if (ft == null) { 
 throw new IllegalArgumentException("unknown index sort field:[" + sortSpec.field + "]");
 }
 boolean reverse = sortSpec.order == null ? false : (sortSpec.order == SortOrder.DESC);
 MultiValueMode mode = sortSpec.mode;
 if (mode == null) { 
 mode = reverse ? MultiValueMode.MAX : MultiValueMode.MIN;
 }
 IndexFieldData<?> fieldData;
 try { 
 fieldData = fieldDataLookup.apply(ft);
 } catch (Exception e) { 
 throw new IllegalArgumentException("docvalues not found for index sort field:[" + sortSpec.field + "]");
 }
 if (fieldData == null) { 
 throw new IllegalArgumentException("docvalues not found for index sort field:[" + sortSpec.field + "]");
 }
 sortFields[i] = fieldData.sortField(sortSpec.missingValue, mode, null, reverse);
 validateIndexSortField(sortFields[i]);
 }
 // 生成Lucene的Sort，在创建 IndexWriter 时使用
 return new Sort(sortFields);
 }

### Lucene源码解析 ###

#### 写入时源码 ####

// To store an index on disk, use this instead，此目录里的文件列表格式可见上一篇博客
    Directory directory = FSDirectory.open(Paths.get("D:\\lucene-data"));
    IndexWriterConfig config = new IndexWriterConfig(new WhitespaceAnalyzer());
    
    // 设置document在segment里的顺序, 默认是docId, 如果设置成某个Field或者多个Field, 则在检索时能够实现EarlyTerminate
    Sort sort = new Sort(new SortField("timestamp", Type.LONG, true), new SortField("age", Type.INT, false));
    config.setIndexSort(sort);
    
    IndexWriter indexWriter = new IndexWriter(directory, config);
    
    return indexWriter;

`org.apache.lucene.index.IndexWriter` 是Lucene在Index数据时的处理类，在生成此对象时，可以指定Index Sort，也就是倒排索引顺序，默认是Index Order，也就是索引顺序。当重新指定Sort时，生成的Segment里的数据顺序就是Sort里指定的顺序了。

#### 检索时源码 ####

private static class SimpleFieldCollector extends TopFieldCollector { 
 	......
 // 获取单个Segment的Collector
 public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException { 
 docBase = context.docBase;
 
 final LeafFieldComparator[] comparators = queue.getComparators(context);
 final int[] reverseMul = queue.getReverseMul();
 // Index Sorting 配置
 final Sort indexSort = context.reader().getMetaData().getSort();
 		// 是否可以提前结束
 final boolean canEarlyTerminate = trackTotalHits == false	// 不统计总数
 && trackMaxScore == false								// 不统计最大评分
 && indexSort != null // Document默认在写入时是没有顺序的，默认是Index顺序, 如果在在写入时是有Sort的, 则indexSort != null
 && canEarlyTerminate(sort, indexSort); // 检索的Sort和Index的Sort是一致的, 则可以提前结束
 
 
 final int initialTotalHits = totalHits;
 
 return new MultiComparatorLeafCollector(comparators, reverseMul, mayNeedScoresTwice) { 
 
 /** 根据倒排索引里的docID顺序，依次查看评分，默认按照Sort排序取最高的TopN的docID; 未指定Sort则默认是Score * @param doc document在segment里的序号，从0开始，全局序号加上docBase * @throws IOException */
 @Override
 public void collect(int doc) throws IOException { 
 ......
 ++totalHits;
 // 队列满了, 假设取Top10, 也就是已经取到了10条合适的数据
 if (queueFull) { 
 	// 对比当前doc和已经收集到的queue里的bottom的doc的Sort, Score太小或者Field值顺序不合适
 if (reverseMul * comparator.compareBottom(doc) <= 0) { 
 // since docs are visited in doc Id order, if compare is 0, it means
 // this document is largest than anything else in the queue, and
 // therefore not competitive.
 // 如果可以提前结, 收集的数据量足够了，那么提早结束能极大提高性能, es的Index Sorting 使用了这个特性
 if (canEarlyTerminate) { 
 // scale totalHits linearly based on the number of docs
 // and terminate collection
 totalHits += estimateRemainingHits(totalHits - initialTotalHits, doc, context.reader().maxDoc());
 earlyTerminated = true;
 // 抛出异常形式结束整个循环
 throw new CollectionTerminatedException();
 } else { 
 // just move to the next doc
 return;
 }
 }
 
 ......
 } 
 				......
 };
 }
 
 }

通过倒排索引检索数据，判断当前document是否应该被纳入候选者的queue时，如果有Sort，则要判断是通过`关联性评分Score，还是Field值来排序`。假设要取Top10，当queue里已经取了10条数据，且下一条数据的Score或者Field值都比queue里的小，则判断是否能够`Early Terminate`，如果可以，那么就不需要在遍历剩下的所有数据，抛出异常中断整个循环，提前结束。 当倒排索引数据量很大时，能够极大的提高性能。

### 总结 ###

Elasticsearch Index Sorting 是通过`调整Segment里的数据顺序来契合检索时的顺序`，以达到在`检索时能够按照Sort里的顺序来对document做处理`，当取到足量的满足条件的数据时 `Early Terminate`，忽略后续所有的document，当指定Field值的倒排索引的数据量很大时，能够极大的提升检索时的效率。