深入实战：ElasticSearch的Rest API与迭代器模式在高效查询中的应用

在我们公司，大多数Java开发工程师在项目中都有使用Elasticsearch的经验。通常，他们会通过引入第三方工具包或使用Elasticsearch Client等方式来进行数据查询。然而，当涉及到基于Elasticsearch Rest API的/_sql?format=json接口时，即使是有Elasticsearch使用经验的开发人员也可能感到困惑。这是因为在开发过程中，我们通常习惯于使用基于JSON定义的DSL语言，利用Elasticsearch的标准工具包、Query、Filter、termsQuery等方法，或使用scrollId来查询大量数据集。

在开发某个客户定制项目过程中，，客户提出了希望能够依据SQL查询设定的条件来执行数据查询的需求。鉴于此种情况，我们必须舍弃原先常用的DSL语言,而转向利用/_sql?format=json接口实行Elasticsearch数据检索。/_sql?format=json接口进行Elasticsearch数据查询的过程往往在整个项目设计上依赖于Elasticsearch Rest API。这种方式的挑战在于，开发者需要自行处理scrollId的迭代查询，因为没有第三方工具来自动封装这一过程。这意味着我们需要手动控制scrollId，每次查询最多10000条数据，并重复使用该接口直到获取全部所需数据。

本文将结合项目开发过程中的实际经验，详细介绍/_sql?format=json接口的调用机制和返回值格式，深入探讨迭代器模式在实际Elasticsearch查询中的应用。文章内容包括：

Elasticsearch SQL Rest API的/_sql?format=json调用机制及其返回值格式。
迭代器模式的实际应用：包括类结构分析、方法定义及Elasticsearch查询实例。

本篇文章的编排旨在将常见的设计模式中的“迭代器模式”与“Elasticsearch RestAPI 查询实战”结合起来。这样的安排虽然提高了代码理解的难度，但对于经验稍显不足的开发人员来说，将是一个极好的挑战和学习机会。

1 ElasticSearch SQL Rest API 机制介绍

1.1 SQL Rest API接口信息和入参

POST /_sql?format=json
{
  "query": "SELECT * FROM library ORDER BY page_count DESC",
  "fetch_size": 5
}

通过分析图示，我们可以详细地理解该API的工作方式。此API采用POST方法访问，其统一资源标识符（URI）设置为/_sql?format=json。在发送请求时，RequestBody主要包含两个关键属性：

"query"属性：其值为SQL语句。这里使用的SQL语句遵循标准的SQL语法规则，与MySQL的语法极为相似，使得熟悉传统数据库开发的工程师更容易上手。
"fetch_size"属性：这个值为数字类型，用于指定返回结果的限制数量，类似于SQL中的LIMIT子句。

此外，该API允许通过format=json参数来指定返回数据的格式。默认情况下，这一参数设置返回格式为JSON，但API同样支持其他格式，如CSV、TSV、TEXT、YAML、CBOR和SMILE。在我们的项目实践中，JSON格式因其易于解析和通用性而被频繁使用。

1.2 SQL Rest API返回值

{
  "columns": [
    {"name": "author",       "type": "text"},
    {"name": "name",         "type": "text"},
    {"name": "page_count",   "type": "short"},
    {"name": "release_date", "type": "datetime"}
  ],
  "rows": [
    ["Peter F. Hamilton",  "Pandora's Star",       768, "2004-03-02T00:00:00.000Z"],
    ["Vernor Vinge",       "A Fire Upon the Deep", 613, "1992-06-01T00:00:00.000Z"],
    ["Frank Herbert",      "Dune",                 604, "1965-06-01T00:00:00.000Z"],
    ["Alastair Reynolds",  "Revelation Space",     585, "2000-03-15T00:00:00.000Z"],
    ["James S.A. Corey",   "Leviathan Wakes",      561, "2011-06-02T00:00:00.000Z"]
  ],
  "cursor": "sDXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAAEWWWdrRlVfSS1TbDYtcW9lc1FJNmlYdw==:BAFmBmF1dGhvcgFmBG5hbWUBZgpwYWdlX2NvdW50AWYMcmVsZWFzZV9kYXRl+v///w8="
}

分析API的响应内容，我们可以明确地识别出其由三个主要部分组成：

columns：这部分包含了所有返回字段的名称和类型。它为数据解析和后续操作提供了必要的结构信息。
rows：此部分包含了查询结果的具体值，其排列顺序与columns部分中定义的字段顺序严格对应。这种一致性确保了数据的完整性和易用性。
cursor：这是实现分页功能的关键元素。cursor的存在表明，当前返回的数据集只是满足查询条件的一部分，由于fetch_size的设置，初次响应只包含了限定数量的数据。要访问后续的数据页，我们需要将cursor值回传至API。这种机制允许高效地遍历大量数据，而不必一次性加载全部结果。

