在我们公司,大多数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 和 cursorpublic 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@Overridepublic 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();}@Overridepublic Map next() {return iterator.next();}//将 ES SQL Rest API的返回值转化为 Listprivate 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 {@Autowiredprivate 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的查询结果。
参考文章:
- [迭代器模式 | 菜鸟教程]
- Response Data Formats | Elasticsearch Guide [7.17] | Elastic