Elasticsearch聚合查询案例分享
1.案例介绍
统计特定时间范围内每个应用的总访问量、访问成功数、访问失败数,每个应用请求响应时间分段统计（1秒内，1-3秒，3-5秒，5秒以上
2.准备工作
参考文档《
高性能elasticsearch ORM开发库使用介绍》中的第1章节和第2章节，在自己的工程中导入bboss es依赖包和配置es参数
3.定义统计dsl
在源码目录下新建文件esmapper/estrace/ESTracesMapper.xml，内容如下
<properties>
<!--
应用汇总统计：总访问量，成功数，失败数
bboss es dao通过名称applicationSumStatic引用脚本
-->
<property name="applicationSumStatic">
<![CDATA[
{
"query": {
"bool": {
"filter": [
#if($channelApplications && $channelApplications.size() > 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 }
]
}
}
}
}
}
}
]]>
</property>
</properties>
4.编写统计dao及统计方法
public class TraceESDao {
public List<ApplicationStatic> getApplicationSumStatic(TraceExtraCriteria traceExtraCriteria){
init();
//返回json统计报文，调试用，一遍根据json报文组装统计结果列表
// String response = clientUtil.executeRequest("trace-*/_search",
// "applicationSumStatic",traceExtraCriteria);
//根据条件进行统计，在对象traceExtraCriteria中指定开始时间和结束时间
MapRestResponse restResponse = clientUtil.search("trace-*/_search",
"applicationSumStatic",traceExtraCriteria);
//组装统计结果
//获取应用统计列表，包含每个应用的名称、总访问量以及成功数和失败数
List<Map<String,Object>> appstatics = (List<Map<String,Object>>)restResponse.getAggBuckets("applicationsums");
if(appstatics != null && appstatics.size() > 0) {
List<ApplicationStatic> applicationStatics = new ArrayList<ApplicationStatic>(appstatics.size());
ApplicationStatic applicationStatic = null;
for (int i = 0; i < appstatics.size(); i++) {
applicationStatic = new ApplicationStatic();
Map<String, Object> 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<Map<String, Object>> appstatic = (List<Map<String, Object>>)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<String, Object> 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<ApplicationPeriodStatic> applicationPeriodStatics = new ArrayList<ApplicationPeriodStatic>(4);
ApplicationPeriodStatic applicationPeriodStatic = null;
//获取响应时间分段统计信息
Map<String, Map<String, Object>> appPeriodstatic = (Map<String, Map<String, Object>>)ResultUtil.getAggBuckets(map, "elapsed_ranges");
//1秒
Map<String, Object> 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<ApplicationStatic> 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<Map<String,Object>> appstatics = restResponse.getAggBuckets("applicationsums",new ESTypeReference<List<Map<String,Object>>>(){});
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<String,Object> map = appstatics.get(i);
//应用名称
String appName = (String)map.get("key");
//应用总访问量
int totalsize = (int)map.get("doc_count");
//获取成功数和失败数
List<Map<String,Object>> appstatic = ResultUtil.getAggBuckets(map ,"successsums",new ESTypeReference<List<Map<String,Object>>>(){});
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<String,Object> 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开发库使用介绍
https://my.oschina.net/bboss/blog/1556866
bboss elasticsearch交流群：166471282
