Elasticsearch聚合查询案例分享-白红宇

Elasticsearch聚合查询案例分享

1.案例介绍

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

2.准备工作

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

3.定义统计dsl

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


        
         
              0)                        {                            "terms": {                                "applicationName.keyword": [                                #foreach($application in $channelApplications)                                   #if($velocityCount > 0),#end $application.applicationName                                #end                                ]                            }                        },                        #end                        {"range": {                                "startTime": {                                    "gte": #[startTime],##统计开始时间                                    "lt": #[endTime]  ##统计截止时间                                }                            }                        }                    ]                }            },            "size":0,            "aggs": {                "applicationsums": {                      "terms": {                        "field": "applicationName.keyword",##按应用名称进行统计计数                        "size":10000                      },                      "aggs":{                            "successsums" : {                                "terms" : {                                    "field" : "err" ##按err标识统计每个应用的成功数和失败数，0标识成功，1标识失败                                }                            },                            "elapsed_ranges" : {                                "range" : {                                    "field" : "elapsed", ##按响应时间分段统计                                    "keyed" : true,                                    "ranges" : [                                        { "key" : "1秒", "to" : 1000 },                                        { "key" : "3秒", "from" : 1000, "to" : 3000 },                                        { "key" : "5秒", "from" : 3000, "to" : 5000 },                                        { "key" : "5秒以上", "from" : 5000 }                                    ]                                }                            }                      }                }            }        }        ]]>

4.编写统计dao及统计方法

public class TraceESDao {        public List
    
      getApplicationSumStatic(TraceExtraCriteria traceExtraCriteria){    	init();    	//返回json统计报文，调试用，一遍根据json报文组装统计结果列表//		String response = clientUtil.executeRequest("trace-*/_search",//                                  "applicationSumStatic",traceExtraCriteria);		//根据条件进行统计，在对象traceExtraCriteria中指定开始时间和结束时间		MapRestResponse restResponse = clientUtil.search("trace-*/_search",				                      "applicationSumStatic",traceExtraCriteria);		//组装统计结果		//获取应用统计列表，包含每个应用的名称、总访问量以及成功数和失败数		List
     
      
       > appstatics = (List
       
        
         >)restResponse.getAggBuckets("applicationsums");		if(appstatics != null && appstatics.size() > 0) {			List
         
           applicationStatics = new ArrayList
          
           (appstatics.size()); ApplicationStatic applicationStatic = null; for (int i = 0; i < appstatics.size(); i++) { applicationStatic = new ApplicationStatic(); Map
           
             map = appstatics.get(i); //应用名称 String appName = (String) map.get("key"); applicationStatic.setApplicationName(appName); //应用总访问量 Long totalsize = ResultUtil.longValue( map.get("doc_count"),0l); applicationStatic.setTotalSize(totalsize); //获取成功数和失败数 List
            
             
              > appstatic = (List
              
               
                >)ResultUtil.getAggBuckets(map, "successsums"); /** "buckets": [ { "key": 0, "doc_count": 30 } ] */ //key 0 Long success = 0l;//成功数 Long failed = 0l;//失败数 for (int j = 0; j < appstatic.size(); j++) { Map
                
                  stats = appstatic.get(j); Integer key = (Integer) stats.get("key");//成功和错误标识 if (key == 0)//成功 success = ResultUtil.longValue( stats.get("doc_count"),0l); else if (key == 1)//失败 failed = ResultUtil.longValue( stats.get("doc_count"),0l); } applicationStatic.setSuccessCount(success); applicationStatic.setFailCount(failed); List
                 
                   applicationPeriodStatics = new ArrayList
                  
                   (4); ApplicationPeriodStatic applicationPeriodStatic = null; //获取响应时间分段统计信息 Map
                   
                    > appPeriodstatic = (Map
                    
                     >)ResultUtil.getAggBuckets(map, "elapsed_ranges"); //1秒 Map
                     
                       period = appPeriodstatic.get("1秒"); applicationPeriodStatic = new ApplicationPeriodStatic(); applicationPeriodStatic.setPeriod("1秒"); applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l)); applicationPeriodStatic.setTo(ResultUtil.intValue(period.get("to"),1000)); applicationPeriodStatics.add(applicationPeriodStatic); //3秒 period = appPeriodstatic.get("3秒"); applicationPeriodStatic = new ApplicationPeriodStatic(); applicationPeriodStatic.setPeriod("3秒"); applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l)); applicationPeriodStatic.setFrom(ResultUtil.intValue(period.get("from"),1000)); applicationPeriodStatic.setTo(ResultUtil.intValue(period.get("to"),3000)); applicationPeriodStatics.add(applicationPeriodStatic); //5秒 period = appPeriodstatic.get("5秒"); applicationPeriodStatic = new ApplicationPeriodStatic(); applicationPeriodStatic.setPeriod("5秒"); applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l)); applicationPeriodStatic.setFrom(ResultUtil.intValue(period.get("from"),3000)); applicationPeriodStatic.setTo(ResultUtil.intValue(period.get("to"),5000)); applicationPeriodStatics.add(applicationPeriodStatic); //5秒以上 period = appPeriodstatic.get("5秒以上"); applicationPeriodStatic = new ApplicationPeriodStatic(); applicationPeriodStatic.setPeriod("5秒以上"); applicationPeriodStatic.setDocCount(ResultUtil.longValue(period.get("doc_count"),0l)); applicationPeriodStatic.setFrom(ResultUtil.intValue(period.get("from"),5000)); applicationPeriodStatics.add(applicationPeriodStatic); applicationStatic.setApplicationPeriodStatics(applicationPeriodStatics); applicationStatics.add(applicationStatic); } //返回统计结果 return applicationStatics; } return null; }}

5.执行测试用例

@Test	public void testAppliationstaticList(){		TraceExtraCriteria traceExtraCriteria = new TraceExtraCriteria();		traceExtraCriteria.setStartTime(1516304868072l);//以long方式设置统计开始时间，Date的getTime方法获取		traceExtraCriteria.setEndTime(1516349516377l);//以long方式设置统计截止时间，Date的getTime方法获取		TraceESDao traceESDao = new TraceESDao();//定义dao组件		List
    
      applicationStatics = traceESDao.getApplicationSumStatic(traceExtraCriteria);		System.out.println(applicationStatics.size());	}

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

@Test	public void testAppStatic(){		TraceExtraCriteria traceExtraCriteria = new TraceExtraCriteria();		traceExtraCriteria.setStartTime(1516304868072l);		traceExtraCriteria.setEndTime(1516349516377l);		ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/estrace/ESTracesMapper.xml");		//通过下面的方法先得到查询的json报文，然后再通过MapRestResponse查询遍历结果，调试的时候打开String response的注释		//String response = clientUtil.executeRequest("trace-*/_search","applicationSumStatic",traceExtraCriteria);		//System.out.println(response);		MapRestResponse restResponse = clientUtil.search("trace-*/_search","applicationSumStatic",traceExtraCriteria);		List
     
      > appstatics = restResponse.getAggBuckets("applicationsums",new ESTypeReference
      
       
        
         >>(){});		int doc_count_error_upper_bound = restResponse.getAggAttribute("applicationsums","doc_count_error_upper_bound",int.class);		int sum_other_doc_count = restResponse.getAggAttribute("applicationsums","sum_other_doc_count",int.class);		System.out.println("doc_count_error_upper_bound:"+doc_count_error_upper_bound);		System.out.println("sum_other_doc_count:"+sum_other_doc_count);		for(int i = 0; i < appstatics.size(); i ++){			Map
         
           map = appstatics.get(i); //应用名称 String appName = (String)map.get("key"); //应用总访问量 int totalsize = (int)map.get("doc_count"); //获取成功数和失败数 List
          
           
            > appstatic = ResultUtil.getAggBuckets(map ,"successsums",new ESTypeReference
            
             
              
               >>(){}); doc_count_error_upper_bound = ResultUtil.getAggAttribute(map ,"successsums","doc_count_error_upper_bound",int.class); sum_other_doc_count = ResultUtil.getAggAttribute(map ,"successsums","sum_other_doc_count",int.class); System.out.println("doc_count_error_upper_bound:"+doc_count_error_upper_bound); System.out.println("sum_other_doc_count:"+sum_other_doc_count); /** "buckets": [ { "key": 0, "doc_count": 30 } ] */ //key 0 int success = 0;//成功数 int failed = 0;//失败数 for(int j = 0; j < appstatic.size(); i ++){ Map
               
                 stats = appstatic.get(i); int key = (int) stats.get("key");//成功和错误标识 if(key == 0) success = (int)stats.get("doc_count"); else if(key == 1) failed = (int)stats.get("doc_count"); } } }

7.相关资料

高性能elasticsearch ORM开发库使用介绍

bboss elasticsearch交流群：166471282