聚合概念
聚合就相当于是数据库中的分组（GROUP BY） 但是他比GROUP BY更加的强大 \
聚合类型三大类
Bucketing（桶聚合）
- Date Histogram Aggregation：根据日期阶梯分组，例如给定阶梯为周，会自动每周分为一组
- Histogram Aggregation：根据数值阶梯分组，与日期类似
- Terms Aggregation：根据词条内容分组，词条内容完全匹配的为一组
- Range Aggregation：数值和日期的范围分组，指定开始和结束，然后按段分组
- Missing Aggregation：统计文档中缺失字段的数量，缺失字段包含值为null的情况
- Filter Aggregation：对经过Filter条件过滤后的结果集进行聚合查询
每个桶都与一个键和一个文档标准相关联，通过桶的聚合查询，我们将得到一个桶的列表，即：满足条 件的文档集合。是按照某种方式对数据进行分组 比如对1，2，3，1,3使用Terms对其聚合，可以得到1桶，2桶，3桶。
看出ES的分组方式相当强大，mysql的group by只能实现类似Terms Aggregation的分组效果，而ES还可以根据阶梯和范围来分组。
Pipeline（管道）
对其他聚合的输出或相关指标进行二次聚合 比如Terms聚合后拿到了1，2，3桶这个时候我们可以在对他其他的属性进行聚合 也就是对结果在聚合 和数据库中 多字段分组一个意思（多字段分组可以使用script脚本去聚合需要添加配置 但是我试了下没什么用 如果有那个大神有好的解决办法可以分享出来 我这边用的是管道聚合也就是聚合套聚合 或者是在洗数据的时候拼接好分组字段）
Metric（指标）
指标聚合类似于 COUNT() 、 SUM() 、 MAX() 等统计方法
- Avg Aggregation：求平均值
- Max Aggregation：求最大值
- Min Aggregation：求最小值
- Percentiles Aggregation：求百分比
- Stats Aggregation：同时返回avg、max、min、sum、count等
- Sum Aggregation：求和
- Top hits Aggregation：求前几
- Value Count Aggregation：求总数
Term Aggregation
GET pre_package/_search
{
"aggs": {
"Group": {
"terms": {
"field": "productGroupId",
"size": 10
}
}
}
}
这个表示，查询索引为pre_package中的文档数据，并按照cargoOwnerName进行聚合查询，命名为：Group，且只查询前10条
Range Aggregation
GET pre_package/_search
{
"aggs": {
"Group": {
"range": {
"field": "isPrePackage",
"ranges": [
{
"to": 1
},
{
"from": 1,
"to": 2
},
{
"from": 2
}
]
}
}
}
}
按照isPrePackage属性，分为三档，分别为：小于1，1到2，大于2
Date Range Aggregation
GET pre_package/_search
{
"aggs": {
"Group": {
"date_range": {
"field": "update_date",
"ranges": [
{
"to": "2020-05-01 00:00:00"
},
{
"from": "2020-05-02 00:00:00",
"to": "2020-08-01 00:00:00"
},
{
"from": "2020-08-02 00:00:00"
}
]
}
}
}
}
基于时间范围的聚合查询
Filter Aggregation
GET pre_package/_search
{
"aggs": {
"flight_Miles": {
"filter": {
"term": {
"cargoOwnerName": "联合利华测试"
}
}
}
}
}
对经过Filter条件过滤后的结果集进行聚合查询
Missing Aggregation
GET pre_package/_search
{
"aggs": {
"without_age": {
"missing": {
"field": "orderTypeName"
}
}
}
}
统计文档中缺失字段的数量，缺失字段包含值为null的情况
Histogram Aggregation
GET pre_package/_search
{
"aggs": {
"test": {
"histogram": {
"field": "id",
"interval": 100
}
}
}
}
直方图聚合，可按照一定的区间进行统计
ElasticsearchRestTemplate操作帮助类
package com.xxl.job.executor.utils;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.xxl.job.executor.es.EsSortPage;
import com.xxl.job.executor.es.QueryCondition;
import com.xxl.job.executor.es.TermQueryReq;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.AvgAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedAvg;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.*;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Component;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @description: es操作类(ElasticsearchRestTemplate)
* @Author:
* @NAME: ElasticsearchRestEsUtils
* @date: 2022/7/1 14:17
*/
@Component
public class ElasticsearchRestEsUtils {
@Autowired
private ElasticsearchRestTemplate elasticsearchTemplate;
/**
* @description: 创建索引
* @author:
* @date 2022/7/1 14:07
* @param: [cla]
* @return: boolean
*/
public boolean createIndexOps(Class cla) {
// 索引别名
IndexOperations ops = elasticsearchTemplate.indexOps(cla);
if (!ops.exists()) {
ops.create();
ops.refresh();
ops.putMapping(ops.createMapping());
}
return true;
}
/**
* @description: 删除索引
* @author:
* @date 2022/7/1 15:23
* @param: [cla]
* @return: void
*/
public void deleteIndex(Class cla) {
// 这里使用了 restTemplate 的 indexOps() 获取 IndexOperations 对象操作索引
// dao 所提供的方法不支持操作索引
boolean delete = elasticsearchTemplate.indexOps(cla).delete();
System.out.println("delete = " + delete);
}
/**
* @description: 分页查询全部
* @author:
* @date 2022/7/1 15:37
* @param: [esSortPage]
* @return: java.util.List<?>
*/
public List<?> findByPageable(EsSortPage esSortPage) {
List<Object> relist = new ArrayList<>();
// 设置排序(排序方式，正序还是倒序，排序的 id)
Sort sort = null;
if (esSortPage.getSort() == 1) {
sort = Sort.by(Sort.Direction.DESC, esSortPage.getSortField());
} else {
sort = Sort.by(Sort.Direction.ASC, esSortPage.getSortField());
}
int currentPage = esSortPage.getCurrentPage();
int pageSize = esSortPage.getPageSize();
// 设置查询分页
PageRequest pageRequest = PageRequest.of(currentPage, pageSize, sort);
ElasticsearchRepository<?, Long> dao = esSortPage.getDao();
//分页查询
Page<?> daoAll = dao.findAll(pageRequest);
List<?> content = daoAll.getContent();
for (Object o : content) {
relist.add(o);
}
return relist;
}
/**
* @description: 根据组装条件分页查询 会根据类型去匹配全等还是分词模糊查询
* @author:
* @date 2022/7/3 11:07
* @param: [termQueryReq]
* @return: java.util.List<?>
*/
public List<?> findByTermQuery(TermQueryReq termQueryReq) {
List<Object> relist = new ArrayList<>();
// 设置排序(排序方式，正序还是倒序)
FieldSortBuilder balance = null;
if (termQueryReq.getSort() == 1) {
balance = new FieldSortBuilder(termQueryReq.getSortField()).order(SortOrder.DESC);
} else {
balance = new FieldSortBuilder(termQueryReq.getSortField()).order(SortOrder.ASC);
}
//分页条件
int currentPage = termQueryReq.getCurrentPage();
int pageSize = termQueryReq.getPageSize();
// 构建查询条件
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
;
List<QueryCondition> mapList = termQueryReq.getMapList();
for (QueryCondition queryCondition : mapList) {
if (queryCondition.getFiltration() == 1) {//过滤
RangeQueryBuilder date = QueryBuilders.rangeQuery(queryCondition.getKey()).gte(queryCondition.getValue()).lte(queryCondition.getEndValue());
boolQueryBuilder.filter(date);
} else {
if (queryCondition.getExtractive() == 1) {
if (queryCondition.getValue() != null) {
boolQueryBuilder.must(QueryBuilders.matchQuery(queryCondition.getKey(), queryCondition.getValue()).minimumShouldMatch(termQueryReq.getScore()));//多条件全匹配 and 因为text会分词匹配 所以得设置一个百分比得分 越高月精确
}
} else {
if (queryCondition.getValue() instanceof Integer[]) {
Integer[] pulldown = (Integer[]) queryCondition.getValue();
for (Integer integer : pulldown) {
boolQueryBuilder.should(QueryBuilders.matchQuery(queryCondition.getKey(), integer));//多条件匹配之一 or
}
}
}
}
}
// 分页
Pageable pageable = PageRequest.of(currentPage, pageSize);
// 执行查询
NativeSearchQuery query = new NativeSearchQueryBuilder()
.withQuery(boolQueryBuilder)
.withPageable(pageable)
.withSort(balance)
.build();
List<Object> resList = new ArrayList<>();
SearchHits<?> search = elasticsearchTemplate.search(query, termQueryReq.getCls());
for (SearchHit<?> searchHit : search) {
resList.add(searchHit.getContent());
}
return resList;
}
/**
* @description: 聚合搜索 类似于group by，对termQueryReq.getAggregate()字段进行聚合，
* @author:
* @date 2022/7/3 11:32
* @param: [termQueryReq]
* @return: java.util.List<?>
*/
public List<Map<String, Object>> findPolymerization(TermQueryReq termQueryReq) {
NativeSearchQuery query = new NativeSearchQueryBuilder()
.addAggregation(AggregationBuilders.terms("count").field(termQueryReq.getAggregate() + ".keyword"))
.build();
SearchHits<?> searchHits = elasticsearchTemplate.search(query, termQueryReq.getCls());
//取出聚合结果
Aggregations aggregations = searchHits.getAggregations();
Terms terms = (Terms) aggregations.asMap().get("count");
List<Map<String, Object>> mapList = new ArrayList<>();
for (Terms.Bucket bucket : terms.getBuckets()) {
Map<String, Object> map = new HashMap<>();
String keyAsString = bucket.getKeyAsString(); // 聚合字段列的值
long docCount = bucket.getDocCount(); // 聚合字段对应的数量
map.put("keyAsString", keyAsString);
map.put("docCount", docCount);
mapList.add(map);
}
return mapList;
}
/**
* @description: 嵌套聚合 统计出相同termQueryReq.getAggregate()的文档数量，再统计出termQueryReq.getNes()的平均值，带排序
* @author:
* @date 2022/7/3 11:36
* @param: [termQueryReq]
* @return: java.util.List<?>
*/
public List<Map<String, Object>> findNest(TermQueryReq termQueryReq) {
// 创建聚合查询条件
TermsAggregationBuilder stateAgg = AggregationBuilders.terms("count").field(termQueryReq.getAggregate() + ".keyword");
AvgAggregationBuilder balanceAgg = AggregationBuilders.avg("avg_" + termQueryReq.getNes()).field(termQueryReq.getNes());
// 嵌套
stateAgg.subAggregation(balanceAgg);
// 按balance的平均值降序排序
if (termQueryReq.getSort() == 1) {
stateAgg.order(BucketOrder.aggregation("avg_" + termQueryReq.getNes(), true));
} else {
stateAgg.order(BucketOrder.aggregation("avg_" + termQueryReq.getNes(), false));
}
NativeSearchQuery build = new NativeSearchQueryBuilder()
.addAggregation(stateAgg)
.build();
//执行查询
SearchHits<?> searchHits = elasticsearchTemplate.search(build, termQueryReq.getCls());
// 取出聚合结果
Aggregations aggregations = searchHits.getAggregations();
Terms terms = (Terms) aggregations.asMap().get("count");
List<Map<String, Object>> mapList = new ArrayList<>();
for (Terms.Bucket bucket : terms.getBuckets()) {
// state : count : avg
Map<String, Object> map = new HashMap<>();
ParsedAvg avg = bucket.getAggregations().get("avg_" + termQueryReq.getNes());
map.put("state", bucket.getKeyAsString());
map.put("count", bucket.getDocCount());
map.put("avg", avg.getValueAsString());
mapList.add(map);
}
return mapList;
}
/**
* @description: 过滤搜索
* @author:
* @date 2022/7/3 13:15
* @param: [termQueryReq]
* @return: java.util.List<?>
*/
public List<?> findFiltration(TermQueryReq termQueryReq) {
// 构建条件
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
RangeQueryBuilder balance = QueryBuilders.rangeQuery("balance").gte(20000).lte(30000);
boolQueryBuilder.filter(balance);
NativeSearchQuery query = new NativeSearchQueryBuilder()
.withQuery(boolQueryBuilder)
.build();
List<Object> resList = new ArrayList<>();
SearchHits<?> search = elasticsearchTemplate.search(query, termQueryReq.getCls());
for (SearchHit<?> searchHit : search) {
resList.add(searchHit.getContent());
}
return resList;
}
/**
* @description: 判断索引是否已存在
* @author:
* @date 2022/7/3 13:21
* @param: [indexName]
* @return: boolean
*/
public boolean indexExist(String indexName) {
if (StringUtils.isBlank(indexName)) {
return false;
}
IndexCoordinates indexCoordinates = IndexCoordinates.of(indexName);
return elasticsearchTemplate.indexOps(indexCoordinates).exists();
}
/**
* @description: 根据索引名称，删除索引
* @author:
* @date 2022/7/3 13:24
* @param: [index]
* @return: void
*/
public void indexDelete(String index) {
elasticsearchTemplate.indexOps(IndexCoordinates.of(index)).delete();
}
/**
* @description: 索引添加别名
* @author:
* @date 2022/7/3 13:25
* @param: [indexName, aliasName]
* @return: boolean
*/
public boolean indexAddAlias(String indexName, String aliasName) {
if (StringUtils.isBlank(indexName) || StringUtils.isBlank(aliasName)) {
return false;
}
// 索引封装类
IndexCoordinates indexCoordinates = IndexCoordinates.of(indexName);
// 判断索引是否存在
if (elasticsearchTemplate.indexOps(indexCoordinates).exists()) {
// 索引别名
AliasQuery query = new AliasQuery(aliasName);
// 添加索引别名
boolean bool = elasticsearchTemplate.indexOps(indexCoordinates).addAlias(query);
return bool;
}
return false;
}
/**
* @description: 索引别名删除
* @author:
* @date 2022/7/3 13:25
* @param: [indexName, aliasName]
* @return: boolean
*/
public boolean indexRemoveAlias(String indexName, String aliasName) {
if (StringUtils.isBlank(indexName) || StringUtils.isBlank(aliasName)) {
return false;
}
// 索引封装类
IndexCoordinates indexCoordinates = IndexCoordinates.of(indexName);
// 判断索引是否存在
if (elasticsearchTemplate.indexOps(indexCoordinates).exists()) {
// 索引别名
AliasQuery query = new AliasQuery(aliasName);
// 删除索引别名
boolean bool = elasticsearchTemplate.indexOps(indexCoordinates).removeAlias(query);
return bool;
}
return false;
}
/**
* @description: 索引新增数据
* @author:
* @date 2022/7/3 13:26
* @param: [t]
* @return: void
*/
public <T> void save(T t) {
// 根据索引实体名新增数据
elasticsearchTemplate.save(t);
}
/**
* @description: 批量插入数据
* @author:
* @date 2022/7/3 13:26
* @param: [queries, index]
* @return: void
*/
public void bulkIndex(List<IndexQuery> queries, String index) {
// 索引封装类
IndexCoordinates indexCoordinates = IndexCoordinates.of(index);
// 批量新增数据，此处数据，不要超过100m，100m是es批量新增的筏值，修改可能会影响性能
elasticsearchTemplate.bulkIndex(queries, indexCoordinates);
}
/**
* @description: 批量插入数据
* @author:
* @date 2022/7/3 13:26
* @param: [queries, index]
* @return: void
*/
public void bulkCls(List<IndexQuery> queries, Class cls) {
// 索引封装类
// 批量新增数据，此处数据，不要超过100m，100m是es批量新增的筏值，修改可能会影响性能
createIndexOps(cls);
elasticsearchTemplate.bulkIndex(queries, cls);
}
/**
* 根据条件删除对应索引名称的数据
*
* @param c 索引类对象
* @param filedName 索引中字段
* @param val 删除条件
* @param index 索引名
*/
public void delete(Class c, String filedName, Object val, String index) {
// 匹配文件查询
TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(filedName, val);
NativeSearchQuery nativeSearchQuery = new NativeSearchQuery(termQueryBuilder);
// 删除索引数据
elasticsearchTemplate.delete(nativeSearchQuery, c, IndexCoordinates.of(index));
}
/**
* 根据数据id删除索引
*
* @param id 索引id
* @param index
*/
public void deleteById(Object id, String index) {
if (null != id && StringUtils.isNotBlank(index)) {
// 根据索引删除索引id数据
elasticsearchTemplate.delete(id.toString(), IndexCoordinates.of(index));
}
}
/**
* 根据id更新索引数据,不存在则创建索引
*
* @param t 索引实体
* @param id 主键
* @param index 索引名称
* @param <T> 索引实体
*/
public <T> void update(T t, Integer id, String index) {
// 查询索引中数据是否存在
Object data = elasticsearchTemplate.get(id.toString(), t.getClass(), IndexCoordinates.of(index));
if (data != null) {
// 存在则更新
UpdateQuery build = UpdateQuery.builder(id.toString()).withDocument(Document.parse(JSON.toJSONString(t))).build();
elasticsearchTemplate.update(build, IndexCoordinates.of(index));
} else {
// 不存在则创建
elasticsearchTemplate.save(t);
}
}
/**
* 拼接推送数据信息
*
* @param obj
* @return
*/
public IndexQuery assembleDataEs(Object obj) {
//拼接数据
IndexQuery indexQuery = new IndexQuery();
try {
Field field = getDeclareField(obj, "id");
field.setAccessible(true);
indexQuery.setId(field.get(obj).toString());
} catch (IllegalAccessException e) {
e.printStackTrace();
}
indexQuery.setSource(JSON.toJSONString(obj));
return indexQuery;
}
/**
* 读取Field（含父类属性）
*
* @param o
* @param fieldName
* @return
*/
public static Field getDeclareField(Object o, String fieldName) {
Field field = null;
Class<?> clazz = o.getClass();
for (; field == null; clazz = clazz.getSuperclass()) {
try {
field = clazz.getDeclaredField(fieldName);
} catch (NoSuchFieldException e) {
//e.printStackTrace();
}
}
return field;
}
}
RestHighLevelClient操作帮助类
package com.xxl.job.executor.utils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.executor.es.QueryCondition;
import com.xxl.job.executor.es.TermQueryReq;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.Fuzziness;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
//import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
/**
* @description: es操作类(RestHighLevelClient)
* @Author:
* @NAME: EsUtils
* @date: 2022/7/1 13:39
*/
@Slf4j
@Component
public class RestHighLeveEsUtils {
@Autowired
private RestHighLevelClient restHighLevelClient;
/**
* @description: 创建索引
* @author:
* @date 2022/7/1 14:02
* @param: [indexName]
* @return: boolean
*/
public boolean createIndex(String indexName) {
//返回结果
boolean exists = true;
try {
// 1、创建索引请求
CreateIndexRequest request = new CreateIndexRequest(indexName);
// 2、客户端执行请求 indexResponse, 请求后获得相应
CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
//判断响应对象是否为空
if (createIndexResponse.equals("") || createIndexResponse != null) {
exists = false;
}
} catch (IOException e) {
exists = false;
}
return exists;
}
/**
* @description: 判断索引是否存在
* @author:
* @date 2022/7/1 14:03
* @param: [indexName]
* @return: boolean
*/
public boolean isIndexExists(String indexName) {
boolean exists = true;
try {
GetIndexRequest request = new GetIndexRequest(indexName);
exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
} catch (IOException e) {
exists = false;
}
return exists;
}
/**
* @description: 删除索引
* @author:
* @date 2022/7/1 14:03
* @param: [indexName]
* @return: boolean
*/
public boolean delIndex(String indexName) {
boolean exists = true;
try {
DeleteIndexRequest request = new DeleteIndexRequest(indexName);
AcknowledgedResponse delete = restHighLevelClient.indices().delete(request, RequestOptions.DEFAULT);
exists = delete.isAcknowledged();
} catch (IOException e) {
exists = false;
}
return exists;
}
/**
* @description: 更新文档的信息
* @author:
* @date 2022/7/1 14:28
* @param: [indexName, obj, id]
* @return: boolean
*/
public boolean updateDocument(String indexName, Object obj, String id) {
boolean exists = true;
try {
UpdateRequest updateRequest = new UpdateRequest(indexName, id);
updateRequest.timeout("1s");
updateRequest.doc(JSON.toJSONString(obj), XContentType.JSON);
UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
if (!updateResponse.status().equals("OK")) {
exists = false;
}
} catch (IOException e) {
e.printStackTrace();
}
return exists;
}
/**
* @description: 删除文档记录
* @author:
* @date 2022/7/1 14:28
* @param: [indexName, id]
* @return: boolean
*/
public boolean deleteRequest(String indexName, String id) {
boolean exists = true;
try {
DeleteRequest request = new DeleteRequest(indexName, id);
request.timeout("1s");
DeleteResponse delete = restHighLevelClient.delete(request, RequestOptions.DEFAULT);
if (!delete.status().equals("OK")) {
exists = false;
}
} catch (IOException e) {
e.printStackTrace();
}
return exists;
}
/**
* @description: 根据id获取文档信息
* @author:
* @date 2022/7/1 14:29
* @param: [indexName, id]
* @return: java.util.Map
*/
public Map getDocument(String indexName, String id) {
Map strToMap = null;
try {
GetRequest request = new GetRequest(indexName, id);
GetResponse getResponse = restHighLevelClient.get(request, RequestOptions.DEFAULT);
strToMap = JSONObject.parseObject(getResponse.getSourceAsString());
} catch (IOException e) {
e.printStackTrace();
}
return strToMap;
}
/**
* @description: 创建bulkProcessor并初始化
* @author: River
* @date 2022/7/1 14:30
* @param: [client]
* @return: org.elasticsearch.action.bulk.BulkProcessor
*/
private BulkProcessor getBulkProcessor(RestHighLevelClient client) {
BulkProcessor bulkProcessor = null;
try {
BulkProcessor.Listener listener = new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
log.info("Try to insert data number : " + request.numberOfActions());
}
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
log.info("************** Success insert data number : " + request.numberOfActions() + " , id: "
+ executionId);
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
log.error("Bulk is unsuccess : " + failure + ", executionId: " + executionId);
}
};
BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> client
.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
// 设置最大的上传数量
builder.setBulkActions(1000);
builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
// 设置最多的线程并发数
builder.setConcurrentRequests(2);
builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
bulkProcessor = builder.build();
} catch (Exception e) {
e.printStackTrace();
try {
bulkProcessor.awaitClose(100L, TimeUnit.SECONDS);
client.close();
} catch (Exception e1) {
log.error(e1.getMessage());
}
}
return bulkProcessor;
}
/**
* @description: 批量插入
* @author:
* @date 2022/7/1 14:31
* @param: [objectArrayList, indexName, value]
* @return: boolean
*/
public boolean bulkRequest(ArrayList<Map<String, Object>> objectArrayList, String indexName, String value) {
boolean exists = true;
BulkProcessor bulkProcessor = getBulkProcessor(restHighLevelClient);
try {
for (int i = 0; i < objectArrayList.size(); i++) {
bulkProcessor.add(new IndexRequest(indexName)
.id(objectArrayList.get(i).get(value).toString())
.source(JSON.toJSONString(objectArrayList.get(i)), XContentType.JSON));
}
// 将数据刷新到es
bulkProcessor.flush();
} catch (Exception e) {
log.error(e.getMessage());
} finally {
try {
boolean terminatedFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
log.info(String.valueOf(terminatedFlag));
} catch (Exception e) {
log.error(e.getMessage());
}
}
return exists;
}
/**
* @description: 创建文档
* @author:
* @date 2022/7/3 16:03
* @param: [indexName, obj, id]
* @return: boolean
*/
public boolean addDocument(String indexName, Object obj, String id) {
boolean exists = true;
IndexResponse indexResponse = null;
try {
// 创建请求
IndexRequest request = new IndexRequest(indexName);
// 规则 put /kuang_index/_doc/1
request.id(id);
request.timeout(TimeValue.timeValueDays(1));
// 将我们的数据放入请求 json
request.source(JSON.toJSONString(obj), XContentType.JSON);
// 客户端发送请求，获取响应结果
indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);
if (!indexResponse.equals("CREATED")) {//判断响应结果对象是否为CREATED
exists = false;
}
} catch (IOException e) {
exists = false;
}
return exists;
}
/**
* @description: 获取文档，判断是否存在
* @author:
* @date 2022/7/3 16:03
* @param: [indexName, id]
* @return: boolean
*/
public boolean isExists(String indexName, String id) {
boolean exists = true;
try {
GetRequest request = new GetRequest(indexName, id);
// 不获取返回的 _source 的上下文了
request.fetchSourceContext(new FetchSourceContext(false));
request.storedFields("_none_");
exists = restHighLevelClient.exists(request, RequestOptions.DEFAULT);
} catch (IOException e) {
e.printStackTrace();
}
return exists;
}
/**
* @description: 模糊查询
* @author:
* @date 2022/7/3 16:05
* @param: [indexName, key, value]
* @return: java.util.List<java.util.Map<java.lang.String,java.lang.Object>>
*/
public List<Map<String, Object>> searchMatch(String indexName, String key, String value) throws IOException {
List<Map<String, Object>> map = new ArrayList<>();
SearchRequest searchRequest = new SearchRequest(indexName);
// 构建搜索条件
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
MatchQueryBuilder termQueryBuilder = new MatchQueryBuilder(key, value);
termQueryBuilder.fuzziness(Fuzziness.AUTO);
sourceBuilder.query(termQueryBuilder);
sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
searchRequest.source(sourceBuilder);
SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
for (SearchHit documentFields : searchResponse.getHits().getHits()) {
map.add(documentFields.getSourceAsMap());
}
return map;
}
/**
* @description: 精确查询
* @author:
* @date 2022/7/3 16:05
* @param: [indexName, key, value]
* @return: java.util.List<java.util.Map<java.lang.String,java.lang.Object>>
*/
public List<Map<String, Object>> searchQuery(String indexName, String key, String value) throws IOException {
List<Map<String, Object>> map = new ArrayList<>();
SearchRequest searchRequest = new SearchRequest(indexName);
// 构建搜索条件
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, value);
sourceBuilder.query(termQueryBuilder);
sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
searchRequest.source(sourceBuilder);
SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
for (SearchHit documentFields : searchResponse.getHits().getHits()) {
map.add(documentFields.getSourceAsMap());
}
return map;
}
/**
* @description: 根据条件修改
* @author:
* @date 2022/7/3 16:05
* @param: [catalogId, timeliness]
* @return: void
*/
public void EsupdateTimeliness(String catalogId,String timeliness,String indexName,String catalogName){
//构建条件
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(QueryBuilders.termQuery(catalogName, catalogId));
//查询一遍，若没有数据则不执行修改
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().query(boolQueryBuilder);
SearchRequest request = new SearchRequest(indexName);
request.source(sourceBuilder);
//执行查询修改
UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(indexName)
.setQuery(boolQueryBuilder)
.setScript(new Script("ctx._source['timeliness']='" + timeliness + "'"));
try {
SearchResponse search = restHighLevelClient.search(request, RequestOptions.DEFAULT);
long docMun = search.getHits().getTotalHits().value;
if (docMun != 0) {
restHighLevelClient.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT).getUpdated();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
private SearchResponse getCommodityList(String index,TermQueryReq req) {
SearchRequest searchRequest = new SearchRequest(index);
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//根据ID进行排序
sourceBuilder.sort("id", SortOrder.ASC);
//查询条件
BoolQueryBuilder mustQuery = QueryBuilders.boolQuery();
List<QueryCondition> mapList = req.getMapList();
for (QueryCondition queryCondition : mapList) {
if (queryCondition.getExtractive() == 1){//and
if (queryCondition.getValue() != null){
mustQuery.must(QueryBuilders.termQuery(queryCondition.getKey(), queryCondition.getValue()));
}
}else{
Integer[] pulldown = (Integer[]) queryCondition.getValue();
for (Integer integer : pulldown) {
mustQuery.should(QueryBuilders.matchQuery(queryCondition.getKey(), integer));//多条件匹配之一 or
}
}
}
//查询字段
StringBuffer fields = new StringBuffer();
for (QueryCondition queryCondition : mapList) {
fields.append(queryCondition.getKey()).append(",");
}
sourceBuilder.fetchSource(new FetchSourceContext(true, fields.toString().split(","), Strings.EMPTY_ARRAY));
//分页
sourceBuilder.query(mustQuery);
sourceBuilder.from((req.getPageSize() - 1) * req.getPageSize());
sourceBuilder.size(req.getPageSize());
sourceBuilder.trackTotalHits(true);
searchRequest.source(sourceBuilder);
//开始查询
SearchResponse response;
List<Map<String, Object>> list = null;
try {
response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
} catch (Exception e) {
e.printStackTrace();
return null;
}
return response;
}
}
二次聚合
网上说在es5.0以后二次聚合如果使用script脚本去聚合需要添加配置 但是我试了下没什么用 如果有那个大神有好的解决办法可以分享出来 我这边用的是管道聚合也就是聚合套聚合 或者是在洗数据的时候拼接好分组字段
PageInfo<DespatchRes> pageInfo = new PageInfo<>();
List<DespatchRes> relist = Lists.newArrayList();
//起始条数
int from = (req.getPageNum() - 1) * (req.getPageSize() - 1);
//末尾条数
int limit = from+req.getPageNum();
//根据productGroupId进行分桶 分通聚合
TermsAggregationBuilder stateTime = AggregationBuilders.terms("stateTime").field("productGroupId").size(Integer.MAX_VALUE);
TermsAggregationBuilder stateAgg = AggregationBuilders.terms("GroupId").field("pickupTime").size(Integer.MAX_VALUE);
stateAgg.order(BucketOrder.aggregation("_count", true));//根据count数量排序
//取出首条数据作为展示数据
stateAgg.subAggregation(AggregationBuilders.topHits("Group").size(1));
//对聚合后的结果进行排序和分页
stateAgg.subAggregation(new BucketSortPipelineAggregationBuilder("pageInfo", null).from(from).size(limit));
//使用指标聚合 拿到分桶中的非预报数
stateAgg.subAggregation(AggregationBuilders.sum("perSum").field("isPrePackage"));
//将stateTime桶中的数据拼接到stateAgg桶 二次聚合也就是管道聚合
stateAgg.subAggregation(stateTime);
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//时间范围
if (StringUtils.isNotBlank(req.getPickupStartTime()) && StringUtils.isNotBlank(req.getPickupEndTime())){
RangeQueryBuilder date1 = QueryBuilders.rangeQuery("pickupTime").from(req.getPickupStartTime()).to(req.getPickupEndTime());
boolQueryBuilder.filter(date1);
}
boolQueryBuilder.must(QueryBuilders.matchQuery("sts", 0));
boolQueryBuilder.must(QueryBuilders.matchQuery("tenantId", user.getTenantId()));
//添加must条件
if (ObjectUtils.isNotEmpty(req.getCargoOwnerId())){
boolQueryBuilder.must(QueryBuilders.matchQuery("cargoOwnerId", req.getCargoOwnerId()));
}
//添加should条件
if (ObjectUtils.isNotEmpty(req.getWarehouseId())){
List<Long> warehouseId = req.getWarehouseId();
for (Long aLong : warehouseId) {
boolQueryBuilder.should(QueryBuilders.matchQuery("warehouseId", aLong));
}
}
if (ObjectUtils.isNotEmpty(req.getWarehouseCodeId())){
List<Long> warehouseCodeId = req.getWarehouseCodeId();
for (Long aLong : warehouseCodeId) {
boolQueryBuilder.should(QueryBuilders.matchQuery("referenceThree", aLong));
}
}
if (ObjectUtils.isNotEmpty(req.getOrderTypeId())){
List<Long> warehouseCodeId = req.getOrderTypeId();
for (Long aLong : warehouseCodeId) {
boolQueryBuilder.should(QueryBuilders.matchQuery("orderTypeId", aLong));
}
}
// 构建查询条件
NativeSearchQuery build = new NativeSearchQueryBuilder()
.withQuery(boolQueryBuilder)
.withTrackTotalHits(true)
.addAggregation(stateAgg)
.build();
//执行查询
SearchHits<DespatchEs> searchHits = elasticsearchTemplate.search(build, DespatchEs.class);
Aggregations aggregations = searchHits.getAggregations();
Terms terms = (Terms) aggregations.asMap().get("GroupId");
for (Terms.Bucket bucket : terms.getBuckets()) {
DespatchRes rs = new DespatchRes();
//获取count值
rs.setNumber(bucket.getDocCount());
Aggregations bucketAggregations = bucket.getAggregations();
//获取分组数据
TopHits ns = (TopHits) bucketAggregations.asMap().get("Group");
SearchHit[] hits = ns.getHits().getHits();
for (SearchHit hit : hits) {
rs.setId(Long.parseLong(hit.getId()));
Map<String, Object> sourceAsMap = hit.getSourceAsMap();
rs.setPickupTime(sourceAsMap.get("pickupTime") == null ? "" :sourceAsMap.get("pickupTime").toString());
rs.setProductGroupDetail(sourceAsMap.get("productGroupDetail") == null ? "" : sourceAsMap.get("productGroupDetail").toString());
rs.setProductGroupId(sourceAsMap.get("productGroupId") == null ? "":sourceAsMap.get("productGroupId").toString());
rs.setReferenceThree(sourceAsMap.get("referenceThree") == null ? "":sourceAsMap.get("referenceThree").toString());
}
//获取非预包数
Sum perSum = bucketAggregations.get("perSum");
//总数-非预包数 = 预包数
double v = bucket.getDocCount() - perSum.getValue();
rs.setIsPrePackageNumber(v);
relist.add(rs);
}
pageInfo.setList(relist);
pageInfo.setTotal(terms.getBuckets().size());
pageInfo.setPageNum(req.getPageNum());
pageInfo.setPageSize(req.getPageSize());
return ResultRes.success(pageInfo);
我始终相信 在这个世界上一定有着另外一个自己 在做着我不敢做的事情 过着我想过的生活
版权声明：本文为sadnskajf原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。