![复制代码]()
1 package elasticsearch;
2
3 import java.util.HashMap;
4 import java.util.List;
5 import java.util.Map;
6
7 import org.elasticsearch.action.bulk.BulkItemResponse;
8 import org.elasticsearch.action.bulk.BulkRequestBuilder;
9 import org.elasticsearch.action.bulk.BulkResponse;
10 import org.elasticsearch.action.delete.DeleteRequest;
11 import org.elasticsearch.action.delete.DeleteResponse;
12 import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
13 import org.elasticsearch.action.get.GetResponse;
14 import org.elasticsearch.action.index.IndexRequest;
15 import org.elasticsearch.action.index.IndexResponse;
16 import org.elasticsearch.action.search.SearchResponse;
17 import org.elasticsearch.action.search.SearchType;
18 import org.elasticsearch.action.update.UpdateResponse;
19 import org.elasticsearch.client.transport.TransportClient;
20 import org.elasticsearch.cluster.node.DiscoveryNode;
21 import org.elasticsearch.common.collect.ImmutableList;
22 import org.elasticsearch.common.settings.ImmutableSettings;
23 import org.elasticsearch.common.settings.Settings;
24 import org.elasticsearch.common.text.Text;
25 import org.elasticsearch.common.transport.InetSocketTransportAddress;
26 import org.elasticsearch.common.transport.TransportAddress;
27 import org.elasticsearch.common.xcontent.XContentBuilder;
28 import org.elasticsearch.common.xcontent.XContentFactory;
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.terms.Terms;
34 import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
35 import org.elasticsearch.search.aggregations.metrics.sum.Sum;
36 import org.elasticsearch.search.highlight.HighlightField;
37 import org.elasticsearch.search.sort.SortOrder;
38 import org.junit.Before;
39 import org.junit.Test;
40
41 import com.fasterxml.jackson.databind.ObjectMapper;
42
43 public class TestEs {
44
45 TransportClient transportClient = new TransportClient();
46 //before表示在执行每个test方法之前运行
47 @Before
48 public void test0() throws Exception {
49 //指定es集群中的节点信息
50 TransportAddress transportAddress = new InetSocketTransportAddress("192.168.1.99",9300);
51 //TransportAddress transportAddress1 = new InetSocketTransportAddress("192.168.1.98",9300);
52 transportClient.addTransportAddresses(transportAddress);
53 }
54
55 /**
56 * 通过java代码操作es-1
57 * 在实际工作中这样写不是很靠谱,需要完善,做测试可以
58 * @throws Exception
59 */
60 @Test
61 public void test1() throws Exception {
62 //通过TransportClient可以和es集群交互
63 //TransportClient transportClient = new TransportClient();
64
65 //指定es集群中的节点信息, 这个地方指定的端口是节点和节点之间的通信端口是9300,不是Http请求的端口9200.
66 TransportAddress transportAddress = new InetSocketTransportAddress("192.168.1.99",9300);
67 TransportAddress transportAddress1 = new InetSocketTransportAddress("192.168.1.98",9300);
68 transportClient.addTransportAddresses(transportAddress,transportAddress1);//加入多个地址
69
70 //获取当前transportClient连接到了集群多少个节点
71 ImmutableList<DiscoveryNode> connectedNodes = transportClient.connectedNodes();
72 for (DiscoveryNode discoveryNode : connectedNodes) {
73 System.out.println(discoveryNode.getHostAddress());//打印192.168.1.99;192.168.1.98
74 //如果加入transportClient.addTransportAddresses(transportAddress) 只有一个ip,打印的就只有一个.
75 }
76 }
77
78
79
80 /**
81 * 通过java代码操作es-2
82 * 实际工作中使用的时候建议加上下面这些配置信息
83 * @throws Exception
84 */
85 @Test
86 public void test2() throws Exception {
87 //指定es的配置信息 Immutable不可改变的;
88 Settings settings = ImmutableSettings.settingsBuilder()
89 .put("cluster.name", "elasticsearch")//如果集群名称在配置文件中被修改了，那么在这需要显式定义一下
90 //es集群名称默认是 elasticsearch sniff嗅; 发现;
91 .put("client.transport.sniff", true)//开启集群的嗅探功能，这样可以保证es会自动把集群中的其他节点信息添加到transportClient里面
92 //开启嗅探功能后 只要指定集群中的任意一个可用节点就可以了.当把代码运行之后TransportClient里面会把集群中所有节点的信息都拿到,能识别集群中的所有节点.
93 .build();
94
95 //通过TransportClient可以和es集群交互
96 TransportClient transportClient = new TransportClient(settings);
97 //指定es集群中的节点信息
98 TransportAddress transportAddress = new InetSocketTransportAddress("192.168.1.99",9300);
99 transportClient.addTransportAddresses(transportAddress);
100
101 //获取当前transportClient连接到了集群多少个节点
102 ImmutableList<DiscoveryNode> connectedNodes = transportClient.connectedNodes();
103 for (DiscoveryNode discoveryNode : connectedNodes) {
104 System.out.println(discoveryNode.getHostAddress()); //虽然前面只指定了1.99 但是打印192.168.1.99 192.168.1.98
105 }
106 }
107
108 String index = "crxy";
109 String type = "emp";
110 /**
111 * index-1
112 * @throws Exception
113 */
114 @Test
115 public void test3() throws Exception {
116 String jsonStr = "{\"name\":\"zs\",\"age\":20}";//向索引库中传入一个String字符串,还可以接受其他类型
117 IndexResponse indexResponse = transportClient.prepareIndex(index, type, "7")//添加一个id=7的数据
118 .setSource(jsonStr)
119 //.execute().actionGet(); 这个和下面的get()方法是一样的,get()就是对.execute().actionGet() 进行了封装
120 .get();//执行
121 System.out.println(indexResponse.getVersion());//得到这个数据的version,如果version=1代表是新添加的数据
122 }
123
124 /**
125 * index-2
126 * 实际工作中使用
127 * @throws Exception
128 */
129 @Test
130 public void test4() throws Exception {//把hashmap类型的数据放入index库
131 HashMap<String, Object> hashMap = new HashMap<String, Object>();
132 hashMap.put("name", "heeh");
133 hashMap.put("age", 20);
134 IndexResponse indexResponse = transportClient.prepareIndex(index, type, "8").setSource(hashMap).get();
135 System.out.println(indexResponse.getVersion());
136 }
137
138 /**
139 * index -3
140 * 实际工作中使用
141 * 使用对象的时候需要把对象中的属性转化成json字符串
142 *
143 * <dependency>
144 <groupId>com.fasterxml.jackson.core</groupId>
145 <artifactId>jackson-databind</artifactId>
146 <version>2.1.3</version>
147 </dependency>
148 * @throws Exception
149 */
150 @Test
151 public void test5() throws Exception {//传入一个对象到index索引库
152 Person person = new Person();
153 person.setName("lisi");
154 person.setAge(30);
155
156 //如果直接传入一个person对象会报错,java.lang.IllegalArgumentException,必须把对象转换成一个Json字符串,使用jackson依赖
157 //IndexResponse indexResponse = transportClient.prepareIndex(index, type, "9").setSource(person).get();
158
159 ObjectMapper objectMapper = new ObjectMapper();
160 IndexResponse indexResponse = transportClient.prepareIndex(index, type, "9").setSource(objectMapper.writeValueAsString(person)).get();
161 System.out.println(indexResponse.getVersion());
162 }
163
164 /**
165 * index -4
166 * 测试数据这样使用
167 * @throws Exception
168 */
169 @Test
170 public void test6() throws Exception {
171 XContentBuilder builder = XContentFactory.jsonBuilder()//XContentFactory 这个是ES官方提供的可以构建Json字符串的工具类.
172 .startObject()//{
173 .field("name", "zs")
174 .field("age", 18)
175 .endObject();//}
176 IndexResponse indexResponse = transportClient.prepareIndex(index, type, "11").setSource(builder).get();
177 System.out.println(indexResponse.getVersion());
178 }
179
180
181
182 /**
183 * 查询 通过id
184 * @throws Exception
185 */
186 @Test
187 public void test7() throws Exception {
188 GetResponse getResponse = transportClient.prepareGet(index, type, "9").get();//查询id为9的数据
189 System.out.println(getResponse.getSourceAsString());
190 }
191
192 /**
193 * 局部更新
194 * @throws Exception
195 */
196 @Test
197 public void test8() throws Exception {
198 XContentBuilder builder = XContentFactory.jsonBuilder().startObject().field("age", 19).endObject();
199 UpdateResponse updateResponse = transportClient.prepareUpdate(index, type, "9").setDoc(builder).get();
200 System.out.println(updateResponse.getVersion());//version打印2 数据更新
201 }
202
203 /**
204 * 删除-通过id删除
205 * @throws Exception
206 */
207 @Test
208 public void test9() throws Exception {
209 DeleteResponse deleteResponse = transportClient.prepareDelete(index, type, "5").get();//删除比较简单
210 }
211
212 /**
213 * 求总数
214 * 类似于mysql中的select count(*)
215 * @throws Exception
216 */
217 @Test
218 public void test10() throws Exception {//查找索引库中的数据个数
219 long count = transportClient.prepareCount(index).get().getCount();
220 System.out.println(count);
221 }
222
223
224 /**
225 * 批量操作 bulk
226 * @throws Exception
227 */
228 @Test
229 public void test11() throws Exception {
230 BulkRequestBuilder bulkBuilder = transportClient.prepareBulk();
231
232 IndexRequest indexRequest = new IndexRequest(index, type, "12");
233 XContentBuilder builder = XContentFactory.jsonBuilder().startObject().field("name", "haha").field("age", 18).endObject();
234 indexRequest.source(builder);
235
236 DeleteRequest deleteRequest = new DeleteRequest(index, type, "13");
237
238 bulkBuilder.add(indexRequest);//bulkBuilder中可以添加多个操作,这里一个是建立索引的操作,一个是删除的操作.
239 bulkBuilder.add(deleteRequest);
240
241 BulkResponse bulkResponse = bulkBuilder.get();
242 if(bulkResponse.hasFailures()){//批量操作中可能有的操作会出现问题,这个地方对操作失败的处理
243 //获取所有错误信息，并打印
244 BulkItemResponse[] items = bulkResponse.getItems();
245 for (BulkItemResponse bulkItemResponse : items) {
246 System.out.println(bulkItemResponse.getFailureMessage());
247 }
248 }else{
249 System.out.println("全部OK");
250 }
251
252 }
253
254
255 /**
256 * 查询
257 * lt:小于
258 * lte:小于等于
259 * gt:大于
260 * gte:大于等于
261 *
262 * @throws Exception
263 */
264 @Test
265 public void test12() throws Exception {
266 SearchResponse searchResponse = transportClient.prepareSearch(index)//指定索引库
267 .setTypes(type)//指定类型
268 .setQuery(QueryBuilders.matchQuery("name", "zs"))//指定查询条件,不支持通配符
269 //.setQuery(QueryBuilders.multiMatchQuery("zs", "name","title"))//根据多个属性进行查询
270 //.setQuery(QueryBuilders.matchAllQuery())//查询所有
271 //.setQuery(QueryBuilders.queryString("name:z* AND age:20"))//支持通配符* ?，可以实现复杂查询，可以使用AND OR 之类的运算符(运算符要大写)
272 //.setQuery(QueryBuilders.termQuery("name", "zs"))//在查询的时候不分词，主要针对 人名 地名等特殊的词语
273 //工作中没有说明特殊需求，就是用默认的查询类型，如果对搜索准确度要求非常高，建议使用DFS_QUERY_THEN_FETCH,如果只追求查询效果，对其他的指标不关心，可以使用QUERY_AND_FETCH
274 .setSearchType(SearchType.QUERY_THEN_FETCH)//指定查询类型，可以指定四种
275 .setExplain(true)//返回的数据按照搜索词的相关度排序
276 //分页参数
277 .setFrom(0)
278 .setSize(10)
279 //根据某一个字段排序
280 .addSort("age",SortOrder.DESC)
281 //过滤
282 //.setPostFilter(FilterBuilders.rangeFilter("age").from(0).to(18).includeLower(true).includeUpper(false))//默认是闭区间
283 //.setPostFilter(FilterBuilders.rangeFilter("age").gt(0).lt(18))
284 //实现高亮
285 .addHighlightedField("name")//设置高亮字段
286 .setHighlighterPreTags("<font color='red'>")//设置高亮前缀和后缀
287 .setHighlighterPostTags("</font>")
288 .get();
289 SearchHits hits = searchResponse.getHits();
290 long totalHits = hits.getTotalHits();
291 System.out.println("总数："+totalHits);
292 SearchHit[] hits2 = hits.getHits();
293 for (SearchHit searchHit : hits2) {
294 //获取高亮内容
295 Map<String, HighlightField> highlightFields = searchHit.getHighlightFields();
296 HighlightField highlightField = highlightFields.get("name");
297 if(highlightField!=null){
298 String name_h = "";
299 Text[] fragments = highlightField.fragments();
300 for (Text text : fragments) {
301 name_h+=text;
302 }
303 System.out.println("高亮内容："+name_h);
304 }
305 System.out.println(searchHit.getSourceAsString());
306 }
307 }
308
309 //ES是一个分布式是搜索引擎,天生就是为分布式而生的,但是分布式有优点,也是有缺点.
310 //一个索引库crxy的数据分布到了5个分片,去哪个分片去查,由于是分布式,可能每个分片中都有数据,所以一定要到所有分片中去查找.
311
312
313
314 /**
315 * 根据查询条件删除数据
316 * @throws Exception
317 */
318 @Test
319 public void test13() throws Exception {
320 DeleteByQueryResponse deleteByQueryResponse = transportClient.prepareDeleteByQuery(index)
321 .setQuery(QueryBuilders.matchAllQuery())
322 .get();
323 }
324
325
326 /**
327 * 统计分析-count
328 * 根据年龄进行分组，统计相同年龄的数据有多少条
329 *
330 * 默认情况下，如果分组个数大于10条的话，默认只会返回前10条分组数据
331 * 如果想获取所有分组数据，或者想要获取指定数量的分组数据，如何实现呢？
332 * .size(0)
333 *
334 * @throws Exception
335 */
336 @Test
337 public void test14() throws Exception {
338 SearchResponse searchResponse = transportClient.prepareSearch(index)
339 .setTypes(type)
340 .addAggregation(AggregationBuilders.terms("age_term").field("age").size(0))//给分组起个名称，并且指定分组字段
341 .get();
342
343 Terms terms = searchResponse.getAggregations().get("age_term");//指定分组的名字.
344 //获取分组数据
345 List<Bucket> buckets = terms.getBuckets();
346 for (Bucket bucket : buckets) {
347 System.out.println(bucket.getKey()+"----"+bucket.getDocCount());
348 }
349 }
350 /*打印
351 21----3
352 18----2
353 19----1
354 20----1
355 */
356
357
358 /**
359 * 统计分析-sum select name,sum(score) from table group by name;
360 * 需要使用.subAggregation
361 * @throws Exception
362 */
363 /**
364 使用数据:
365 #aggregations-2
366 curl -XPUT 'localhost:9200/crxy/emp/1' -d'{"name":"zs","score":60}'
367 curl -XPUT 'localhost:9200/crxy/emp/2' -d'{"name":"zs","score":90}'
368 curl -XPUT 'localhost:9200/crxy/emp/3' -d'{"name":"ls","score":80}'
369 curl -XPUT 'localhost:9200/crxy/emp/4' -d'{"name":"ls","score":70}'
370 */
371 @Test
372 public void test15() throws Exception {
373 SearchResponse searchResponse = transportClient.prepareSearch(index)
374 .setTypes(type)
375 .addAggregation(AggregationBuilders.terms("name_term").field("name")//指定分组字段
376 .subAggregation(AggregationBuilders.sum("score_sum").field("score")))//指定求sum的字段
377 .get();
378
379 Terms terms = searchResponse.getAggregations().get("name_term");
380 List<Bucket> buckets = terms.getBuckets();
381 for (Bucket bucket : buckets) {
382 Sum sum = bucket.getAggregations().get("score_sum");
383 System.out.println(bucket.getKey()+"----"+sum.getValue());
384 }
385 }
386 /*
387 * 打印输出
388 ls----150.0
389 zs----150.0
390 */
391
392
393 /**
394 * 指定分片查询(_shards),指定某个节点(_only_node)和某些节点(自定义的_only_nodes)
395 * @throws Exception
396 */
397 @Test
398 public void test16() throws Exception {
399 SearchResponse searchResponse = transportClient.prepareSearch(index)
400 .setTypes(type)
401 .setQuery(QueryBuilders.matchAllQuery())
402 //.setPreference("_shards:0,2") //.setPreference("_local") .setPreference("")
403 .setPreference("_only_node:8PoWbRVvQQ6NU283Bfd_7A,BJwexRvDTJ-VRx7Njs8uxA")//8PoWbRVvQQ6NU283Bfd_7A是一个节点的id
404 .get();
405
406 SearchHits hits = searchResponse.getHits();
407 long totalHits = hits.getTotalHits();
408 System.out.println("总数："+totalHits);
409 SearchHit[] hits2 = hits.getHits();
410 for (SearchHit searchHit : hits2) {
411 System.out.println(searchHit.getSourceAsString());
412 }
413 }
414
415 }
![复制代码]()
1 @Test
2 public void test16() throws Exception {
3 SearchResponse searchResponse = transportClient.prepareSearch("crxy*")
4 .setTypes(type)
5 .setQuery(QueryBuilders.matchAllQuery())
6 //查询指定分片的数据可以使用下面这两种方案，第一种直接指定分片id，第二种根据routing参数的值计算分片id
7 //.setPreference("_shards:0")
8 //.setRouting("hehe")
9 .get();
10
11 SearchHits hits = searchResponse.getHits();
12 long totalHits = hits.getTotalHits();
13 System.out.println("总数："+totalHits);
14 SearchHit[] hits2 = hits.getHits();
15 for (SearchHit searchHit : hits2) {
16 System.out.println(searchHit.getSourceAsString());
17 }
18 }
19
20
21 @Test
22 public void test17() throws Exception {
23 HashMap<String, Object> hashMap = new HashMap<String, Object>();
24 hashMap.put("name", "hello world");
25 hashMap.put("age", 20);
26 IndexResponse indexResponse = transportClient.prepareIndex(index, type, "7")
27 .setSource(hashMap)
28 .setRouting("heha")//指定一个路由参数，参数相同的数据会保存到同一个分片
29 .get();
30 System.out.println(indexResponse.getVersion());
31 }
32
