Elasticsearch聚合查询案例分享

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介: Elasticsearch聚合查询案例分享 1.案例介绍 统计特定时间范围内每个应用的总访问量、访问成功数、访问失败数,每个应用请求响应时间分段统计(1秒内,1-3秒,3-5秒,5秒以上 2.
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
相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
目录
相关文章
|
1月前
|
存储 搜索推荐 Java
|
2月前
|
存储 固态存储 Java
Elasticsearch中查询性能优化
Elasticsearch中查询性能优化
189 0
|
3月前
Elasticsearch之RestClient查询文档
Elasticsearch之RestClient查询文档
139 1
|
3月前
|
缓存 JSON API
Elasticsearch-05Elasticsearch之查询与过滤
Elasticsearch-05Elasticsearch之查询与过滤
163 0
|
2月前
|
缓存 算法 索引
【Elasticsearch专栏 07】深入探索:Elasticsearch的倒排索引如何进行模糊查询和通配符查询
Elasticsearch的倒排索引支持模糊查询和通配符查询,通过特定的算法和数据结构,能够实现对关键词的模糊匹配和通配符匹配。这两种查询类型提供了更灵活的搜索功能,但可能影响查询性能,需结合优化策略使用。
|
2月前
|
存储 SQL Java
聚合在Elasticsearch中的使用及示例验证
聚合在Elasticsearch中的使用及示例验证
71 0
|
2月前
|
缓存 自然语言处理 数据挖掘
一篇文章让你学会Elasticsearch中的查询
一篇文章让你学会Elasticsearch中的查询
137306 118
|
2月前
|
测试技术 定位技术 API
万字长文:一文彻底搞懂Elasticsearch中Geo数据类型查询、聚合、排序
万字长文:一文彻底搞懂Elasticsearch中Geo数据类型查询、聚合、排序
94610 140
|
2月前
|
iOS开发 索引 MacOS
Elasticsearch 聚合字段aggregate-metric-double
Elasticsearch 聚合字段aggregate-metric-double
22 0
|
2月前
|
JSON 前端开发 Java
【Elasticsearch】黑马旅游案例
【Elasticsearch】黑马旅游案例
177 0