1.等价的sql
SELECT DISTINCT field1,field2 FROM test_index.test_type
等价于
SELECT field1,field2 FROM test_index.test_type GROUP BY field1,field2
2.而group by的查询,在es中我们可以用Aggregation(聚合)去实现,等价的DSL查询语句如下:
POST /test_index/test_type/_search
{
"from": 0,
"size": 0,
"aggregations": {
"field1": {
"terms": {
"field": "field1",
"size": 2147483647
},
"aggregations": {
"field2": {
"terms": {
"field": "field2",
"size": 2147483647
}
}
}
}
}
}
3.java的实现:
import com.google.common.collect.ArrayListMultimap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Stack;
import java.util.stream.Collectors;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
/**
*
* @author zhongchenghui
*/
public class EsSearch {
private static final String CLUSTER_NAME = "test_cluster";
private static final String ES_ADDRESSES = "192.168.12.1,192.168.12.2,192.168.12.3";
private static final String INDEX_NAME = "test_index";
private static final Client ES_CLIENT = ESClientFactory.newInstance(CLUSTER_NAME, ES_ADDRESSES);
/**
* 根据多个字段group by的查询
*
* @param type
* @param groupColumnsNames
* @return
* @throws Exception
*/
public List<Map<String, String>> getRowsByGroup(String type, String... groupColumnsNames) throws Exception {
TermsBuilder allTermsBuilder = createTermsAggregationBuilder(groupColumnsNames);
SearchResponse response = createSearchRequestBuilder(type, groupColumnsNames).setSize(0).addAggregation(allTermsBuilder).execute().actionGet();
List<Map<String, String>> rows = new ArrayList<>();
Terms agg = response.getAggregations().get(groupColumnsNames[0]);
int i = 0;
agg.getBuckets().stream().forEach((bucket) -> {
rows.addAll(getFlatBucket(i, bucket, groupColumnsNames));
});
return rows;
}
/**
* 逐层创建AggregationBuilder
* (此处Integer.MAX_VALUE是逐层分组的最大组数)
* @param groupColumnsNames
* @return
* @throws Exception
*/
private TermsBuilder createTermsAggregationBuilder(String... groupColumnsNames) {
TermsBuilder allTermsBuilder = AggregationBuilders.terms(groupColumnsNames[0]).field(groupColumnsNames[0]).size(Integer.MAX_VALUE);
TermsBuilder tmpTermsBuilder = allTermsBuilder;
for (int i = 1; i < groupColumnsNames.length; i++) {
TermsBuilder termsBuilder = AggregationBuilders.terms(groupColumnsNames[i]).field(groupColumnsNames[i]).size(Integer.MAX_VALUE);
tmpTermsBuilder.subAggregation(termsBuilder);
tmpTermsBuilder = termsBuilder;
}
return allTermsBuilder;
}
/**
* 创建查询请求的Builder
*
* @param type
* @param groupColumnsNames
* @return
* @throws Exception
*/
private SearchRequestBuilder createSearchRequestBuilder(String type, String... columnNames) {
SearchRequestBuilder searchRequestBuilder = ES_CLIENT.prepareSearch(INDEX_NAME).setTypes(type).setSize(50000);
if (Objects.nonNull(columnNames) && columnNames.length > 0) {
searchRequestBuilder.addFields(columnNames);
}
return searchRequestBuilder;
}
/**
* 用堆栈将respone中的聚合结果拉平返回(广度优先遍历)
*
* @param layer
* @param bucket
* @param groupColumnsNames
* @return
*/
private List<Map<String, String>> getFlatBucket(int layer, Terms.Bucket bucket, String... groupColumnsNames) {
ArrayListMultimap<BucketNode, BucketNode> bucketRowMultimap = ArrayListMultimap.create();
Stack<BucketNode> nodeStack = new Stack<>();
BucketNode bucketNode = new BucketNode(layer, groupColumnsNames[layer], bucket);
nodeStack.add(bucketNode);
bucketRowMultimap.put(bucketNode, bucketNode);
while (!nodeStack.isEmpty()) {
bucketNode = nodeStack.pop();
List<BucketNode> childerNodes = getChildrenBucketNodes(bucketNode, groupColumnsNames);
if (childerNodes != null && !childerNodes.isEmpty()) {
List<BucketNode> parentRoute = bucketRowMultimap.removeAll(bucketNode);
for (BucketNode child : childerNodes) {
nodeStack.push(child);
bucketRowMultimap.putAll(child, parentRoute);
bucketRowMultimap.put(child, child);
}
}
}
return convertToRows(bucketRowMultimap.asMap().values());
}
/**
* 获得下一层Bucket的节点列表
*
* @param parentNode
* @param groupColumnsNames
* @return
*/
private List<BucketNode> getChildrenBucketNodes(BucketNode parentNode, String... groupColumnsNames) {
int currentLayer = parentNode.layer + 1;
if (currentLayer < groupColumnsNames.length) {
String currentAggName = groupColumnsNames[currentLayer];
Terms currentAgg = parentNode.bucket.getAggregations().get(currentAggName);
if (Objects.nonNull(currentAgg)) {
return currentAgg.getBuckets().stream().map(bucket -> new BucketNode(currentLayer, currentAggName, bucket)).collect(Collectors.toList());
}
}
return null;
}
private List<Map<String, String>> convertToRows(Collection<Collection<BucketNode>> bucketRoutes) {
return bucketRoutes.stream().map(bucketRoute -> convertToRow(bucketRoute)).collect(Collectors.toList());
}
private Map<String, String> convertToRow(Collection<BucketNode> bucketRoute) {
Map<String, String> row = new HashMap<>();
bucketRoute.stream().forEach(bucketNode -> row.put(bucketNode.aggName, bucketNode.bucket.getKeyAsString()));
return row;
}
class BucketNode {
int layer;
String aggName;
Terms.Bucket bucket;
public BucketNode(int layer, String aggName, Terms.Bucket bucket) {
BucketNode.this.layer = layer;
BucketNode.this.aggName = aggName;
BucketNode.this.bucket = bucket;
}
@Override
public String toString() {
return "BucketNode{" + "layer=" + layer + ", aggName=" + aggName + ", bucket_key=" + bucket.getKeyAsString() + '}';
}
}
}
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author zhongchenghui
*/
public class ESClientFactory {
private static final Logger LOGGER = LoggerFactory.getLogger(ESClientFactory.class);
private static final ConcurrentHashMap<String, Client> CLIENT_CACHE = new ConcurrentHashMap<>();
public static Client newInstance(String clusterName, String hostStr) {
Client client = CLIENT_CACHE.get(clusterName);
if (client == null) {
Map<String, Integer> addressMap = ESClientFactory.getESAddressMap(hostStr);
Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName).build();
TransportClient newClient = TransportClient.builder().settings(settings).build();
addressMap.keySet().forEach((host) -> {
try {
newClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), addressMap.get(host)));
} catch (UnknownHostException ex) {
LOGGER.error("Init ES client failure,cluster name is:{},Error:{}", clusterName, ex);
}
});
client = newClient;
CLIENT_CACHE.put(clusterName, newClient);
}
return client;
}
private static Map<String, Integer> getESAddressMap(String hostStr) {
Map<String, Integer> hostMap = new HashMap<>();
String[] hosts = hostStr.split(",");
for (String host : hosts) {
String[] hostPort = host.trim().split(":");
Integer port = hostPort.length < 2 ? 9300 : Integer.valueOf(hostPort[1]);
hostMap.put(hostPort[0], port);
}
return hostMap;
}
}
4.存在的问题:
a.因为实现的方式是一层层往下聚合,当es中的document出现field1的字段为null的时候,该条件就不会往下聚合,即使该document的field2字段有值。(可用指定字符代替null来解决这个问题)
b.不适合于数据量太大表,3中的代码要求最大每个字段的分组数不能大于Integer.MAX_VALUE
c.返回的字段只能是参与group by的字段
分享到:
相关推荐
方法如果传总页数了,es就不用查询总页数,直接通过开始位置到结束位置取数即可
今天小编就为大家分享一篇关于JAVA使用Elasticsearch查询in和not in的实现方式,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧
ES 聚合查询结果转换成相应的对象集合,ES 聚合查询结果转换成相应的对象集合
es的简单查询和聚合查询用法说明,用于理解es的聚合查询和简单查询的思路过程,便于快速的应用es, 希望能有所帮助,谢谢
springboot整合elasticsearch7,进行数据同步。elasticsearch相关度查询、排序。高亮显示;自动补全等功能。代码仅供参考,代码中有具体的注释,可以根据代码及注释内容,对自己项目架构及业务进行修改、整合。
elasticsearch聚合
对Elasticsearch-PHP进行查询语句封装 可实现链式调用 方便 es查询
Elasticsearch聚合 之 Terms
4.(后端技术篇java)ElasticSearch实现反向地址匹配服务(点周边查询服务) 5.(后端技术篇java)ElasticSearch实现矩形空间查询服务 6.(后端技术篇java)ElasticSearch实现圆形空间查询服务 7.(后端技术篇java)...
聚合查询分页测试termsAgg.size(2147483647); //指定最大统计显示多少行步骤1:全量聚合,size设置为: 2147483647。 ES5.X/6.X版本设置为2147483647 ,它等于2^31-1,请看该地方代码
elasticsearch脚本实现先聚合,然后使用聚合的值进行后过滤
searchEngine 是基于 ElasticSearch 和 Java 实现的搜索引擎系统,实现关键字高亮搜索、添加文本等功能。 该项目集成了 Spring Boot、ElasticSearch、RestHighLevelClient、Vue.js、Element-ui、Log4j 和 Fastjson ...
十分钟学会使用 Elasticsearch 优雅搭建自己的搜索系统。 什么是elasticsearch Elasticsearch 是一个开源的高度可扩展的全文搜索和分析引擎,拥有查询近实时的超强性能。 大名鼎鼎的Lucene 搜索引擎被广泛用于搜索...
创建索引,删除索引,更新索引,模糊搜索,模糊全文搜索,精确查找
elasticsearch先聚合然后通过聚合后的值再进行过滤 a
ElasticSearch查询term,terms,match,...ElasticSearch查询cardinality,range,extended_stats聚合统计aggregations查询 ElasticSearch查询geo_distance,geo_bounding_box,geo_polygon地图检索geo查询 的Java实现
ElasticSearch查询学习;ElasticSearch查询学习;ElasticSearch查询学习;ElasticSearch查询学习;ElasticSearch查询学习;ElasticSearch查询学习;
主要介绍了java使用elasticsearch分组进行聚合查询过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
本实例属于简单易上手并且能够拿来就用的SpringBoot ES 项目,全文使用的是ElasticsearchTemplate进行开发。 本实例涵盖ES中的各类操作,如索引操作、CRUD操作、批处理、结果排序、分页查询、检索查询、关键字查询、...
es也是用Java开发并使用Lucene作为其核心来实现所有索引和搜索功能,但是它的目的是通过简单的RESTful API来隐藏Lucene的复杂性,从而让全文搜索变得简单。es基本是开箱即用(解压就可以用!),非常简单。Solr安装...