Elasticsearch聚合查询案例分享

Elasticsearch聚合查询案例分享

1.案例介绍

本文包含三个案例:

案例1：统计特定时间范围内每个应用的总访问量、访问成功数、访问失败数,每个应用请求响应时间分段统计（1秒内，1-3秒，3-5秒，5秒以上 ）

案例3：简单的cardinality统计

2.准备工作

参考文档《》中的第1章节和第2章节，在自己的工程中导入bboss es依赖包和配置es参数

3.案例

在源码目录下新建文件esmapper/estrace/ESTracesMapper.xml，内容如下

3.1.2 编写统计dao及统计方法

  2.     public List<ApplicationStatic> getApplicationSumStatic(TraceExtraCriteria traceExtraCriteria){
  3.         ClientInterface clientUtil= ElasticSearchHelper.getConfigRestClientUtil("esmapper/estrace/ESTracesMapper.xml");
  4.         //返回json统计报文，调试用，一遍根据json报文组装统计结果列表
  5. //        String response = clientUtil.executeRequest("trace-*/_search",
  6. //                                  "applicationSumStatic",traceExtraCriteria);
  7.         //根据条件进行统计，在对象traceExtraCriteria中指定开始时间和结束时间
  8.         MapRestResponse restResponse = clientUtil.search("trace-*/_search",
  9.                                       "applicationSumStatic",traceExtraCriteria);
  11.         //组装统计结果
  12.         //获取应用统计列表，包含每个应用的名称、总访问量以及成功数和失败数
  13.         List<Map<String,Object>> appstatics = (List<Map<String,Object>>)restResponse.getAggBuckets("applicationsums");
  14.         if(appstatics != null && appstatics.size() > 0) {
  15.             List<ApplicationStatic> applicationStatics = new ArrayList<ApplicationStatic>(appstatics.size());
  16.             ApplicationStatic applicationStatic = null;
  17.             for (int i = 0; i < appstatics.size(); i++) {
  18.                 applicationStatic = new ApplicationStatic();
  19.                 Map<String, Object> map = appstatics.get(i);
  20.                 //应用名称
  21.                 String appName = (String) map.get("key");
  22.                 applicationStatic.setApplicationName(appName);
  23.                 //应用总访问量
  24.                 Long totalsize = ResultUtil.longValue( map.get("doc_count"),0l);
  25.                 applicationStatic.setTotalSize(totalsize);
  26.                 //获取成功数和失败数
  27.                 List<Map<String, Object>> appstatic = (List<Map<String, Object>>)ResultUtil.getAggBuckets(map, "successsums");
  29.                 /**
  30.                  "buckets": [
  31.                  {
  32.                  "key": 0,
  33.                  "doc_count": 30
  34.                  }
  35.                  ]
  36.                  */
  37.                 //key 0
  38.                 Long success = 0l;//成功数
  39.                 Long failed = 0l;//失败数
  40.                 for (int j = 0; j < appstatic.size(); j++) {
  41.                     Map<String, Object> stats = appstatic.get(j);
  42.                     Integer key = (Integer) stats.get("key");//成功和错误标识
  43.                     if (key == 0)//成功
  44.                         success = ResultUtil.longValue( stats.get("doc_count"),0l);
  45.                     else if (key == 1)//失败
  46.                         failed = ResultUtil.longValue( stats.get("doc_count"),0l);
  47.                 }
  48.                 applicationStatic.setSuccessCount(success);
  49.                 applicationStatic.setFailCount(failed);
  50.                 List<ApplicationPeriodStatic> applicationPeriodStatics = new ArrayList<ApplicationPeriodStatic>(4);
  51.                 //获取响应时间分段统计信息
  52.                 Map<String, Map<String, Object>> appPeriodstatic = (Map<String, Map<String, Object>>)ResultUtil.getAggBuckets(map, "elapsed_ranges");
  53.                 //1秒
  54.                 Map<String, Object> period = appPeriodstatic.get("1秒");
  55.                 applicationPeriodStatic = new ApplicationPeriodStatic();
  56.                 applicationPeriodStatic.setPeriod("1秒");
  57.                 applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l));
  58.                 applicationPeriodStatic.setTo(ResultUtil.intValue(period.get("to"),1000));
  59.                 applicationPeriodStatics.add(applicationPeriodStatic);
  62.                 period = appPeriodstatic.get("3秒");
  63.                 applicationPeriodStatic = new ApplicationPeriodStatic();
  64.                 applicationPeriodStatic.setPeriod("3秒");
  65.                 applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l));
  66.                 applicationPeriodStatic.setFrom(ResultUtil.intValue(period.get("from"),1000));
  67.                 applicationPeriodStatic.setTo(ResultUtil.intValue(period.get("to"),3000));
  68.                 applicationPeriodStatics.add(applicationPeriodStatic);
  70.                 //5秒
  71.                 period = appPeriodstatic.get("5秒");
  72.                 applicationPeriodStatic = new ApplicationPeriodStatic();
  73.                 applicationPeriodStatic.setPeriod("5秒");
  74.                 applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l));
  75.                 applicationPeriodStatic.setFrom(ResultUtil.intValue(period.get("from"),3000));
  76.                 applicationPeriodStatic.setTo(ResultUtil.intValue(period.get("to"),5000));
  77.                 applicationPeriodStatics.add(applicationPeriodStatic);
  79.                 //5秒以上
  80.                 period = appPeriodstatic.get("5秒以上");
  81.                 applicationPeriodStatic = new ApplicationPeriodStatic();
  82.                 applicationPeriodStatic.setPeriod("5秒以上");
  83.                 applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l));
  84.                 applicationPeriodStatic.setFrom(ResultUtil.intValue(period.get("from"),5000));
  85.                 applicationPeriodStatics.add(applicationPeriodStatic);
  87.                 applicationStatic.setApplicationPeriodStatics(applicationPeriodStatics);
  88.                 applicationStatics.add(applicationStatic);
  90.             }
  91.             //返回统计结果
  92.             return applicationStatics;
  93.         }
  94.         return null;
  95.     }
  96. }