1.3 回传cursor，获取其他的分页

继续使用前述数据，若我们需要访问查询结果的第二页或第三页，可以简单地将cursor值用作RequestBody，并再次调用相同的接口。这个过程遵循与首次查询相同的“配方”。

POST /_sql?format=json
{
  "cursor": "sDXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAAEWWWdrRlVfSS1TbDYtcW9lc1FJNmlYdw"
 }

在后续的响应中，API通常只返回rows和cursor属性。这是因为columns属性，即字段的名称和类型，已经在第一次响应中提供，无需重复返回。当我们到达查询结果的最后一页时，响应中将只包含rows属性，不再包含cursor，表明所有数据已被完全检索。

{
  "rows": [
    ["Peter F. Hamilton",  "Pandora's Star",       768, "2004-03-02T00:00:00.000Z"],
    ["Vernor Vinge",       "A Fire Upon the Deep", 613, "1992-06-01T00:00:00.000Z"],
    ["Frank Herbert",      "Dune",                 604, "1965-06-01T00:00:00.000Z"],
    ["Alastair Reynolds",  "Revelation Space",     585, "2000-03-15T00:00:00.000Z"],
    ["James S.A. Corey",   "Leviathan Wakes",      561, "2011-06-02T00:00:00.000Z"]
  ]
}

例如，若要查询总共49000条数据，流程大致如下：

首次查询：获取所有字段（columns）和首批10000条数据（rows），同时获得一个cursor值。
第二次查询：使用已获得的cursor值作为RequestBody，检索下一批10000条数据，并再次获得cursor值。
第三次和第四次查询：重复第二次查询的步骤。
第五次查询：最后获取剩余的9000条数据，此时响应不再包含cursor，表示查询已完结。

在编写对应的代码逻辑时，可以考虑使用递归或者while循环来判断cursor值是否为null，从而决定是否继续查询。还可能有其他编程方法可用于实现这一逻辑。

2 迭代器模式实战

2.1 UML类结果分解、方法定义以及实战

迭代器模式是一种常用的设计模式，其主要目的是对数据结构中的所有元素进行逐一遍历，直到所有元素均被访问一次。大多数Java开发人员在学习Java SE时，通过List数据结构就已经接触到了迭代器的概念。利用List的迭代器遍历列表元素通常是一项基本且简单的任务。
在这里插入图片描述

我们将首先学习迭代器模式的UML类图，然后针对每个角色进行具体类的创建和方法的定义。迭代器模式的UML类图主要包含四个角色，但我们只需要创建其中的三个：

Iterator（抽象迭代器） ：定义了访问和遍历元素的接口。
ConcreteIterator（具体迭代器） ：实现迭代器接口，负责完成对容器元素的实际遍历。
Aggregate（抽象容器） ：提供创建具体迭代器对象的接口。
ConcreteAggregate（具体容器） ：实现了创建具体迭代器对象的方法。

在UML类图中，除Iterator（JDK的java.util.Iterator）外，我们需要实现其余三个角色。核心的逻辑主要集中在ConcreteIterator中。此外，我们还需要定义一个实体类来接收Elasticsearch SQL Rest API的响应数据，该实体类应包含columns、rows和cursor属性，并提供相应的getter和setter方法。接下来，我们将深入探讨UML类结构的分解和方法定义。

创建与ESSOL RestAPI返回值对应的实体对象一一EsResponseData
该对象并不是UML类图中的角色，但是要处理ES SQL Rest API 的返回值，此类必不可少。代码和注释如下：

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class EsResponseData {
   //所有的字段
   private List> columns;
   //返回的数据值
   private List> rows;
   //用于分页的 cursor 值
   private String cursor;
}

创建Aggregate抽象容器一-EsSqlQueryInterface
抽象容器角色负资提供创建具体迭代器角色的抽象方法，我们使用泛型T保证了该类的扩展性。我们定义的抽象 iterator 方法，是为了new 一个具体的选代器对象，当然了，这部分逻辑会在子类中进行实现。代码如下:

public interace EsSqlQueryInterface {
   public T iterator();
}

创建ConcreteAggregate具体容器-EsSqlQuery
具体容器实现容器接口定义的抽象方法，创建迭代器对象。代码及注释如下:

@Data
@JsonIgnoreProperties
public class EsSqlQuery implements EsSqlQueryInterface{
   private String query;
   private Long fetchSize;
   private String cursor;

   public EsSqlQuery(String cursor) {
       this.cursor = cursor;
   }

   public EsSqlQuery(String query, Long fetchSize) {
       this.query = query;
       this.fetchSize = fetchSize;
   }
   public EsQueryIterator iterator(){
       return new EsQueryIterator(this.query, this.fetchSize);
   }
}

