ElasticSearch入门之彼行我释（四）

散仙在上篇文章中，介绍了关于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都有很好的支持。

下面截图看下散仙的测试数据值：



源码demo如下：

Java代码

1. package com.dongliang.es;

3. import java.util.Date;

4. import java.util.Map;

5. import java.util.Map.Entry;

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;

28. /**

29. * @author 三劫散仙

30. * 搜索技术交流群：324714439

31. * 一个关于elasticsearch批量提交

32. * 和search query的的例子

33. * **/

34. public class ElasticSearchDao {

37. //es的客户端实例

38. Client client=null;

39. {

40. //连接单台机器，注意ip和端口号，不能写错

41. client=new TransportClient().

42. addTransportAddress(new InetSocketTransportAddress("192.168.46.16", 9300));

44. }

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. }

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();

72. // Get your facet results

73. org.elasticsearch.search.aggregations.bucket.terms.Terms a = sr.getAggregations().get("1");

75. for(org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket bk:a.getBuckets()){

76. System.out.println("类型: "+bk.getKey()+" 分组统计数量 "+bk.getDocCount()+" ");

77. }

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. }

91. /**多个不一样的请求组装*/

92. public void mutilCombineQuery(){

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();

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

116. client.close();

117. }

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

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();

151. }

153. /**简单查询*/

154. public void querySimple()throws Exception{

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. }

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"}

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. }

190. System.out.println("=============================================");

192. }

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. // =============================================

211. client.close();

212. }

216. /**索引数据*/

217. public void indexdata()throws Exception{

219. BulkRequestBuilder bulk=client.prepareBulk();

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();

232. //批处理添加

233. bulk.add(client.prepareIndex("collection1", "core1").setSource(doc));

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));

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));

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. }

272. client.close();//释放资源

273. // System.out.println("索引成功！");

278. }

284. }

想了解更多有关电商互联网公司的搜索技术和大数据技术的使用，请欢迎扫码关注微信公众号：我是攻城师（woshigcs）
本公众号的内容是有关搜索和大数据技术和互联网等方面内容的分享，也是一个温馨的技术互动交流的小家园，有什么问题随时都可以留言，欢迎大家来访！