IceStreamLab 2019-11-18
elasticsearch:7.0.0
kibana:7.0.0
JDK: 1.8.0_201
maven: 3.6.1
详细请查看:
elasticsearch学习笔记(四)——在windows上安装和启动Elasticsearch
https://segmentfault.com/a/11...
增量同步mysql 数据到elasticsearch canal adapter方式(binlog)实现(从零到一超级详
https://segmentfault.com/a/11...
里面有做详细讲解,对于搭建ES集群目前还没有写相关文章,读者可以自行google
<dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch</artifactId> <version>7.0.0</version> </dependency> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>7.0.0</version> </dependency>
我的ES机器IP是192.168.254.131,读者可自行更换
RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200, "http") ) ); client.close();
对于多个ip的情况,直接使用如下格式即可:
RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("IP", PORT, "http"), new HttpHost("IP", PORT, "http") ... ) ); client.close();
我个人比较喜欢这种格式的实现:
还有基于json字符串和map的,读者可查看:
https://www.elastic.co/guide/...
格式如下
XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field(name, value); builder.field(name, value); builder.field(name, value); builder.field(name, value); builder.field(name, value); ... } builder.endObject(); IndexRequest request = new IndexRequest(index).id(id).source(builder); IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
示例:
XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("name", "jack"); builder.field("age", 27); builder.field("position", "technique"); builder.field("country", "china"); builder.field("join_date", "2017-01-01"); builder.field("salary", 10000); } builder.endObject(); IndexRequest request = new IndexRequest("employee").id("1").source(builder); IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
详细请查看:
https://www.elastic.co/guide/...
格式:
GetRequest request = new GetRequest(index, id); GetResponse response = client.get(request, RequestOptions.DEFAULT);
示例:
GetRequest request = new GetRequest("employee", "1"); GetResponse response = client.get(request, RequestOptions.DEFAULT);
详细请查看:
https://www.elastic.co/guide/...
格式:
XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field(name, value); } builder.endObject(); UpdateRequest request = new UpdateRequest(index, id).doc(builder); UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
示例:
XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("position", "technique"); } builder.endObject(); UpdateRequest request = new UpdateRequest("employee", "1").doc(builder); UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
详细请查看:
https://www.elastic.co/guide/...
格式:
DeleteRequest request = new DeleteRequest(index, id); DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
示例:
DeleteRequest request = new DeleteRequest("employee", "1"); DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
完整测试代码:
/** * 创建员工信息 * @param client * @throws Exception */ private static void createEmployee(RestHighLevelClient client) throws Exception { XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("name", "jack"); builder.field("age", 27); builder.field("position", "technique"); builder.field("country", "china"); builder.field("join_date", "2017-01-01"); builder.field("salary", 10000); } builder.endObject(); IndexRequest request = new IndexRequest("employee").id("1").source(builder); IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT); System.out.println(indexResponse.getResult()); } /** * 获取员工信息 * @param client */ private static void getEmployee(RestHighLevelClient client) throws IOException { GetRequest request = new GetRequest("employee", "1"); GetResponse response = client.get(request, RequestOptions.DEFAULT); System.out.println(response.getSourceAsString()); } /** * 修改员工信息 * @param client */ private static void updateEmployee(RestHighLevelClient client) throws IOException { XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("position", "technique"); } builder.endObject(); UpdateRequest request = new UpdateRequest("employee", "1").doc(builder); UpdateResponse response = client.update(request, RequestOptions.DEFAULT); System.out.println(response.getResult()); } /** * 删除员工信息 * @param client */ private static void deleteEmployee(RestHighLevelClient client) throws IOException { DeleteRequest request = new DeleteRequest("employee", "1"); DeleteResponse response = client.delete(request, RequestOptions.DEFAULT); System.out.println(response.getResult()); }
1、测试新增
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200, "http") ) ); createEmployee(client); client.close(); }
结果:
CREATED
2、测试获取数据
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200, "http") ) ); getEmployee(client); client.close(); }
结果:
{"name":"jack","age":27,"position":"technique manager","country":"china","join_date":"2017-01-01","salary":10000}
3、测试更新数据
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200, "http") ) ); getEmployee(client); updateEmployee(client); getEmployee(client) client.close(); }
结果:
{"name":"jack","age":27,"position":"technique manager","country":"china","join_date":"2017-01-01","salary":10000} UPDATE {"name":"jack","age":27,"position":"technique","country":"china","join_date":"2017-01-01","salary":10000}
4、测试删除数据
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200, "http") ) ); getEmployee(client); deleteEmployee(client); getEmployee(client); client.close(); }
结果:
{"name":"jack","age":27,"position":"technique","country":"china","join_date":"2017-01-01","salary":10000} DELETED null
需求:
(1)搜索职位中包含technique的员工
(2)同时要求age在30到40岁之间
(3)分页查询,查找第一页
准备数据:
private static void prepareData(RestHighLevelClient client) throws IOException { client.index(new IndexRequest("employee").id("1").source(XContentFactory.jsonBuilder() .startObject() .field("name", "jack") .field("age", 27) .field("position", "technique software") .field("country", "china") .field("join_date", "2017-01-01") .field("salary", 10000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("2").source(XContentFactory.jsonBuilder() .startObject() .field("name", "marry") .field("age", 35) .field("position", "technique manager") .field("country", "china") .field("join_date", "2017-01-01") .field("salary", 12000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("3").source(XContentFactory.jsonBuilder() .startObject() .field("name", "tom") .field("age", 32) .field("position", "senior technique software") .field("country", "china") .field("join_date", "2016-01-01") .field("salary", 11000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("4").source(XContentFactory.jsonBuilder() .startObject() .field("name", "jen") .field("age", 25) .field("position", "junior finance") .field("country", "usa") .field("join_date", "2016-01-01") .field("salary", 7000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("5").source(XContentFactory.jsonBuilder() .startObject() .field("name", "mike") .field("age", 37) .field("position", "finance manager") .field("country", "usa") .field("join_date", "2015-01-01") .field("salary", 15000) .endObject() ), RequestOptions.DEFAULT); }
执行:
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200) ) ); prepareData(client); client.close(); }
然后构建查询语句:
GET /employee/_search { "query": { "bool": { "must": [ { "match": { "position": "technique" } } ], "filter": { "range": { "age": { "gte": 30, "lte": 40 } } } } }, "from": 0, "size": 1 } { "took" : 3, "timed_out" : false, "_shards" : { "total" : 1, "successful" : 1, "skipped" : 0, "failed" : 0 }, "hits" : { "total" : { "value" : 2, "relation" : "eq" }, "max_score" : 0.5598161, "hits" : [ { "_index" : "employee", "_type" : "_doc", "_id" : "2", "_score" : 0.5598161, "_source" : { "name" : "marry", "age" : 35, "position" : "technique manager", "country" : "china", "join_date" : "2017-01-01", "salary" : 12000 } } ] } }
代码:
private static void executeSearch(RestHighLevelClient client) throws IOException { SearchSourceBuilder builder = new SearchSourceBuilder(); builder.query(QueryBuilders.matchQuery("position", "technique")); builder.postFilter(QueryBuilders.rangeQuery("age").from(30).to(40)); builder.from(0); builder.size(1); SearchRequest request = new SearchRequest(); request.indices("employee"); request.source(builder); SearchResponse response = client.search(request, RequestOptions.DEFAULT); SearchHit[] hits = response.getHits().getHits(); for (SearchHit hit : hits) { System.out.println(hit.getSourceAsString()); } }
执行:
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200) ) ); executeSearch(client); client.close(); }
结果:
{"name":"marry","age":35,"position":"technique manager","country":"china","join_date":"2017-01-01","salary":12000}
需求:
(1)首先按照country国家来进行分组
(2)然后在每个country分组内,再按照入职年限进行分组
(3)最后计算每个分组内的平均薪资
GET /employee/_search { "size": 0, "aggs": { "group_by_country": { "terms": { "field": "country.keyword" }, "aggs": { "group_by_join_date": { "date_histogram": { "field": "join_date", "interval": "year" }, "aggs": { "avg_salary": { "avg": { "field": "salary" } } } } } } } }
{ "took" : 46, "timed_out" : false, "_shards" : { "total" : 1, "successful" : 1, "skipped" : 0, "failed" : 0 }, "hits" : { "total" : { "value" : 5, "relation" : "eq" }, "max_score" : null, "hits" : [ ] }, "aggregations" : { "group_by_country" : { "doc_count_error_upper_bound" : 0, "sum_other_doc_count" : 0, "buckets" : [ { "key" : "china", "doc_count" : 3, "group_by_join_date" : { "buckets" : [ { "key_as_string" : "2016-01-01T00:00:00.000Z", "key" : 1451606400000, "doc_count" : 1, "avg_salary" : { "value" : 11000.0 } }, { "key_as_string" : "2017-01-01T00:00:00.000Z", "key" : 1483228800000, "doc_count" : 2, "avg_salary" : { "value" : 11000.0 } } ] } }, { "key" : "usa", "doc_count" : 2, "group_by_join_date" : { "buckets" : [ { "key_as_string" : "2015-01-01T00:00:00.000Z", "key" : 1420070400000, "doc_count" : 1, "avg_salary" : { "value" : 15000.0 } }, { "key_as_string" : "2016-01-01T00:00:00.000Z", "key" : 1451606400000, "doc_count" : 1, "avg_salary" : { "value" : 7000.0 } } ] } } ] } } }
执行:
public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("192.168.254.131", 9200) ) ); SearchSourceBuilder builder = new SearchSourceBuilder(); builder.aggregation(AggregationBuilders.terms("group_by_country").field("country.keyword") .subAggregation(AggregationBuilders.dateHistogram("group_by_join_date").field("join_date") .dateHistogramInterval(DateHistogramInterval.YEAR) .subAggregation(AggregationBuilders.avg("avg_salary").field("salary")))); SearchRequest request = new SearchRequest(); request.indices("employee"); request.source(builder); SearchResponse response = client.search(request, RequestOptions.DEFAULT); Map<String, Aggregation> aggregationMap = response.getAggregations().asMap(); ParsedStringTerms group_by_country = (ParsedStringTerms) aggregationMap.get("group_by_country"); for (Terms.Bucket groupByCountryBucket : group_by_country.getBuckets()) { System.out.println(groupByCountryBucket.getKey() + ":" + groupByCountryBucket.getDocCount()); Histogram groupByJoinDate = (Histogram) groupByCountryBucket.getAggregations().asMap().get("group_by_join_date"); for (Histogram.Bucket groupByJoinDateBucket : groupByJoinDate.getBuckets()) { System.out.println(groupByJoinDateBucket.getKey() + ":" + groupByJoinDateBucket.getDocCount()); Avg avg = (Avg) groupByJoinDateBucket.getAggregations().asMap().get("avg_salary"); System.out.println(avg.getValue()); } } client.close(); }
结果:
china:3 2016-01-01T00:00Z:1 11000.0 2017-01-01T00:00Z:2 11000.0 usa:2 2015-01-01T00:00Z:1 15000.0 2016-01-01T00:00Z:1 7000.0
EmployeeCRUDApp.java
import org.apache.http.HttpHost; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentFactory; import java.io.IOException; public class EmployeeCRUDApp { public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("127.0.0.1", 9200, "http") ) ); createEmployee(client); getEmployee(client); updateEmployee(client); getEmployee(client); deleteEmployee(client); getEmployee(client); client.close(); } /** * 创建员工信息 * @param client * @throws Exception */ private static void createEmployee(RestHighLevelClient client) throws Exception { XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("name", "jack"); builder.field("age", 27); builder.field("position", "technique manager"); builder.field("country", "china"); builder.field("join_date", "2017-01-01"); builder.field("salary", 10000); } builder.endObject(); IndexRequest request = new IndexRequest("employee").id("1").source(builder); IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT); System.out.println(indexResponse.getResult()); } /** * 获取员工信息 * @param client */ private static void getEmployee(RestHighLevelClient client) throws IOException { GetRequest request = new GetRequest("employee", "1"); GetResponse response = client.get(request, RequestOptions.DEFAULT); System.out.println(response.getSourceAsString()); } /** * 修改员工信息 * @param client */ private static void updateEmployee(RestHighLevelClient client) throws IOException { XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("position", "technique"); } builder.endObject(); UpdateRequest request = new UpdateRequest("employee", "1").doc(builder); UpdateResponse response = client.update(request, RequestOptions.DEFAULT); System.out.println(response.getResult()); } /** * 删除员工信息 * @param client */ private static void deleteEmployee(RestHighLevelClient client) throws IOException { DeleteRequest request = new DeleteRequest("employee", "1"); DeleteResponse response = client.delete(request, RequestOptions.DEFAULT); System.out.println(response.getResult()); } }
EmployeeSearchApp.java
import org.apache.http.HttpHost; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.builder.SearchSourceBuilder; import java.io.IOException; /** * 员工搜索应用程序 */ public class EmployeeSearchApp { public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("127.0.0.1", 9200) ) ); prepareData(client); executeSearch(client); client.close(); } /** * 执行搜索操作 * @param client */ private static void executeSearch(RestHighLevelClient client) throws IOException { SearchSourceBuilder builder = new SearchSourceBuilder(); builder.query(QueryBuilders.matchQuery("position", "technique")); builder.postFilter(QueryBuilders.rangeQuery("age").from(30).to(40)); builder.from(0); builder.size(1); SearchRequest request = new SearchRequest(); request.indices("employee"); request.source(builder); SearchResponse response = client.search(request, RequestOptions.DEFAULT); SearchHit[] hits = response.getHits().getHits(); for (SearchHit hit : hits) { System.out.println(hit.getSourceAsString()); } } /** * 准备数据 * @param client */ private static void prepareData(RestHighLevelClient client) throws IOException { client.index(new IndexRequest("employee").id("1").source(XContentFactory.jsonBuilder() .startObject() .field("name", "jack") .field("age", 27) .field("position", "technique software") .field("country", "china") .field("join_date", "2017-01-01") .field("salary", 10000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("2").source(XContentFactory.jsonBuilder() .startObject() .field("name", "marry") .field("age", 35) .field("position", "technique manager") .field("country", "china") .field("join_date", "2017-01-01") .field("salary", 12000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("3").source(XContentFactory.jsonBuilder() .startObject() .field("name", "tom") .field("age", 32) .field("position", "senior technique software") .field("country", "china") .field("join_date", "2016-01-01") .field("salary", 11000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("4").source(XContentFactory.jsonBuilder() .startObject() .field("name", "jen") .field("age", 25) .field("position", "junior finance") .field("country", "usa") .field("join_date", "2016-01-01") .field("salary", 7000) .endObject() ), RequestOptions.DEFAULT); client.index(new IndexRequest("employee").id("5").source(XContentFactory.jsonBuilder() .startObject() .field("name", "mike") .field("age", 37) .field("position", "finance manager") .field("country", "usa") .field("join_date", "2015-01-01") .field("salary", 15000) .endObject() ), RequestOptions.DEFAULT); } }
EmployeeAggrApp.java
public class EmployeeAggrApp { public static void main(String[] args) throws Exception { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder( new HttpHost("127.0.0.1", 9200) ) ); SearchSourceBuilder builder = new SearchSourceBuilder(); builder.aggregation(AggregationBuilders.terms("group_by_country").field("country.keyword") .subAggregation(AggregationBuilders.dateHistogram("group_by_join_date").field("join_date") .dateHistogramInterval(DateHistogramInterval.YEAR) .subAggregation(AggregationBuilders.avg("avg_salary").field("salary")))); SearchRequest request = new SearchRequest(); request.indices("employee"); request.source(builder); SearchResponse response = client.search(request, RequestOptions.DEFAULT); Map<String, Aggregation> aggregationMap = response.getAggregations().asMap(); ParsedStringTerms group_by_country = (ParsedStringTerms) aggregationMap.get("group_by_country"); for (Terms.Bucket groupByCountryBucket : group_by_country.getBuckets()) { System.out.println(groupByCountryBucket.getKey() + ":" + groupByCountryBucket.getDocCount()); Histogram groupByJoinDate = (Histogram) groupByCountryBucket.getAggregations().asMap().get("group_by_join_date"); for (Histogram.Bucket groupByJoinDateBucket : groupByJoinDate.getBuckets()) { System.out.println(groupByJoinDateBucket.getKey() + ":" + groupByJoinDateBucket.getDocCount()); Avg avg = (Avg) groupByJoinDateBucket.getAggregations().asMap().get("avg_salary"); System.out.println(avg.getValue()); } } client.close(); } }
另外一部分,则需要先做聚类、分类处理,将聚合出的分类结果存入ES集群的聚类索引中。数据处理层的聚合结果存入ES中的指定索引,同时将每个聚合主题相关的数据存入每个document下面的某个field下。