ElasticSearch入门之彼行我释（四）

怼烎@ 2022-08-07 05:52 99阅读 0赞

![点击查看原始大小图片][1d928eea-8562-3f0c-b31c-22c0d9c02b8b.jpg]   
  
散仙在上篇[文章][Link 1]中，介绍了关于ElasticSearch基本的增删改查的基本粒子，本篇呢，我们来学下稍微高级一点的知识：  
  
  
（1）如何在ElasticSearch中批量提交索引 ？  
（2）如何使用高级查询（包括，检索，排序，过滤，分页） ？  
（3）如何组合多个查询 ？  
（4）如何使用翻页深度查询 ？  
（5）如何使用基本的聚合查询 ？  
   
  
  
（一）首先，我们思考下，为什么要使用批量添加，这个毫无疑问，因为效率问题，举个在生活中的例子，假如我们有50个人，要去美国旅游，不使用批处理的方式是，给每一个人派一架飞机送到美国，那么这就需要50次飞机的来回往来，假如使用了批处理，现在的情况就是一个飞机坐50个人，只需一次即可把所有人都送到美国，效率可想而知，生活也有很多实际的例子，大家可以自己想想。  
  
在原生的lucene中，以及solr中，这个批处理方式，实质是控制commit的时机，比如多少个提交一次，或者超过ranbuffersize的大小后自动提交，es封装了lucene的api提供bulk的方式来批量添加，原理也是，聚集一定的数量doc，然后发送一次添加请求。  
  
  
（二）只要我们使用了全文检索，我们的业务就会有各种各样的api操作，包括，任意维度的字段查询，过滤掉某些无效的信息，然后根据某个字段排序，再取topN的结果集返回，使用数据库的小伙伴们，相信大家都不陌生，在es中，这些操作都是支持的，而且还非常高效，它能满足我们大部分的需求  
  
  
（三）在es中，我们可以查询多个index，以及多个type，这一点是非常灵活地，我们，我们可以一次组装两个毫无关系的查询，发送到es服务端进行检索，然后获取结果。  
  
  
（四）es中，通过了scorll的方式，支持深度分页查询，在数据库里，我们使用的是一个cursor游标来记录读取的偏移量，同样的在es中也支持，这样的查询方式，它通过一个scrollid记录了上一次查询的状态，能轻而易举的实现深度翻页，本质上是对了Lucene的SearchAfter的封装。  
  
（五）es中，也提供了对聚合函数的支持，比如一些max，min，avg，count，sum等支持，除此之外还支持group，facet等操作，这些功能，在电商中应用非常广泛，基于lucene的solr和es都有很好的支持。  
  
下面截图看下散仙的测试数据值：  
   
  
![点击查看原始大小图片][b918454c-3138-3309-aca6-e35e2f59114a.jpg]   
源码demo如下：

Java代码 ![收藏代码][icon_star.png]

