基于Spring Data Jest的Elasticsearch数据统计示例

命令查询职责分离模式(command query responsibility segregation，cqrs)从业务上分离修改 (command，增，删，改，会对系统状态进行修改)和查询（query，查，不会对系统状态进行修改)的行为。从而使得逻辑更加清晰，便于对不同部分进行针对性的优化。

cqrs有以下几点有点：

1.分工明确，可以负责不同的部分；
2.将业务上的命令和查询的职责分离能够提高系统的性能、可扩展性和安全性。并且在系统的演化中能够保持高度的灵活性，能够防止出现crud模式中，对查询或者修改中的某一方进行改动，导致另一方出现问题的情况；
3.逻辑清晰，能够看到系统中的那些行为或者操作导致了系统的状态变化；
4.可以从数据驱动(data-driven) 转到任务驱动(task-driven)以及事件驱动(event-driven)。
因此command使用数据库，query使用效率查询效率更高的elasticsearch。

如何确保数据库和elasticsearch的数据的一致性？

我们可以使用事件驱动(event-driven)即spring data的domain event同步数据，可参考文章： 。

当老数据库有大量数据需要导入elasticsearch时，可参考文章：

spring data elasticsearch使用的是transport client，而elasticsearch官网推荐使用rest client。阿里云的elasticsearch使用transport client目前还在存在问题，阿里云推荐使用rest client。

本示例使用的是spring data jest链接elasticsearch（目前只有spring boot2.0以上版本支持），elasticsearch的版本为：5.5.3

1.项目构建

1.pom依赖如下：

<dependency>
  <groupid>com.github.vanroy</groupid>
  <artifactid>spring-boot-starter-data-jest</artifactid>
  <version>3.0.0.release</version>
</dependency>

<dependency>
  <groupid>io.searchbox</groupid>
  <artifactid>jest</artifactid>
  <version>5.3.2</version>
</dependency>

2.配置文件

spring:
 data:
  jest:
   uri: http://127.0.0.1:9200
   username: elastic
   password: changeme

2.构造查询条件

以简单的实体类为例

package com.hfcsbc.esetl.domain;
import lombok.data;
import org.springframework.data.elasticsearch.annotations.document;
import org.springframework.data.elasticsearch.annotations.field;
import org.springframework.data.elasticsearch.annotations.fieldtype;
import javax.persistence.entity;
import javax.persistence.id;
import javax.persistence.onetoone;
import java.util.date;
import java.util.list;

/**
 * create by pengchao on 2018/2/23
 */
@document(indexname = "person", type = "person", shards = 1, replicas = 0, refreshinterval = "-1")
@entity
@data
public class person {
  @id
  private long id;
  private string name;
  @onetoone
  @field(type = fieldtype.nested)
  private list<address> address;
  private integer number;
  private integer status;
  private date birthday;
}

package com.hfcsbc.esetl.domain;
import lombok.data;
import javax.persistence.entity;
import javax.persistence.id;

/**
 * create by pengchao on 2018/2/23
 */
@entity
@data
public class address {
  @id
  private long id;
  private string name;
  private integer number;
}

1.根据多个状态查询(类似于sql的in)

boolquerybuilder orderstatuscondition = querybuilders.boolquery()
    .should(querybuilders.termquery("status", 1))
    .should(querybuilders.termquery("status", 2))
    .should(querybuilders.termquery("status", 3))
    .should(querybuilders.termquery("status", 4))
    .should(querybuilders.termquery("status", 5));

2.and链接查询(类似于sql的and)

boolquerybuilder querybuilder = querybuilders.boolquery();
querybuilder
    .must(querybuilder1)
    .must(querybuilder2)
    .must(querybuilder3);

3.range查询（类似于sql的between .. and ..)

querybuilder rangequery = querybuilders.rangequery("birthday").from(yesterday).to(today);

4.嵌套对象查询

querybuilder querybuilder = querybuilders.nestedquery("nested", querybuilders.termquery("address.id", 100001), scoremode.none);

scoremode: 定义other join side中score是如何被使用的。如果不关注scoring，我们只需要设置成scoremode.none，此种方式会忽略评分因此会更高效和节约内存

3.获取统计数据

1.非嵌套获取数据求和

sumaggregationbuilder sumbuilder = aggregationbuilders.sum("sum").field("number");
searchquery searchquery = new nativesearchquerybuilder()
    .withindices(query_index)
    .withtypes(query_type)
    .withquery(boolquerybuilder)
    .addaggregation(sumbuilder).build();
    
aggregatedpage<parkingorder> account = (aggregatedpage<parkingorder>) esparkingorderrepository.search(esquerybuilders.buildyesterdayarrearssumquery(employeeid));

int sum = account.getaggregation("sum", sumaggregation.class).getsum().intvalue();

2.嵌套数据求和

sumaggregationbuilder sumbuilder = aggregationbuilders.sum("sum").field("adress.num");
aggregationbuilder aggregationbuilder = aggregationbuilders.nested("nested", "adress").subaggregation(sumbuilder);
searchquery searchquery = new nativesearchquerybuilder()
    .withindices(query_index)
    .withtypes(query_type)
    .withquery(boolquerybuilder)
    .addaggregation((abstractaggregationbuilder) aggregationbuilder).build();
aggregatedpage<parkingorder> account = (aggregatedpage<parkingorder>) esparkingorderrepository.search(esquerybuilders.buildyesterdayarrearssumquery(employeeid));
int sum = account.getaggregation("nested", sumaggregation.class).getaggregation("sum", sumaggregation.class).getsum().intvalue();

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。