java使用elasticsearch分組進行聚合查詢（group by）-項目中實際應用

java連接elasticsearch 進行聚合查詢進行相應操作

一：對單個字段進行分組求和

1、表結構圖片：

根據任務id分組，分別統計出每個任務id下有多少個文字標題

1.SQL：select id, count(*) as sum from task group by taskid;   

java ES連接工具類

public class ESClientConnectionUtil {
    public static TransportClient client=null;
    public final static String HOST = "192.168.200.211"; //服務器部署
    public final static Integer PORT = 9301; //端口

    public static TransportClient  getESClient(){
        System.setProperty("es.set.netty.runtime.available.processors", "false");
        if (client == null) {
            synchronized (ESClientConnectionUtil.class) {
                try {
                    //設置集群名稱
                    Settings settings = Settings.builder().put("cluster.name", "es5").put("client.transport.sniff", true).build();
                    //創建client
                    client = new PreBuiltTransportClient(settings).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(HOST), PORT));
                } catch (Exception ex) {
                    ex.printStackTrace();

                    System.out.println(ex.getMessage());
                }
            }
        }
        return client;
    }
    public static TransportClient  getESClientConnection(){
        if (client == null) {
            System.setProperty("es.set.netty.runtime.available.processors", "false");
                try {
                    //設置集群名稱
                    Settings settings = Settings.builder().put("cluster.name", "es5").put("client.transport.sniff", true).build();
                    //創建client
                    client = new PreBuiltTransportClient(settings).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(HOST), PORT));
                } catch (Exception ex) {
                    ex.printStackTrace();
                    System.out.println(ex.getMessage());
            }
        }
        return client;
    }

    //判斷索引是否存在
    public static boolean judgeIndex(String index){
        client= getESClientConnection();
         IndicesAdminClient adminClient;
        //查詢索引是否存在
        adminClient= client.admin().indices();
        IndicesExistsRequest request = new IndicesExistsRequest(index);
        IndicesExistsResponse responses = adminClient.exists(request).actionGet();

        if (responses.isExists()) {
            return true;
        }
        return false;
    }
}

 

java ES語句（根據單列進行分組求和）

//根據 任務id分組進行求和
  SearchRequestBuilder sbuilder = client.prepareSearch("hottopic").setTypes("hot");
//根據taskid進行分組統計，統計出的列別名叫sum
  TermsAggregationBuilder termsBuilder = AggregationBuilders.terms("sum").field("taskid");

 sbuilder.addAggregation(termsBuilder);
  SearchResponse responses= sbuilder.execute().actionGet();
//得到這個分組的數據集合
  Terms terms = responses.getAggregations().get("sum");
  List<BsKnowledgeInfoDTO> lists = new ArrayList<>();
for(int i=0;i<terms.getBuckets().size();i++){
    //statistics
    String id =terms.getBuckets().get(i).getKey().toString();//id
    Long sum =terms.getBuckets().get(i).getDocCount();//數量
System.out.println("=="+terms.getBuckets().get(i).getDocCount()+"------"+terms.getBuckets().get(i).getKey());
}
//分別打印出統計的數量和id值

 根據多列進行分組求和

//根據 任務id分組進行求和
  SearchRequestBuilder sbuilder = client.prepareSearch("hottopic").setTypes("hot");
//根據taskid進行分組統計，統計出的列別名叫sum
  TermsAggregationBuilder termsBuilder = AggregationBuilders.terms("sum").field("taskid");
//根據第二個字段進行分組
 TermsAggregationBuilder aAggregationBuilder2 = AggregationBuilders.terms("region_count").field("birthplace");
//如果存在第三個，以此類推；
  sbuilder.addAggregation(termsBuilder.subAggregation(aAggregationBuilder2));
  SearchResponse responses= sbuilder.execute().actionGet();
//得到這個分組的數據集合
  Terms terms = responses.getAggregations().get("sum");
  List<BsKnowledgeInfoDTO> lists = new ArrayList<>();
for(int i=0;i<terms.getBuckets().size();i++){
    //statistics
    String id =terms.getBuckets().get(i).getKey().toString();//id
    Long sum =terms.getBuckets().get(i).getDocCount();//數量
System.out.println("=="+terms.getBuckets().get(i).getDocCount()+"------"+terms.getBuckets().get(i).getKey());
}
//分別打印出統計的數量和id值

對多個field求max/min/sum/avg

SearchRequestBuilder requestBuilder = client.prepareSearch("hottopic").setTypes("hot");
//根據taskid進行分組統計，統計別名為sum
        TermsAggregationBuilder aggregationBuilder1 = AggregationBuilders.terms("sum").field("taskid") 
//根據tasktatileid進行升序排列
                .order(Order.aggregation("tasktatileid", true));
// 求tasktitleid 進行求平均數 別名為avg_title

        AggregationBuilder aggregationBuilder2 = AggregationBuilders.avg("avg_title").field("tasktitleid");
//
        AggregationBuilder aggregationBuilder3 = AggregationBuilders.sum("sum_taskid").field("taskid");
        requestBuilder.addAggregation(aggregationBuilder1.subAggregation(aggregationBuilder2).subAggregation(aggregationBuilder3));
        SearchResponse response = requestBuilder.execute().actionGet();

        Terms aggregation = response.getAggregations().get("sum");
        Avg terms2 = null;
        Sum term3 = null;
        for (Terms.Bucket bucket : aggregation.getBuckets()) {
            terms2 = bucket.getAggregations().get("avg_title"); // org.elasticsearch.search.aggregations.metrics.avg.InternalAvg
            term3 = bucket.getAggregations().get("sum_taskid"); // org.elasticsearch.search.aggregations.metrics.sum.InternalSum
            System.out.println("編號=" + bucket.getKey() + ";平均=" + terms2.getValue() + ";總=" + term3.getValue());
        }

如上內容若有不恰當支持，請各位多多包涵並進行點評。技術在於溝通！