devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

ElaticSearchUtil工具类封装_elasticsearch util

作者：小小林熬夜学编程 | 2024-03-15 11:22:31

赞

踩

elasticsearch util
  1. package com.xxx.util;
  2.  
  3. import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
  4. import org.elasticsearch.action.admin.indices.analyze.AnalyzeAction;
  5. import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
  6. import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse.AnalyzeToken;
  7. import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
  8. import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
  9. import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
  10. import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
  11. import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
  12. import org.elasticsearch.action.bulk.BulkRequestBuilder;
  13. import org.elasticsearch.action.bulk.BulkResponse;
  14. import org.elasticsearch.action.delete.DeleteRequestBuilder;
  15. import org.elasticsearch.action.delete.DeleteResponse;
  16. import org.elasticsearch.action.get.GetResponse;
  17. import org.elasticsearch.action.index.IndexResponse;
  18. import org.elasticsearch.action.search.SearchRequestBuilder;
  19. import org.elasticsearch.action.search.SearchResponse;
  20. import org.elasticsearch.action.search.SearchType;
  21. import org.elasticsearch.action.update.UpdateRequest;
  22. import org.elasticsearch.client.transport.TransportClient;
  23. import org.elasticsearch.cluster.node.DiscoveryNode;
  24. import org.elasticsearch.common.settings.Settings;
  25. import org.elasticsearch.common.transport.InetSocketTransportAddress;
  26. import org.elasticsearch.common.unit.TimeValue;
  27. import org.elasticsearch.index.query.BoolQueryBuilder;
  28. import org.elasticsearch.index.query.QueryBuilder;
  29. import org.elasticsearch.index.query.QueryBuilders;
  30. import org.elasticsearch.search.SearchHit;
  31. import org.elasticsearch.search.SearchHits;
  32. import org.elasticsearch.search.aggregations.AggregationBuilders;
  33. import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
  34. import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
  35. import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
  36. import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
  37. import org.elasticsearch.search.aggregations.bucket.terms.Terms;
  38. import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
  39. import org.elasticsearch.search.aggregations.metrics.avg.Avg;
  40. import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
  41. import org.elasticsearch.search.aggregations.metrics.sum.Sum;
  42. import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
  43. import org.elasticsearch.search.sort.SortBuilders;
  44. import org.elasticsearch.transport.client.PreBuiltTransportClient;
  45.  
  46. import java.net.InetAddress;
  47. import java.util.ArrayList;
  48. import java.util.HashMap;
  49. import java.util.List;
  50. import java.util.Map;
  51.  
  52. import org.json.JSONException;
  53. import org.json.JSONObject;
  54. import org.slf4j.Logger;
  55. import org.slf4j.LoggerFactory;
  56.  
  57. import com.xxx.common.StaticContant;
  58. import com.xxx.util.DateUtil;
  59.  
  60. /**
  61.  * Title: ElasticSearchUtil
  62.  * Description: ES工具类
  63.  * @author wufei
  64.  * @date 2017/12/26 下午2:07:06
  65.  */
  66. public class ElasticSearchUtil {
  67.  
  68.     private static final Logger logger = LoggerFactory.getLogger(ElasticSearchUtil.class);
  69.  
  70.     private static TransportClient client = StaticContant.ESCLIENT;
  71.     
  72.     /**
  73.      * 功能描述：服务初始化(本地测试使用,实际开发写为长连接)
  74.      * @param clusterName 集群名称
  75.      * @param ip 地址
  76.      * @param port 端口
  77.      */
  78.     @SuppressWarnings({ "resource", "static-access" })
  79. 	public ElasticSearchUtil (String clusterName, String ip, int port) {
  80.     	try {
  81.     		// 通过setting对象指定集群配置信息, 配置集群名称
  82.     		Settings settings = Settings.builder()
  83.     				.put("cluster.name", clusterName) 	//设置集群名
  84. //                    .put("client.transport.sniff", true) 		//启动嗅探功能,自动嗅探整个集群的状态，把集群中其他ES节点的ip添加到本地的客户端列表中
  85. //                    .put("client.transport.ignore_cluster_name", true)//忽略集群名字验证, 打开后集群名字不对也能连接上
  86. //                    .put("client.transport.nodes_sampler_interval", 5)//报错
  87. //                    .put("client.transport.ping_timeout", 5) 		//报错, ping等待时间
  88.                     .build();
  89.             // 创建client,通过setting来创建，若不指定则默认链接的集群名为elasticsearch,链接使用tcp协议即9300
  90.             // addTransportAddress此步骤添加IP，至少一个，其实一个就够了，因为添加了自动嗅探配置
  91.     		this.client = new PreBuiltTransportClient(settings)
  92.                     .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), port));
  93.         } catch (Exception e) {
  94.             logger.error("es init failed! " + e.getMessage());
  95.         }
  96.     }
  97.  
  98.     /**
  99.      * 查看集群信息
  100.      */
  101.     public static void getClusterInfo() {
  102.         List<DiscoveryNode> nodes = client.connectedNodes();
  103.         for (DiscoveryNode node : nodes) {
  104.             System.out.println("HostId:"+node.getHostAddress()+" hostName:"+node.getHostName()+" Address:"+node.getAddress());
  105.         }
  106.     }
  107.     
  108.     /**
  109.      * 功能描述：新建索引
  110.      * @param indexName 索引名
  111.      */
  112.     public static void createIndex(String indexName) {
  113.         try {
  114. 			if (indexExist(indexName)) {
  115. 			    System.out.println("The index " + indexName + " already exits!");
  116. 			} else {
  117. 			    CreateIndexResponse cIndexResponse = client.admin().indices()
  118. 			    		.create(new CreateIndexRequest(indexName))
  119. 			            .actionGet();
  120. 			    if (cIndexResponse.isAcknowledged()) {
  121. 			        System.out.println("create index successfully！");
  122. 			    } else {
  123. 			        System.out.println("Fail to create index!");
  124. 			    }
  125. 			}
  126. 		} catch (Exception e) {
  127. 			e.printStackTrace();
  128. 		}
  129.     }
  130.  
  131.     /**
  132.      * 功能描述：新建索引
  133.      * @param index 索引名
  134.      * @param type 类型
  135.      */
  136.     public static void createIndex(String index, String type) {
  137.         try {
  138. 			client.prepareIndex(index, type).setSource().get();
  139. 		} catch (Exception e) {
  140. 			System.out.println("Fail to create index!");
  141. 			e.printStackTrace();
  142. 		}
  143.     }
  144.  
  145.     /**
  146.      * 功能描述：删除索引
  147.      * @param index 索引名
  148.      */
  149.     public static void deleteIndex(String index) {
  150.         try {
  151. 			if (indexExist(index)) {
  152. 			    DeleteIndexResponse dResponse = client.admin().indices().prepareDelete(index)
  153. 			            .execute().actionGet();
  154. 			    if (!dResponse.isAcknowledged()) {
  155. 			        logger.info("failed to delete index " + index + "!");
  156. 			    }else {
  157. 			    	logger.info("delete index " + index + " successfully!");
  158. 			    }
  159. 			} else {
  160. 			    logger.error("the index " + index + " not exists!");
  161. 			}
  162. 		} catch (Exception e) {
  163. 			e.printStackTrace();
  164. 		}
  165.     }
  166.  
  167.     /**
  168.      * 功能描述：验证索引是否存在
  169.      * @param index 索引名
  170.      */
  171.     public static boolean indexExist(String index) {
  172.         IndicesExistsRequest inExistsRequest = new IndicesExistsRequest(index);
  173.         IndicesExistsResponse inExistsResponse = client.admin().indices()
  174.                 .exists(inExistsRequest).actionGet();
  175.         return inExistsResponse.isExists();
  176.     }
  177.  
  178.     /**
  179.      * 功能描述：插入数据
  180.      * @param index 索引名
  181.      * @param type 类型
  182.      * @param json 数据
  183.      */
  184.     @SuppressWarnings("deprecation")
  185. 	public static void insertData(String index, String type, String json) {
  186.         IndexResponse response = client.prepareIndex(index, type)
  187.                 .setSource(json)
  188.                 .get();
  189.         System.out.println(response.getVersion());
  190.     }
  191.  
  192.     /**
  193.      * 功能描述：获取所有的索引
  194.      */
  195.     public static void getAllIndex(){
  196.         ClusterStateResponse response = client.admin().cluster().prepareState().execute().actionGet();
  197.         //获取所有索引
  198.         String[] indexs=response.getState().getMetaData().getConcreteAllIndices();
  199.         System.out.println("Index总数为: " + indexs.length);
  200.         for (String index : indexs) {
  201.            System.out.println("获取的Index: " + index);
  202.         }
  203.      }
  204.     
  205.     /**
  206.      * 通过prepareIndex增加文档，参数为json字符串
  207.      * @param index 索引名
  208.      * @param type  类型
  209.      * @param _id   数据id
  210.      * @param json  数据
  211.      */
  212.     @SuppressWarnings("deprecation")
  213. 	public static void insertData(String index, String type, String _id, String json) {
  214.         IndexResponse indexResponse = client.prepareIndex(index, type).setId(_id)
  215.                 .setSource(json)
  216.                 .get();
  217.         System.out.println(indexResponse.getVersion());
  218. 		logger.info("数据插入ES成功！");
  219.     }
  220.  
  221.     /**
  222.      * 功能描述：更新数据
  223.      * @param index 索引名
  224.      * @param type  类型
  225.      * @param _id   数据id
  226.      * @param json  数据
  227.      */
  228.     @SuppressWarnings("deprecation")
  229. 	public static void updateData(String index, String type, String _id, String json){
  230.         try {
  231.             UpdateRequest updateRequest = new UpdateRequest(index, type, _id).doc(json);
  232. //            client.prepareUpdate(index, type, _id).setDoc(json).get();
  233.             client.update(updateRequest).get();
  234.         } catch (Exception e) {
  235.             logger.error("update data failed." + e.getMessage());
  236.         }
  237.     }
  238.  
  239.     /**
  240.      * 功能描述：删除指定数据
  241.      * @param index 索引名
  242.      * @param type  类型
  243.      * @param _id   数据id
  244.      */
  245.     public static void deleteData(String index, String type, String _id) {
  246.         try {
  247. 			DeleteResponse response = client.prepareDelete(index, type, _id).get();
  248. 			System.out.println(response.isFragment());
  249. 			logger.info("删除指定数据成功！");
  250. 		} catch (Exception e) {
  251. 			logger.error("删除指定数据失败！" + e);
  252. 		}
  253.     }
  254.     
  255.     /**
  256. 	 * 删除索引类型表所有数据，批量删除
  257. 	 * @param index
  258. 	 * @param type
  259. 	 */
  260. 	public static void deleteIndexTypeAllData(String index, String type) {
  261. 		SearchResponse response = client.prepareSearch(index).setTypes(type)
  262. 				.setQuery(QueryBuilders.matchAllQuery()).setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
  263. 					.setScroll(new TimeValue(60000)).setSize(10000).setExplain(false).execute().actionGet();
  264. 		BulkRequestBuilder bulkRequest = client.prepareBulk();
  265. 		while (true) {
  266. 			SearchHit[] hitArray = response.getHits().getHits();
  267. 			SearchHit hit = null;
  268. 			for (int i = 0, len = hitArray.length; i < len; i++) {
  269. 				hit = hitArray[i];
  270. 				DeleteRequestBuilder request = client.prepareDelete(index, type, hit.getId());
  271. 				bulkRequest.add(request);
  272. 			}
  273. 			BulkResponse bulkResponse = bulkRequest.execute().actionGet();
  274. 			if (bulkResponse.hasFailures()) {
  275. 				logger.error(bulkResponse.buildFailureMessage());
  276. 			}
  277. 			if (hitArray.length == 0) break;
  278. 			response = client.prepareSearchScroll(response.getScrollId())
  279. 							.setScroll(new TimeValue(60000)).execute().actionGet();
  280. 		}
  281. 	}
  282.  
  283.     /**
  284.      * 功能描述：批量插入数据
  285.      * @param index    索引名
  286.      * @param type     类型
  287.      * @param jsonList 批量数据
  288.      */
  289.     @SuppressWarnings("deprecation")
  290. 	public void bulkInsertData(String index, String type, List<String> jsonList) {
  291.         BulkRequestBuilder bulkRequest = client.prepareBulk();
  292.         jsonList.forEach(item -> {
  293.             bulkRequest.add(client.prepareIndex(index, type).setSource(item));
  294.         });
  295.         BulkResponse bulkResponse = bulkRequest.get();
  296.         if(!bulkResponse.hasFailures()) {
  297.         	System.out.println(bulkResponse.getItems().length + "条数据插入完成！");
  298.         }
  299.     }
  300.     
  301.     /**
  302.      * 通过prepareGet方法获取指定文档信息
  303.      */
  304.     public static void getOneDocument(String index, String type, String id) {
  305.     	// 搜索数据  
  306.         GetResponse response = client.prepareGet(index, type, id)
  307. //                .setOperationThreaded(false)    // 线程安全
  308.                 .get();
  309.         System.out.println(response.isExists());  // 查询结果是否存在
  310.         System.out.println("***********"+response.getSourceAsString());//获取文档信息
  311. //        System.out.println(response.toString());//获取详细信息
  312.     }
  313.     
  314.     /**
  315.      * 通过prepareSearch方法获取指定索引所有文档信息
  316.      */
  317.     public static List<Map<String, Object>> getDocuments(String index) {
  318.     	List<Map<String, Object>> mapList = new ArrayList<>();
  319.     	// 搜索数据  
  320.     	SearchResponse response = client.prepareSearch(index)
  321. //    			.setTypes("type1","type2"); //设置过滤type
  322. //    			.setTypes(SearchType.DFS_QUERY_THEN_FETCH)  精确查询
  323. //    			.setQuery(QueryBuilders.matchQuery(term, queryString));
  324. //    			.setFrom(0) //设置查询数据的位置,分页用
  325. //    			.setSize(60) //设置查询结果集的最大条数
  326. //    			.setExplain(true) //设置是否按查询匹配度排序
  327.     			.get(); //最后就是返回搜索响应信息
  328.     	System.out.println("共匹配到:"+response.getHits().getTotalHits()+"条记录!");
  329.     	SearchHit[] hits = response.getHits().getHits();
  330.     	for (SearchHit hit : hits) {
  331. 			Map<String, Object> source = hit.getSource();
  332. 			mapList.add(source);
  333. 		}
  334.         System.out.println(response.getTotalShards());//总条数
  335.         return mapList;
  336.     }
  337.     
  338.     /**
  339.      * 获取指定索引库下指定type所有文档信息
  340.      * @param index
  341.      * @param type
  342.      * @return
  343.      */
  344.     public static List<Map<String, Object>> getDocuments(String index, String type) {
  345.     	List<Map<String, Object>> mapList = new ArrayList<>();
  346.     	SearchResponse response = client.prepareSearch(index).setTypes(type).get();
  347.     	SearchHit[] hits = response.getHits().getHits();
  348.     	for (SearchHit hit : hits) {
  349. 			Map<String, Object> source = hit.getSource();
  350. 			mapList.add(source);
  351. 		}
  352.     	return mapList;
  353.     }
  354.     
  355.     /**
  356.      * 带有搜索条件的聚合查询（聚合相当于关系型数据库里面的group by）
  357.      * @param index
  358.      * @param type
  359.      * @return
  360.      */
  361.     public static Map<String, Long> searchBucketsAggregation(String index, String type) {
  362.     	long total = 0;
  363.     	Map<String, Long> rtnMap = new HashMap<>();
  364.     	SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
  365.     	//搜索条件  
  366.     	String dateStr = DateUtil.getDays();
  367.         BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
  368.         /*queryBuilder.must(QueryBuilders.matchQuery("source", "resource"))精确匹配*/
  369.         queryBuilder.filter(QueryBuilders.boolQuery()
  370.         					.must(QueryBuilders.rangeQuery("logTime")
  371.         							.gte(dateStr + "000000")
  372.         							.lte(dateStr + "235959") //时间为当天
  373.         							.format("yyyyMMddHHmmss")));//时间匹配格式
  374.         // 获取当日告警总数
  375. //        long total = searchRequestBuilder.setQuery(queryBuilder).get().getHits().getTotalHits();
  376.     	// 聚合分类（以告警类型分类）
  377.     	TermsAggregationBuilder teamAggBuilder = AggregationBuilders.terms("source_count").field("source.keyword");
  378.     	searchRequestBuilder.addAggregation(teamAggBuilder);
  379.     	// 不指定 "size":0 ，则搜索结果和聚合结果都将被返回,指定size：0则只返回聚合结果
  380.     	searchRequestBuilder.setSize(0);
  381.     	searchRequestBuilder.setQuery(queryBuilder);
  382.     	SearchResponse response = searchRequestBuilder.execute().actionGet();
  383.     	
  384. //    	System.out.println("++++++聚合分类："+response.toString());
  385.     	
  386.     	// 聚合结果处理
  387.     	Terms genders = response.getAggregations().get("source_count");
  388.         for (Terms.Bucket entry : genders.getBuckets()) {
  389.             Object key = entry.getKey();      // Term  
  390.             Long count = entry.getDocCount(); // Doc count
  391.             
  392.             rtnMap.put(key.toString(), count);
  393.             
  394. //            System.out.println("Term: "+key);  
  395. //            System.out.println("Doc count: "+count);  
  396.         }
  397.         if(!rtnMap.isEmpty()) {
  398.         	if(!rtnMap.containsKey("system")) {
  399.         		rtnMap.put("system", 0L);
  400.         	}
  401.         	if(!rtnMap.containsKey("resource")) {
  402.         		rtnMap.put("resource", 0L);
  403.         	}
  404.         	if(!rtnMap.containsKey("scheduler")) {
  405.         		rtnMap.put("scheduler", 0L);
  406.         	}
  407.         	total = rtnMap.get("system") + rtnMap.get("resource") + rtnMap.get("scheduler");
  408.         }
  409.         if(total != 0) {
  410.         	rtnMap.put("total", total);
  411.         }
  412.         return rtnMap;
  413.     }
  414.     
  415.     /**
  416.      * 当日流数据总量汇总
  417.      * @param index
  418.      * @param type
  419.      * @param dataworkerType
  420.      * @return
  421.      */
  422.     public static Map<String, Long> searchBucketsAggregation(String index, String type, String dataworkerType) {
  423.     	Map<String, Long> rtnMap = new HashMap<>();
  424.     	long count = 0;
  425.     	SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
  426.     	
  427. //    	SumAggregationBuilder sAggBuilder = AggregationBuilders.sum("agg").field("count.keyword");
  428.     	
  429.     	//搜索条件  
  430.     	String dateStr = DateUtil.getDays();
  431.         BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
  432. //        queryBuilder.must(QueryBuilders.matchQuery("dataworkerType", dataworkerType))
  433.         queryBuilder.must(QueryBuilders.matchQuery("dataworkerType", dataworkerType))
  434.         			.filter(QueryBuilders.boolQuery()
  435.         					.must(QueryBuilders.rangeQuery("logTime")
  436.         							.gte(dateStr + "000000")
  437.         							.lte(dateStr + "235959")
  438.         							.format("yyyyMMddHHmmss")));
  439.     	searchRequestBuilder.setQuery(queryBuilder);
  440. //    	searchRequestBuilder.addAggregation(sAggBuilder);
  441.     	SearchResponse response = searchRequestBuilder.execute().actionGet();
  442.     	
  443.     	System.out.println("++++++条件查询结果："+response.toString());
  444.     	
  445.     	SearchHit[] hits = response.getHits().getHits();
  446.     	for (SearchHit searchHit : hits) {
  447. 			Object object = searchHit.getSource().get("count");
  448. 			if(object != null && !"".equals(object)) {
  449. 				System.out.println("=============Count:"+object.toString());
  450. 				count += Long.parseLong(object.toString());
  451. 			}
  452. 		}
  453.     	System.out.println("======流数据量：" + count);
  454.     	rtnMap.put(dataworkerType + "Total", count);
  455.     	System.out.println("======rtnMap:"+rtnMap.toString());
  456.         return rtnMap;
  457.     }
  458.     
  459.     /**
  460. 	 * 读取索引类型表指定列名的平均值
  461. 	 * @param index
  462. 	 * @param type
  463. 	 * @param avgField
  464. 	 * @return
  465. 	 */
  466. 	public static double readIndexTypeFieldValueWithAvg(String index, String type, String avgField) {
  467. 		String avgName = avgField + "Avg";
  468. 		AvgAggregationBuilder aggregation = AggregationBuilders.avg(avgName).field(avgField);
  469. 		SearchResponse response = client.prepareSearch(index).setTypes(type)
  470. 				.setQuery(QueryBuilders.matchAllQuery())
  471. 					.addAggregation(aggregation).execute().actionGet();
  472. 		Avg avg = response.getAggregations().get(avgName);
  473. 		return avg.getValue();
  474. 	}
  475. 	
  476. 	/**
  477. 	 * 读取索引类型表指定列名的总和
  478. 	 * @param index
  479. 	 * @param type
  480. 	 * @param sumField
  481. 	 * @return
  482. 	 */
  483. 	public static Map<String, Long> readIndexTypeFieldValueWithSum(String index, String type, String dataworkerType, String sumField) {
  484. 		Map<String, Long> rtnMap = new HashMap<>();
  485. 		long count = 0;
  486. 		// 聚合结果
  487. 		String sumName = sumField + "Sum";
  488. 		// 搜索条件  
  489.     	String dateStr = DateUtil.getDays();
  490.         BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
  491.         queryBuilder.must(QueryBuilders.matchQuery("dataworkerType", dataworkerType))
  492.         			.filter(QueryBuilders.boolQuery()
  493.         					.must(QueryBuilders.rangeQuery("logTime")
  494.         							.gte(dateStr + "000000")
  495.         							.lte(dateStr + "235959")
  496.         							.format("yyyyMMddHHmmss")));
  1. 	    // 时间前缀匹配（如：20180112）
  2. //        PrefixQueryBuilder preQueryBuild = QueryBuilders.prefixQuery(dataworkerType, dateStr);
  3.         // 模糊匹配
  4. //        FuzzyQueryBuilder fuzzyQueryBuild = QueryBuilders.fuzzyQuery(name, value);
  5.         // 范围匹配
  6. //        RangeQueryBuilder rangeQueryBuild = QueryBuilders.rangeQuery(name);
  7.         // 查询字段不存在及字段值为空 filterBuilder = QueryBuilders.boolQuery().should(new BoolQueryBuilder().mustNot(existsQueryBuilder)) .should(QueryBuilders.termsQuery(field, ""));
  8. //    	ExistsQueryBuilder existsQueryBuilder = QueryBuilders.existsQuery(field);
  9.         // 对某字段求和聚合（sumField字段）
  10. 		SumAggregationBuilder aggregation = AggregationBuilders.sum(sumName).field(sumField);
  11. 		SearchResponse response = client.prepareSearch(index).setTypes(type)
  12. 				.setQuery(QueryBuilders.matchAllQuery())
  13. 				.setQuery(queryBuilder)
  14. 					.addAggregation(aggregation).execute().actionGet();
  15. 		Sum sum = response.getAggregations().get(sumName);
  16. 		count = new Double(sum.getValue()).longValue();
  17. 		rtnMap.put(dataworkerType + "Total", count);
  18.     	System.out.println("======rtnMap:"+rtnMap.toString());
  19. 		return rtnMap;
  20. 	}
  21.     
  22.     /**
  23.      * 按时间统计聚合
  24.      * @param index
  25.      * @param type
  26.      */
  27.     public static void dataHistogramAggregation(String index, String type) {  
  28.         try {  
  29.             SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);  
  30.               
  31.             DateHistogramAggregationBuilder field = AggregationBuilders.dateHistogram("sales").field("value");  
  32.             field.dateHistogramInterval(DateHistogramInterval.MONTH);  
  33. //          field.dateHistogramInterval(DateHistogramInterval.days(10))  
  34.             field.format("yyyy-MM");
  35.             
  36.             //强制返回空 buckets,既空的月份也返回
  37.             field.minDocCount(0);
  38.             
  39.             // Elasticsearch 默认只返回你的数据中最小值和最大值之间的 buckets
  40.             field.extendedBounds(new ExtendedBounds("2018-01", "2018-12"));
  41.               
  42.             searchRequestBuilder.addAggregation(field);  
  43.             searchRequestBuilder.setSize(0);  
  44.             SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();  
  45.               
  46.             System.out.println(searchResponse.toString());  
  47.               
  48.             Histogram histogram = searchResponse.getAggregations().get("sales");  
  49.             for (Histogram.Bucket entry : histogram.getBuckets()) {  
  50. //              DateTime key = (DateTime) entry.getKey();   
  51.                 String keyAsString = entry.getKeyAsString();   
  52.                 Long count = entry.getDocCount(); // Doc count  
  53.                   
  54.                 System.out.println("======="+keyAsString + "，销售" + count + "辆");  
  55.             }  
  56.         } catch (Exception e) {  
  57.              e.printStackTrace();  
  58.         }  
  59.    }
  60.     
  61.     /**
  62.      * 功能描述：统计查询
  63.      * @param index 索引名
  64.      * @param type 类型
  65.      * @param constructor 查询构造
  66.      * @param groupBy 统计字段
  67.      */
  68.     /*public Map<Object, Object> statSearch(String index, String type, ESQueryBuilderConstructor constructor, String groupBy) {
  69.         Map<Object, Object> map = new HashedMap();
  70.         SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
  71.         //排序
  72.         if (StringUtils.isNotEmpty(constructor.getAsc()))
  73.             searchRequestBuilder.addSort(constructor.getAsc(), SortOrder.ASC);
  74.         if (StringUtils.isNotEmpty(constructor.getDesc()))
  75.             searchRequestBuilder.addSort(constructor.getDesc(), SortOrder.DESC);
  76.         //设置查询体
  77.         if (null != constructor) {
  78.             searchRequestBuilder.setQuery(constructor.listBuilders());
  79.         } else {
  80.             searchRequestBuilder.setQuery(QueryBuilders.matchAllQuery());
  81.         }
  82.         int size = constructor.getSize();
  83.         if (size < 0) {
  84.             size = 0;
  85.         }
  86.         if (size > MAX) {
  87.             size = MAX;
  88.         }
  89.         //返回条目数
  90.         searchRequestBuilder.setSize(size);
  92.         searchRequestBuilder.setFrom(constructor.getFrom() < 0 ? 0 : constructor.getFrom());
  93.         SearchResponse sr = searchRequestBuilder.addAggregation(
  94.                 AggregationBuilders.terms("agg").field(groupBy)
  95.         ).get();
  97.         Terms stateAgg = sr.getAggregations().get("agg");
  98.         Iterator<Terms.Bucket> iter = (Iterator<Terms.Bucket>) stateAgg.getBuckets().iterator();
  100.         while (iter.hasNext()) {
  101.             Terms.Bucket gradeBucket = iter.next();
  102.             map.put(gradeBucket.getKey(), gradeBucket.getDocCount());
  103.         }
  105.         return map;
  106.     }*/
  107.  
  108.     /**
  109.      * 功能描述：统计查询
  110.      * @param index       索引名
  111.      * @param type 	  类型
  112.      * @param constructor 查询构造
  113.      * @param agg         自定义计算
  114.      */
  115.     /*public Map<Object, Object> statSearch(String index, String type, ESQueryBuilderConstructor constructor, AggregationBuilder agg) {
  116.         if (agg == null) {
  117.             return null;
  118.         }
  119.         Map<Object, Object> map = new HashedMap();
  120.         SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
  121.         //排序
  122.         if (StringUtils.isNotEmpty(constructor.getAsc()))
  123.             searchRequestBuilder.addSort(constructor.getAsc(), SortOrder.ASC);
  124.         if (StringUtils.isNotEmpty(constructor.getDesc()))
  125.             searchRequestBuilder.addSort(constructor.getDesc(), SortOrder.DESC);
  126.         //设置查询体
  127.         if (null != constructor) {
  128.             searchRequestBuilder.setQuery(constructor.listBuilders());
  129.         } else {
  130.             searchRequestBuilder.setQuery(QueryBuilders.matchAllQuery());
  131.         }
  132.         int size = constructor.getSize();
  133.         if (size < 0) {
  134.             size = 0;
  135.         }
  136.         if (size > MAX) {
  137.             size = MAX;
  138.         }
  139.         //返回条目数
  140.         searchRequestBuilder.setSize(size);
  142.         searchRequestBuilder.setFrom(constructor.getFrom() < 0 ? 0 : constructor.getFrom());
  143.         SearchResponse sr = searchRequestBuilder.addAggregation(
  144.                 agg
  145.         ).get();
  147.         Terms stateAgg = sr.getAggregations().get("agg");
  148.         Iterator<Terms.Bucket> iter = (Iterator<Terms.Bucket>) stateAgg.getBuckets().iterator();
  150.         while (iter.hasNext()) {
  151.             Terms.Bucket gradeBucket = iter.next();
  152.             map.put(gradeBucket.getKey(), gradeBucket.getDocCount());
  153.         }
  155.         return map;
  156.     }*/
  157.  
  158.     /**
  159.      * 范围查询
  160.      * @throws Exception
  161.      */
  162.     public static void rangeQuery() {
  163.     	
  164.     	//查询字段不存在及字段值为空 filterBuilder = QueryBuilders.boolQuery().should(new BoolQueryBuilder().mustNot(existsQueryBuilder)) .should(QueryBuilders.termsQuery(field, ""));
  165. //    	ExistsQueryBuilder existsQueryBuilder = QueryBuilders.existsQuery(field);
  166.     	
  167.         //term精确查询
  168. //        QueryBuilder queryBuilder = QueryBuilders.termQuery("age", 50) ;  //年龄等于50
  169.         //range查询
  170.         QueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("age").gt(20); //年龄大于20
  171.         SearchResponse searchResponse = client.prepareSearch("index_test")
  172.               .setTypes("type_test")
  173.               .setQuery(rangeQueryBuilder)     //query
  174.               .setPostFilter(QueryBuilders.rangeQuery("age").from(40).to(50)) // Filter
  175. //              .addSort("age", SortOrder.DESC)
  176.               .setSize(120)   // 不设置的话，默认取10条数据
  177.               .execute().actionGet();
  178.         SearchHits hits = searchResponse.getHits();
  179.         System.out.println("查到记录数："+hits.getTotalHits());
  180.         SearchHit[] searchHists = hits.getHits();
  181.         if(searchHists.length>0){
  182.            for(SearchHit hit:searchHists){
  183.               String name =  (String) hit.getSource().get("username");
  184.               Integer age = Integer.parseInt(hit.getSource().get("age").toString());
  185.               System.out.println("姓名：" + name + " 年龄：" + age);
  186.            }
  187.         }
  188.      }
  189.     
  190.     /**
  191.      * 时间范围查询
  192.      * @param index
  193.      * @param type
  194.      * @param startDate
  195.      * @param endDate
  196.      */
  197.     public static void rangeQuery(String index, String type, String startDate, String endDate) {
  198.     	BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();      
  199. //        boolQueryBuilder.must(queryBuilders);      
  200.         boolQueryBuilder.filter(QueryBuilders.boolQuery().must(      
  201.                 QueryBuilders.rangeQuery("time").gte(startDate).lte(endDate).format("yyyyMMddHHmmss")));      
  202.       
  203.         SearchResponse searchResponse = client.prepareSearch(index).setTypes(type)      
  204.                 .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)      
  205.                 .setQuery(boolQueryBuilder)  
  206.                 .addAggregation(AggregationBuilders.terms("value_count").field("value.keyword")).get();   
  207.                 //Terms Aggregation  名称为terms1_count 字段为field1   下面的也类似  
  208. //                .addAggregation(AggregationBuilders.terms("terms2_count").field("field2"));
  209.         
  210.         System.out.println("时间范围查询结果： "+searchResponse.toString());
  211.         
  212.         SearchHit[] hits = searchResponse.getHits().getHits();
  213.     	List<Map<String, Object>> mapList = new ArrayList<>();
  214.     	for (SearchHit searchHit : hits) {
  215. 			Map<String, Object> map = searchHit.getSource();
  216. 			mapList.add(map);
  217. 		}
  218.         System.out.println(mapList.toString());
  219.     }
  220.     
  221.     /**
  222.      * 功能描述：关闭链接
  223.      */
  224.     public static void close() {
  225.         client.close();
  226.     }
  227.     
  228.     /**
  229.      * 在Elasticsearch老版本中做数据遍历一般使用Scroll-Scan。Scroll是先做一次初始化搜索把所有符合搜索条件的结果缓存起来生成一个快照，
  230.      * 然后持续地、批量地从快照里拉取数据直到没有数据剩下。而这时对索引数据的插入、删除、更新都不会影响遍历结果，因此scroll 并不适合用来做实时搜索。
  231.      * Scan是搜索类型，告诉Elasticsearch不用对结果集进行排序，只要分片里还有结果可以返回，就返回一批结果。
  232.      * 在5.X版本中SearchType.SCAN已经被去掉了。根据官方文档说明，使用“_doc”做排序可以达到更高性能的Scroll查询效果，
  233.      * 这样可以遍历所有文档而不需要进行排序。
  234.      * @param index
  235.      * @param type
  236.      */
  237.     @SuppressWarnings("deprecation")
  238. 	public static void scroll(String index, String type) {
  239.     	System.out.println("scroll()方法开始.....");
  240.     	List<JSONObject> lst = new ArrayList<JSONObject>();
  241.         SearchResponse searchResponse = client.prepareSearch(index)  
  242.                 .setTypes(type)  
  243.                 .setQuery(QueryBuilders.matchAllQuery())  
  244.                 .addSort(SortBuilders.fieldSort("_doc"))
  245.                 .setSize(30)  
  246.                 // 这个游标维持多长时间  
  247.                 .setScroll(TimeValue.timeValueMinutes(8)).execute().actionGet();  
  248.           
  249.         System.out.println("getScrollId: "+searchResponse.getScrollId());  
  250.         System.out.println("匹配记录数："+searchResponse.getHits().getTotalHits());  
  251.         System.out.println("hits长度："+searchResponse.getHits().hits().length);  
  252.         for (SearchHit hit : searchResponse.getHits()) {
  253.             String json = hit.getSourceAsString();
  254.             try {
  255.             	JSONObject jsonObject = new JSONObject(json);
  256.             	lst.add(jsonObject);
  257.             } catch (JSONException e) {
  258.                 e.printStackTrace();
  259.             }
  260.         }
  261.         System.out.println("======" + lst.toString());
  262.         System.out.println("======" + lst.get(0).get("username"));
  263.         // 使用上次的scrollId继续访问  
  264. //        croll scroll = new ScrollTest2();
  265. //        do{  
  266. //            int num = scroll.scanData(esClient,searchResponse.getScrollId())
  267. //            if(num ==0) break;  
  268. //        }while(true);  
  269.         System.out.println("------------------------------END");
  270.     }
  271.     
  272.     /**
  273. 	 * 分词
  274. 	 * @param index
  275. 	 * @param text
  276. 	 */
  277. 	public static void analyze(String index, String text) {
  278. 		TransportClient client = StaticContant.ESCLIENT;
  279. 		AnalyzeRequestBuilder request = new AnalyzeRequestBuilder(client, AnalyzeAction.INSTANCE, index, text);
  280. 		request.setAnalyzer("ik");
  281. 		List<AnalyzeToken> analyzeTokens = request.execute().actionGet().getTokens();
  282. 		for (int i = 0, len = analyzeTokens.size(); i < len; i++) {
  283. 			AnalyzeToken analyzeToken = analyzeTokens.get(i);
  284. 			System.out.println(analyzeToken.getTerm());
  285. 		}
  286. 	}
  287. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/240318
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号