springmvc4注解方式整合elasticsearch

本文主要关于如何使用注解的方式整合elasticsearch，并使用client的方式操作es数据库，查询数据的常用接口。

1. 注入elasticsearch TransportClient

在项目的配置文件中找到含有@Configuration的类，并在其中增加下面的@Bean

import java.net.InetAddress;
import java.net.UnknownHostException;

import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
@Bean
public TransportClient client() throws UnknownHostException {
// 设置端口名字
InetSocketTransportAddress node = new InetSocketTransportAddress(InetAddress.getByName("192.168.1.121"), 9300);
// 设置名字
Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();

TransportClient client = new PreBuiltTransportClient(settings);
client.addTransportAddress(node);
return client;
}

注入client

在需要使用es的类中注入client

@Autowired
private TransportClient client;

测试中的es表索引如下

{
"testHello": {
"mappings": {
"testHello": {
"properties": {
"ip": {
"type": "keyword"
},
"hash": {
"type": "keyword"
},
"time": {
"type": "date",
"format": "epoch_millis"
},
"type": {
"type": "keyword"
},
"value": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
}
}
}
}
}
}

常用的一些查询方法

根据type查询，按照时间降序

public Object findByType(@RequestParam("type") String type,
@RequestParam(value = "from", required = false, defaultValue = "0") int from,
@RequestParam(value = "size", required = false, defaultValue = "20") int size) {

TermQueryBuilder termQuery = QueryBuilders.termQuery("type", type);
SearchResponse response = this.client.prepareSearch().setTypes("testHello").setQuery(termQuery)
.setFrom(from).setSize(size).addSort("time", SortOrder.DESC).get();

JSONObject json = new JSONObject();
List<Map<String, Object>> list = new ArrayList<>();

long totalHits = response.getHits().getTotalHits();
json.put("total", totalHits);
SearchHit[] hits = response.getHits().getHits();
for (int i = 0; i < hits.length; i++) {
list.add(hits[i].getSource());
}
json.put("list", list);
return new ResponseEntity<>(json, HttpStatus.OK);
}

根据type和value查询，按照时间降序

public Object findByTypeAndValue(@RequestParam("type") String type, @RequestParam("value") String value,
@RequestParam(value = "from", required = false, defaultValue = "0") int from,
@RequestParam(value = "size", required = false, defaultValue = "20") int size) {

TermQueryBuilder typeQuery = QueryBuilders.termQuery("type", type);
MatchPhraseQueryBuilder valueQuery = QueryBuilders.matchPhraseQuery("value", value);

BoolQueryBuilder bool = QueryBuilders.boolQuery();
bool.must(typeQuery).must(valueQuery);
SearchResponse response = this.client.prepareSearch().setTypes("testHello").setQuery(bool).setFrom(from)
.setSize(size).addSort("time", SortOrder.DESC).get();

JSONObject json = new JSONObject();
List<Map<String, Object>> list = new ArrayList<>();

long totalHits = response.getHits().getTotalHits();
json.put("total", totalHits);
SearchHit[] hits = response.getHits().getHits();
for (int i = 0; i < hits.length; i++) {
list.add(hits[i].getSource());
}
json.put("list", list);
return new ResponseEntity<>(json, HttpStatus.OK);
}

获取与指定ip和type的数据

public Object findByIpAndType(@RequestParam("ip") String ip, @RequestParam("type") String type,
@RequestParam(value = "from", required = false, defaultValue = "0") int from,
@RequestParam(value = "size", required = false, defaultValue = "20") int size) {

TermQueryBuilder ipQuery = QueryBuilders.termQuery("ip", ip);
TermQueryBuilder typeQuery = QueryBuilders.termQuery("type", type);

BoolQueryBuilder bool = QueryBuilders.boolQuery();
bool.must(ipQuery).must(typeQuery);

SearchResponse response = this.client.prepareSearch().setTypes("testHello").setFrom(from).setSize(size)
.addSort("time", SortOrder.DESC).setQuery(bool).get();

JSONObject json = new JSONObject();
List<Map<String, Object>> list = new ArrayList<>();

long totalHits = response.getHits().getTotalHits();
json.put("total", totalHits);
SearchHit[] hits = response.getHits().getHits();
for (int i = 0; i < hits.length; i++) {
list.add(hits[i].getSource());
}
json.put("list", list);
return new ResponseEntity<>(json, HttpStatus.OK);
}

分组查询所有的type值，并查询每个type值的top-n条记录

public Object findTypeDistinct(
@RequestParam(value = "field", required = false, defaultValue = "type") String field,
@RequestParam(value = "topn", required = false, defaultValue = "10") int topn) {

AggregationBuilder aggregation = AggregationBuilders.terms("agg").field(field)
// .size(2)
.subAggregation(AggregationBuilders.topHits("top").sort("time", SortOrder.DESC).size(topn));

SearchResponse searchResponse = this.client.prepareSearch().setTypes("testHello")
.addAggregation(aggregation).get();

JSONObject json = new JSONObject();
Set<String> typeList = new HashSet<>();
Map<String, List<Object>> topnList = new HashMap<>(16);

searchResponse.getAggregations().get("agg");
Terms agg = searchResponse.getAggregations().get("agg");
for (Terms.Bucket entry : agg.getBuckets()) {
String key = entry.getKeyAsString();
typeList.add(key);
List<Object> list = new ArrayList<>();
TopHits topHits = entry.getAggregations().get("top");
for (SearchHit hit : topHits.getHits()) {
list.add(hit.getSourceAsMap());
}
topnList.put(key, list);
}
json.put("typeList", typeList);
json.put("topnList", topnList);

return new ResponseEntity<>(json, HttpStatus.OK);
}

获取所有不同的ip值，类似MySQL中的distinct操作,cardinality获取ip数量，collapse获取ip值，并支持分页，agg聚合并不使用，因为ip的值比较多，超出了聚合桶 的数量，而且不支持分页

public Object findIpDistinct(@RequestParam(value = "from", required = false, defaultValue = "0") int from,
@RequestParam(value = "size", required = false, defaultValue = "20") int size) {

JSONObject json = new JSONObject();

CardinalityAggregationBuilder field = AggregationBuilders.cardinality("ip").field("ip");
SearchResponse response = this.client.prepareSearch().setTypes("testHello").addAggregation(field).get();
InternalCardinality ca = response.getAggregations().get("ip");
long total = ca.getValue();
json.put("total", total);

CollapseBuilder collapse = new CollapseBuilder("ip");
SearchResponse searchResponse = this.client.prepareSearch().setTypes("testHello").setCollapse(collapse)
.addSort("time", SortOrder.DESC).setFrom(from).setSize(size).get();

SearchHit[] hits = searchResponse.getHits().getHits();
List<Object> list = new ArrayList<>();
for (SearchHit hit : hits) {
list.add(hit.getSourceAsMap());
}
json.put("list", list);
return new ResponseEntity<>(json, HttpStatus.OK);
}