1.  **package** com.dongliang.es; 
2.  
3.  **import** java.util.Date; 
4.  **import** java.util.Map; 
5.  **import** java.util.Map.Entry; 
6.  
7.  **import** org.apache.lucene.index.Terms; 
8.  **import** org.elasticsearch.action.bulk.BulkRequestBuilder; 
9.  **import** org.elasticsearch.action.bulk.BulkResponse; 
10. **import** org.elasticsearch.action.search.MultiSearchResponse; 
11. **import** org.elasticsearch.action.search.SearchRequestBuilder; 
12. **import** org.elasticsearch.action.search.SearchResponse; 
13. **import** org.elasticsearch.action.search.SearchType; 
14. **import** org.elasticsearch.client.Client; 
15. **import** org.elasticsearch.client.transport.TransportClient; 
16. **import** org.elasticsearch.common.transport.InetSocketTransportAddress; 
17. **import** org.elasticsearch.common.unit.TimeValue; 
18. **import** org.elasticsearch.common.xcontent.XContentBuilder; 
19. **import** org.elasticsearch.common.xcontent.XContentFactory; 
20. **import** org.elasticsearch.index.query.FilterBuilders; 
21. **import** org.elasticsearch.index.query.QueryBuilders; 
22. **import** org.elasticsearch.index.query.QueryStringQueryBuilder; 
23. **import** org.elasticsearch.search.SearchHit; 
24. **import** org.elasticsearch.search.aggregations.AggregationBuilders; 
25. **import** org.elasticsearch.search.aggregations.bucket.filters.InternalFilters.Bucket; 
26. **import** org.elasticsearch.search.sort.SortOrder; 
27. 
28. /\*\*
29.  \* @author 三劫散仙
30.  \* 搜索技术交流群：324714439 
31.  \* 一个关于elasticsearch批量提交
32.  \* 和search query的的例子
33.  \* \*\*/
34. **public** **class** ElasticSearchDao \{ 
35. 
36. 
37. //es的客户端实例
38.  Client client=**null**; 
39.  \{ 
40. //连接单台机器，注意ip和端口号，不能写错
41.  client=**new** TransportClient(). 
42.  addTransportAddress(**new** InetSocketTransportAddress("192.168.46.16", 9300)); 
43. 
44.  \} 
45. 
46. 
47. **public** **static** **void** main(String\[\] args)**throws** Exception \{ 
48.  ElasticSearchDao es=**new** ElasticSearchDao(); 
49. //es.indexdata();//索引数据
50. //es.queryComplex();
51.  es.querySimple(); 
52. //es.scorllQuery();
53. //es.mutilCombineQuery();
54. //es.aggregationQuery();
55.  \} 
56. 
57. 
58. /\*\*组合分组查询\*/
59. **public** **void** aggregationQuery()**throws** Exception\{ 
60.  SearchResponse sr = client.prepareSearch() 
61.  .setQuery(QueryBuilders.matchAllQuery()) 
62.  .addAggregation( 
63.  AggregationBuilders.terms("1").field("type") 
64.  ) 
65. // .addAggregation(
66. // AggregationBuilders.dateHistogram("agg2")
67. // .field("birth")
68. // .interval(DateHistogram.Interval.YEAR)
69. // )
70.  .execute().actionGet(); 
71. 
72. // Get your facet results
73.  org.elasticsearch.search.aggregations.bucket.terms.Terms a = sr.getAggregations().get("1"); 
74. 
75. **for**(org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket bk:a.getBuckets())\{ 
76.  System.out.println("类型: "\+bk.getKey()+" 分组统计数量 "\+bk.getDocCount()+" "); 
77.  \} 
78. 
79.  System.out.println("聚合数量："\+a.getBuckets().size()); 
80. //DateHistogram agg2 = sr.getAggregations().get("agg2");
81. //结果：
82. // 类型: 1 分组数量 2 
83. // 类型: 2 分组数量 1 
84. // 类型: 3 分组数量 1 
85. // 聚合数量：3
86.  \} 
87. 
88. 
89. 
90. 
91. /\*\*多个不一样的请求组装\*/
92. **public** **void** mutilCombineQuery()\{ 
93. 
94. //查询请求1
95.  SearchRequestBuilder srb1 =client.prepareSearch().setQuery(QueryBuilders.queryString("eng").field("address")).setSize(1); 
96. //查询请求2//matchQuery
97.  SearchRequestBuilder srb2 = client.prepareSearch().setQuery(QueryBuilders.matchQuery("title", "标题")).setSize(1); 
98. //组装查询
99.  MultiSearchResponse sr = client.prepareMultiSearch().add(srb1).add(srb2).execute().actionGet(); 
100. 
101. // You will get all individual responses from MultiSearchResponse\#getResponses()
102. **long** nbHits = 0; 
103. **for** (MultiSearchResponse.Item item : sr.getResponses()) \{ 
104.  SearchResponse response = item.getResponse(); 
105. **for**(SearchHit hits:response.getHits().getHits())\{ 
106.  String sourceAsString = hits.sourceAsString();//以字符串方式打印
107.  System.out.println(sourceAsString); 
108.  \} 
109.  nbHits += response.getHits().getTotalHits(); 
110.  \} 
111.  System.out.println("命中数据量："\+nbHits); 
112. //输出：
113. // \{"title":"我是标题","price":25.65,"type":1,"status":true,"address":"血落星域风阳星","createDate":"2015-03-16T09:56:20.440Z"\}
114. // 命中数据量：2
115. 
116.  client.close(); 
117.  \} 
118. 
119. 
120. /\*\*
121.  \* 翻页查询
122.  \* \*/
123. **public** **void** scorllQuery()**throws** Exception\{ 
124.  QueryStringQueryBuilder queryString = QueryBuilders.queryString("标题").field("title"); 
125. //TermQueryBuilder qb=QueryBuilders.termQuery("title", "我是标题");
126.  SearchResponse scrollResp = client.prepareSearch("collection1") 
127.  .setSearchType(SearchType.SCAN) 
128.  .setScroll(**new** TimeValue(60000)) 
129.  .setQuery(queryString) 
130.  .setSize(100).execute().actionGet(); //100 hits per shard will be returned for each scroll
131. 
132. 
133. **while** (**true**) \{ 
134. **for** (SearchHit hit : scrollResp.getHits().getHits()) \{ 
135. //Handle the hit...
136.  String sourceAsString = hit.sourceAsString();//以字符串方式打印
137.  System.out.println(sourceAsString); 
138.  \} 
139. //通过scrollid来实现深度翻页
140.  scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(**new** TimeValue(600000)).execute().actionGet(); 
141. //Break condition: No hits are returned
142. **if** (scrollResp.getHits().getHits().length == 0) \{ 
143. **break**; 
144.  \} 
145.  \} 
146. //输出
147. // \{"title":"我是标题","price":25.65,"type":1,"status":true,"address":"血落星域风阳星","createDate":"2015-03-16T09:56:20.440Z"\}
148. // \{"title":"标题","price":251.65,"type":1,"status":true,"address":"美国东部","createDate":"2015-03-16T10:33:58.743Z"\}
149.  client.close(); 
150. 
151.  \} 
152. 
153. /\*\*简单查询\*/
154. **public** **void** querySimple()**throws** Exception\{ 
155. 
156.  SearchResponse sp = client.prepareSearch("collection1").execute().actionGet(); 
157. **for**(SearchHit hits:sp.getHits().getHits())\{ 
158.  String sourceAsString = hits.sourceAsString();//以字符串方式打印
159.  System.out.println(sourceAsString); 
160.  \} 
161. 
162. 
163. //结果
164. // \{"title":"我是标题","price":25.65,"type":1,"status":true,"address":"血落星域风阳星","createDate":"2015-03-16T09:56:20.440Z"\}
165. // \{"title":"中国","price":205.65,"type":2,"status":true,"address":"河南洛阳","createDate":"2015-03-16T10:33:58.740Z"\}
166. // \{"title":"标题","price":251.65,"type":1,"status":true,"address":"美国东部","createDate":"2015-03-16T10:33:58.743Z"\}
167. // \{"title":"elasticsearch是一个搜索引擎","price":25.65,"type":3,"status":true,"address":"china","createDate":"2015-03-16T10:33:58.743Z"\}
168. 
169. 
170.  \} 
171. /\*\*组合查询\*\*/
172. **public** **void** queryComplex()**throws** Exception\{ 
173.  SearchResponse sp=client.prepareSearch("collection1")//检索的目录
174.  .setTypes("core1")//检索的索引
175.  .setSearchType(SearchType.DFS\_QUERY\_THEN\_FETCH)//Query type
176.  .setQuery(QueryBuilders.termQuery("type", "1"))//查询--Query 
177.  .setPostFilter(FilterBuilders.rangeFilter("price").from(10).to(550.23))//过滤 --Filter
178.  .addSort("price",SortOrder.DESC) //排序 -- sort
179.  .setFrom(0).setSize(20).setExplain(**true**)//topN方式
180.  .execute().actionGet();//执行
181.  System.out.println("本次查询命中条数: "\+sp.getHits().getTotalHits()); 
182. **for**(SearchHit hits:sp.getHits().getHits())\{ 
183. //String sourceAsString = hits.sourceAsString();//以字符串方式打印
184. //System.out.println(sourceAsString);
185.  Map<String, Object> sourceAsMap = hits.sourceAsMap(); 
186. **for**(Entry<String, Object> k:sourceAsMap.entrySet())\{ 
187.  System.out.println("name： "\+k.getKey()+" value： "\+k.getValue()); 
188.  \} 
189. 
190.  System.out.println("============================================="); 
191. 
192.  \} 
193. 
194. //结果
195. // 本次查询命中条数: 2
196. // name： title value： 标题
197. // name： price value： 251.65
198. // name： address value： 美国东部
199. // name： status value： true
200. // name： createDate value： 2015-03-16T10:33:58.743Z
201. // name： type value： 1
202. // =============================================
203. // name： title value： 我是标题
204. // name： price value： 25.65
205. // name： address value： 血落星域风阳星
206. // name： status value： true
207. // name： createDate value： 2015-03-16T09:56:20.440Z
208. // name： type value： 1
209. // =============================================
210. 
211.  client.close(); 
212.  \} 
213. 
214. 
215. 
216. /\*\*索引数据\*/
217. **public** **void** indexdata()**throws** Exception\{ 
218. 
219.  BulkRequestBuilder bulk=client.prepareBulk(); 
220. 
221.  XContentBuilder doc=XContentFactory.jsonBuilder() 
222.  .startObject() 
223.  .field("title","中国") 
224.  .field("price",205.65) 
225.  .field("type",2) 
226.  .field("status",**true**) 
227.  .field("address", "河南洛阳") 
228.  .field("createDate", **new** Date()).endObject(); 
229. //collection为索引库名，类似一个数据库，索引名为core，类似一个表
230. // client.prepareIndex("collection1", "core1").setSource(doc).execute().actionGet();
231. 
232. //批处理添加
233.  bulk.add(client.prepareIndex("collection1", "core1").setSource(doc)); 
234. 
235.  doc=XContentFactory.jsonBuilder() 
236.  .startObject() 
237.  .field("title","标题") 
238.  .field("price",251.65) 
239.  .field("type",1) 
240.  .field("status",**true**) 
241.  .field("address", "美国东部") 
242.  .field("createDate", **new** Date()).endObject(); 
243. //collection为索引库名，类似一个数据库，索引名为core，类似一个表
244. // client.prepareIndex("collection1", "core1").setSource(doc).execute().actionGet();
245. //批处理添加
246.  bulk.add(client.prepareIndex("collection1", "core1").setSource(doc)); 
247. 
248.  doc=XContentFactory.jsonBuilder() 
249.  .startObject() 
250.  .field("title","elasticsearch是一个搜索引擎") 
251.  .field("price",25.65) 
252.  .field("type",3) 
253.  .field("status",**true**) 
254.  .field("address", "china") 
255.  .field("createDate", **new** Date()).endObject(); 
256. //collection为索引库名，类似一个数据库，索引名为core，类似一个表
257. //client.prepareIndex("collection1", "core1").setSource(doc).execute().actionGet();
258. //批处理添加
259.  bulk.add(client.prepareIndex("collection1", "core1").setSource(doc)); 
260. 
261. 
262. //发一次请求，提交所有数据
263.  BulkResponse bulkResponse = bulk.execute().actionGet(); 
264. **if** (!bulkResponse.hasFailures()) \{ 
265.  System.out.println("创建索引success!"); 
266.  \} **else** \{ 
267.  System.out.println("创建索引异常:"\+bulkResponse.buildFailureMessage()); 
268.  \} 
269. 
270. 
271. 
272.  client.close();//释放资源
273. // System.out.println("索引成功！");
274. 
275. 
276. 
277. 
278.  \} 
279. 
280. 
281. 
282. 
283. 
284. \} 
285.

[1d928eea-8562-3f0c-b31c-22c0d9c02b8b.jpg]: /images/20220805/21402c3511384c109158e3341022f3fe.png
[Link 1]: http://qindongliang.iteye.com/blog/2183142
[b918454c-3138-3309-aca6-e35e2f59114a.jpg]: /images/20220805/83e4cf98aa644845b5480ce5040c8dc6.png
[icon_star.png]: /images/20220805/29cf2b55ca4741bda0618f4f3eb0ed37.png