ElasticSearch之Java Api聚合分组实战

2018-05-14 16:22:06 浏览数 (1)

最近有个日志收集监控的项目采用的技术栈是ELK JAVA Spring，客户端语言使用的是Java，以后有机会的话可以试一下JavaScript Nodejs的方式，非常轻量级的组合，只不过不太适合服务化的工程，Kibana充当可视化层，功能虽然非常强大和灵活，但是需要业务人员懂Lucene的查询语法和Kibana的Dashboard仪表盘自定义功能才能玩的转，所以Kibana面向专业的开发人员和运维人员比较良好，但面向业务人员则稍微有点难度，我们这边就使用Java进行二次开发，然后前端定义几个业务人员关注的图表，然后把后端查询的数据，按照一定的维度放进去即可。 基础环境： （1）ElasticSearch1.7.2 （2）Logstash2.2.2 （3）Kibana4.1.2 （3）JDK7 （4）Spring4.2 使用到的技术点： （1）ElasticSearch的查询 （2）ElasticSearch的过滤 （3）ElasticSearch的日期聚合 （4）ElasticSearch的Terms聚合 （5）ElasticSearch的多级分组 （6）ElasticSearch Logstash的时区问题 直接上代码：

Java代码

  1. /**
  2. * Created by qindongliang on 2016/4/6.
  3. */
  4. @Repository("esDaoImpl")
  5. public class ESDaoImpl implements ESDao {
  7. private static Logger log= LoggerFactory.getLogger(ESDaoImpl.class);
  8. @Autowired
  9. private ESConf esConf;
  11. @Resource(name = "client")
  12. private Client client;
  15. @Override
  16. public MonitorCount count() {
  17. MonitorCount count=new MonitorCount();
  18. //今天的数量
  19. count.setToday_count(customCount(false,"*:*"));
  20. //今天的入库量
  21. count.setToday_store_count(customCount(false,"-save:1"));
  22. //所有的总量
  23. count.setTotal_count(customCount(true,"*:*"));
  24. //所有的入库总量
  25. count.setTotal_store_count(customCount(true,"-save:1"));
  26. return count;
  27. }
  29. private long customCount(boolean isQueryAll, String queryString){
  30. try {
  31. //今天的开始时间 比如2016-04-01 00:00:00
  32. long today_start = TimeTools.getDayTimeStamp(0);
  33. //今天的结束时间 也就是明天的开始时间 比如2016-04-02 00:00:00
  34. //一闭区间一开区间即得到一天的统计量
  35. long today_end=TimeTools.getDayTimeStamp(1);
  36. StringBuffer fq = new StringBuffer();
  37. fq.append("@timestamp:")
  38. .append(" [ ")
  39. .append(today_start)
  40. .append(" TO ")
  41. .append(today_end)
  42. .append(" } ");
  43. //构建查询请求，使用Lucene高级查询语法
  44. QueryBuilder query=QueryBuilders.queryStringQuery(queryString);
  45. //构建查询请求
  46. SearchRequestBuilder search = client.prepareSearch("crawl*").setTypes("logs");
  47. //非所有的情况下，设置日期过滤
  48. if(isQueryAll){
  49. search.setQuery(query);//查询所有
  50. }else {//加上日期过滤
  51. search.setQuery(QueryBuilders.filteredQuery(query, FilterBuilders.queryFilter(QueryBuilders.queryStringQuery(fq.toString()))));
  52. }
  53. SearchResponse r = search.get();//得到查询结果
  54. long hits = r.getHits().getTotalHits();//读取命中数量
  55. return hits;
  56. }catch (Exception e){
  57. log.error("统计日期数量出错！",e);
  58. }
  59. return 0;
  60. }
  63. @Override
  64. public List<GroupCount> query(Condition condition) {
  65. return grouyQuery(condition);
  66. }
  68. /***
  69. * @param c 查询的条件
  70. * @return 查询的结果
  71. */
  72. private List<GroupCount> grouyQuery(Condition c){
  73. //封装结果集
  74. List<GroupCount> datas=new ArrayList<>();
  75. //组装分组
  76. DateHistogramBuilder dateAgg = AggregationBuilders.dateHistogram("dateagg");
  77. //定义分组的日期字段
  78. dateAgg.field("@timestamp");
  79. //按天分组
  80. if(CountType.EACH_DAY==(c.getType())) {
  81. dateAgg.interval(DateHistogram.Interval.DAY);
  82. dateAgg.timeZone(" 8:00");
  83. dateAgg.format("yyyy-MM-dd");
  84. //按小时分组
  85. }else if(CountType.EACH_HOUR==c.getType()){
  86. dateAgg.interval(DateHistogram.Interval.HOUR);
  87. //按小时分组，必须使用这个方法，不然得到的结果不正确
  88. dateAgg.postZone(" 8:00");
  89. dateAgg.format("yyyy-MM-dd HH");
  90. //无效分组
  91. }else{
  92. throw new NullPointerException("无效的枚举类型");
  93. }
  94. //二级分组，统计入库的成功失败量 0 1 2 , 1为不成功
  95. dateAgg.subAggregation(AggregationBuilders.terms("success").field("save"));
  97. //查询过滤条件
  98. StringBuffer fq = new StringBuffer();
  99. //过滤时间字段
  100. fq.append(" @timestamp:")
  101. .append(" [ ")
  102. .append(c.getStart().getTime())
  103. .append(" TO ")
  104. .append(c.getEnd().getTime())
  105. .append(" } ");
  106. //过滤一级
  107. if(StringUtils.isNotEmpty(c.getT1())){
  108. fq.append(" t1:").append(c.getT1());
  109. }
  110. //过滤二级
  111. if(StringUtils.isNotEmpty(c.getT2())){
  112. fq.append(" t2:").append(c.getT2());
  113. }
  114. //过滤三级
  115. if(StringUtils.isNotEmpty(c.getT3())){
  116. fq.append(" t3:").append(c.getT3());
  117. }
  118. //过滤url
  119. if(StringUtils.isNotEmpty(c.getSourceUrl())){
  120. //对url进行转义，防止查询出现错误
  121. fq.append(" url:").append(QueryParserBase.escape(c.getSourceUrl()));
  122. }
  123. //过滤省份编码
  124. if(StringUtils.isNotEmpty(c.getProvinceCode())){
  125. fq.append(" pcode:").append(c.getProvinceCode());
  126. }
  127. //过滤入库状态
  128. if(c.getSavaState()!=null){
  129. fq.append(" save:").append(c.getSavaState().getCode());
  130. }
  131. //过滤http状态码
  132. if(c.getWebsiteState()!=null){
  133. if(!c.getWebsiteState().getCode().equals("-1")) {
  134. fq.append(" httpcode:").append(c.getWebsiteState().getCode());
  135. }else{
  136. fq.append(" -httpcode:").append("(0 110 200)");
  137. }
  138. }
  139. //过滤配置configid
  140. if(StringUtils.isNotEmpty(c.getConfigId())){
  141. fq.append(" cid:").append(c.getConfigId());
  142. }
  146. //查询索引
  147. SearchRequestBuilder search=client.prepareSearch("crawl*").setTypes("logs");
  148. //组装请求
  149. search.setQuery(QueryBuilders.filteredQuery(QueryBuilders.matchAllQuery(),
  150. FilterBuilders.queryFilter(QueryBuilders.queryStringQuery(fq.toString())
  151. .defaultOperator(QueryStringQueryBuilder.Operator.AND)
  152. ))).addAggregation(dateAgg);
  153. //获取查询结果
  154. SearchResponse r = search.get();//得到查询结果
  155. //获取一级聚合数据
  156. Histogram h=r.getAggregations().get("dateagg");
  157. //得到一级聚合结果里面的分桶集合
  158. List<DateHistogram.Bucket> buckets = (List<DateHistogram.Bucket>) h.getBuckets();
  159. //遍历分桶集
  160. for(DateHistogram.Bucket b:buckets){
  161. //读取二级聚合数据集引用
  162. Aggregations sub = b.getAggregations();
  163. //获取二级聚合集合
  164. StringTerms count = sub.get("success");
  165. GroupCount groupCount=new GroupCount();
  166. //设置x轴分组日期
  167. groupCount.setGroupKey(b.getKey());
  168. //设置指定分组条件下入库总量
  169. groupCount.setTotal_count(b.getDocCount());
  170. //读取指定分组条件下不成功的数量
  171. long bad_count=count.getBucketByKey("1")==null?0:count.getBucketByKey("1").getDocCount();
  172. //设置指定分组条件下成功的入库量
  173. groupCount.setTotal_store_count(b.getDocCount()-bad_count);
  174. //计算成功率
  175. groupCount.setSuccess_rate(groupCount.getTotal_store_count()*1.0/groupCount.getTotal_count());
  176. //添加到集合里面
  177. datas.add(groupCount);
  178. }
  179. return datas;
  180. }
  184. }

总结： （1）关于时区的问题，目前发现在测试按小时，按天分组统计的时候，时区使用的方法不是一致的，而postZone这个方法，在1.5版本已经废弃，说是使用timeZone替代，但经测试发现在按小时分组的时候，使用timeZone加8个时区的并没生效，后续看下最新版本的ElasticSearch是否修复。 （2）使用Terms的聚合分组时，这个字段最好是没有分过词的，否则大量的元数据返回，有可能会发生OOM的异常 （3）在不需要评分排名查询的场景中，尽量使用filter查询，elasticsearch会缓存查询结果，从而能大幅提高检索性能 今天先总结这么多，后续有空再关注下 （1）elasticsearch中的Aggregations和Facet的区别以及对比Solr中的Group和Facet的区别 （2）在不同的聚合渠道中多级分组中是组内有序还是全局有序

es2 java api

0 人点赞