Java代码

3.1.4 获取元数据信息的测试方法

java代码 

  1. @Test
  2.     public void testAppStatic(){
  3.         TraceExtraCriteria traceExtraCriteria = new TraceExtraCriteria();
  4.         traceExtraCriteria.setStartTime(1516304868072l);
  5.         traceExtraCriteria.setEndTime(1516349516377l);
  6.         ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/estrace/ESTracesMapper.xml");
  7.         //通过下面的方法先得到查询的json报文，然后再通过MapRestResponse查询遍历结果，调试的时候打开String response的注释
  8.         //String response = clientUtil.executeRequest("trace-*/_search","applicationSumStatic",traceExtraCriteria);
  9.         //System.out.println(response);
  10.         MapRestResponse restResponse = clientUtil.search("trace-*/_search","applicationSumStatic",traceExtraCriteria);
  12.         List<Map<String,Object>> appstatics = restResponse.getAggBuckets("applicationsums",new ESTypeReference<List<Map<String,Object>>>(){});
  13.         int doc_count_error_upper_bound = restResponse.getAggAttribute("applicationsums","doc_count_error_upper_bound",int.class);
  14.         int sum_other_doc_count = restResponse.getAggAttribute("applicationsums","sum_other_doc_count",int.class);
  15.         System.out.println("doc_count_error_upper_bound:"+doc_count_error_upper_bound);
  16.         System.out.println("sum_other_doc_count:"+sum_other_doc_count);
  17.         for(int i = 0; i < appstatics.size(); i ++){
  18.             Map<String,Object> map = appstatics.get(i);
  19.             //应用名称
  20.             //应用总访问量
  21.             int totalsize =  (int)map.get("doc_count");
  22.             //获取成功数和失败数
  23.             List<Map<String,Object>> appstatic = ResultUtil.getAggBuckets(map ,"successsums",new ESTypeReference<List<Map<String,Object>>>(){});
  24.               doc_count_error_upper_bound = ResultUtil.getAggAttribute(map ,"successsums","doc_count_error_upper_bound",int.class);
  26.             System.out.println("doc_count_error_upper_bound:"+doc_count_error_upper_bound);
  27.             System.out.println("sum_other_doc_count:"+sum_other_doc_count);
  28.             /**
  29.             "buckets": [
  30.             {
  31.                 "key": 0,
  32.                     "doc_count": 30
  33.             }
  34.                         ]
  35.              */
  36.             //key 0
  37.             int success = 0;//成功数
  38.             int failed = 0;//失败数
  39.             for(int j = 0; j < appstatic.size(); i ++){
  40.                 Map<String,Object> stats = appstatic.get(i);
  41.                 int key = (int) stats.get("key");//成功和错误标识
  42.                 if(key == 0)
  43.                     success = (int)stats.get("doc_count");
  44.                 else if(key == 1)
  45.                     failed = (int)stats.get("doc_count");
  46.             }
  48.         }
  51.     }

建立dsl配置文件esmapper/testagg.xml，定义termAgg：

3.2.2 执行dsl

  1.     @Test
  2.     public void termAgg(){
  3.         ClientInterface clientInterface = ElasticSearchHelper.getConfigRestClientUtil("esmapper/testagg.xml");
  4.         //ESDatas<Map> traces = clientInterface.searchAll("trace-*",1000,Map.class);//获取总记录集合
  5.         Map params = new HashMap();//聚合统计条件参数
  6.         params.put("application","testweb");
  7.         SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:dd");
  8.         try {
  9.             params.put("startTime",format.parse("1999-01-01 00:00:00").getTime());
  10.             params.put("endTime",new Date().getTime());
  11.             params.put("rpc","/testweb/jsp/logoutredirect.jsp");
  12.         } catch (ParseException e) {
  13.             e.printStackTrace();
  14.         }
  15.         //一行代码，执行每个服务的访问量总数统计
  16.         ESAggDatas<LongAggHit> response = clientInterface.searchAgg("trace-*/_search",//从trace-开头的索引表中检索数据
  17.                                                                     "termAgg", //配置在esmapper/testagg.xml中的dsl语句
  18.                                                                     params,    //dsl语句termAgg中需要的查询参数
  19.                                                                     LongAggHit.class,  //封装聚合统计中每个服务地址及服务访问量的地址
  20.                                                                     "traces");  //term统计桶的名称，参见dsl语句
  21.         List<LongAggHit> aggHitList = response.getAggDatas();//每个服务的访问量
  22.         long totalSize = response.getTotalSize();//总访问量
  24.     }

3.3.2 执行dsl

  1.     @Test
  2.     public void candicateAgg(){
  3.         ClientInterface clientInterface = ElasticSearchHelper.getConfigRestClientUtil("esmapper/testagg.xml");
  4.         Map params = null;//单值聚合统计条件参数
  5.         //一行代码，执行服务基数统计
  6.         ESAggDatas<SingleLongAggHit> response = clientInterface.searchAgg("trace-*/_search","candicateAgg",params,SingleLongAggHit.class,"traces");
  7.         SingleLongAggHit aggHitList = response.getSingleAggData();
  8.         long value = aggHitList.getValue();
  9.         long totalSize = response.getTotalSize();//总访问量

4.相关资料

bboss elasticsearch交流：166471282