创建ConcreteIterator具体迭代器—EsQueryIterator
此处代码是核心的代码，需要实现 java.util.Iterator 接口，并覆写 hasNext 以及 next方法，同时需要添加自己的 scrolINext 方法用于判断 cursor 是否为 null.如果 cursor 为 null，则说明已经选代完成

public class EsQueryIterator implements Iterator> {
   //记录当前cursor分页
   private String cursor;
   //记录查询的columns，因为只有第一次查询才会返回columns数据
   private List columns;
   //将ES SQL Rest API的返回值封装到List中，以便处理返回值
   Iterator> iterator;
   //此处我们从简而行，不再进行@Autowire注入，把更多的精力放到迭代器模式中
   RestTemplate restTemplate = new RestTemplate();

   //构造函数进行第一次查询，并且初始化我们后续需要使用的 columns 和 iterator 和 cursor
   public EsQueryIterator(String query, Long fetchSize) {
       EsResponseData esResponseData = restTemplate.postForObject("http://localhost:9200/_sql?format=json",
               new EsSqlQuery(query, fetchSize), EsResponseData.class);//第一次访问的结果出来了
       this.cursor = esResponseData.getCursor();
       this.columns = esResponseData.getColumns()
               .stream().map(x -> x.get("name"))
               .collect(Collectors.toList());
       this.iterator = convert(columns, esResponseData).iterator();
   }

   // hasNext 根据 是否 cursor 为null进行后续的 第二次，第三次，，，的访问，直到 cursor 为null
   @Override
   public boolean hasNext() {
       return iterator.hasNext() || scrollNext();
   }

   //获取第二次及以后的查询结果
   private boolean scrollNext() {
       if (iterator == null || this.cursor == null) {
           return false;
       }
       EsResponseData esResponseData = restTemplate.postForObject("http://localhost:9200/_sql?format=json",
               new EsSqlQuery(this.cursor), EsResponseData.class);
       this.cursor = esResponseData.getCursor();
       this.iterator = convert(columns, esResponseData).iterator();
       return iterator.hasNext();
   }

   @Override
   public Map next() {
       return iterator.next();
   }
   //将 ES SQL Rest API的返回值转化为 List
   private List> convert(List columns, EsResponseData esResponseData) {
       List> results = new ArrayList<>();
       for (List row : esResponseData.getRows()) {
           Map map = new HashMap<>();
           for (int i = 0; i < columns.size(); i++) {
               map.put(columns.get(i), row.get(i));
           }
           results.add(map);
       }
       return results;
   }
}

2.2 实战测试

接下来，我们进行迭代器模式的实战测试。测试过程并不复杂，我们会创EsQueryController 和 EsQueryService 类，大家可以更关注 EsOueryService 类的方法，此处我们会使用 Stream 和 Spliterators，可能部分开发未使用过 Spliterators，但是代码不复杂，非常容易理解。

创建EsQueryController和EsQueryService代码如下:

@RestController
public class EsQueryController {
    @Autowired
    private EsQueryService esQueryService;
    @PostMapping("/queryEsBySql")
    public Object queryEsBySql(@RequestBody EsSqlQuery esSqlQuery) {
        return esQueryService.queryEsBySql(esSqlQuery);
    }
}

@Service
public class EsQueryService {
    public Object queryEsBySql(EsSqlQuery esSqlQuery) {
        EsQueryIterator iterator = esSqlQuery.iterator();
        Stream> resultStream = StreamSupport.stream(Spliterators
                .spliteratorUnknownSize(iterator, 0), false);
        return resultStream.collect(Collectors.toList());
    }
}

通过PostMan 请求对应的数据

总结

本文深入探讨了Elasticsearch SQL Rest API及迭代器模式在高效数据查询中的应用。文章介绍了使用Elasticsearch的/_sql?format=json接口进行数据查询的机制，详细讨论了迭代器模式的实现，包括其在Elasticsearch查询中的具体应用。通过介绍UML类图和相关的类结构，解释了如何创建和应用不同的迭代器角色，如抽象迭代器、具体迭代器和抽象容器等。提供了实际的代码示例，以展示如何在实践中使用迭代器模式高效遍历和管理Elasticsearch的查询结果。

其次，文章详细讨论了迭代器模式的实现，包括其在Elasticsearch查询中的具体应用。通过介绍UML类图和相关的类结构，作者清晰地解释了如何创建和应用不同的迭代器角色，如抽象迭代器、具体迭代器和抽象容器等。特别地，文章提供了实际的代码示例，以展示如何在实践中使用迭代器模式高效遍历和管理Elasticsearch的查询结果。

参考文章：