elasticsearch系列七：ES Java客戶端-Elasticsearch Java client（ES Client 簡介、Java REST Client、Java Client、Spring Data Elasticsearch）
一、ES Client 簡介
1. ES是一個服務，采用C/S結構
2. 回顧 ES的架構
3. ES支持的客戶端連接方式
3.1 REST API ，端口 9200
這種連接方式對應於架構圖中的RESTful style API這一層，這種客戶端的連接方式是RESTful風格的，使用http的方式進行連接
3.2 Transport 連接 端口 9300
這種連接方式對應於架構圖中的Transport這一層，這種客戶端連接方式是直接連接ES的節點，使用TCP的方式進行連接
4. ES提供了多種編程語言客戶端
官網可以了解詳情：
https://www.elastic.co/guide/en/elasticsearch/client/index.html
二、Java REST Client介紹
1. ES提供了兩個JAVA REST client 版本
Java Low Level REST Client: 低級別的REST客戶端，通過http與集群交互，用戶需自己編組請求JSON串，及解析響應JSON串。兼容所有ES版本。
Java High Level REST Client: 高級別的REST客戶端，基於低級別的REST客戶端，增加了編組請求JSON串、解析響應JSON串等相關api。使用的版本需要保持和ES服務端的版本一致，否則會有版本問題。
2. Java Low Level REST Client 說明
特點，maven 引入、使用介紹： https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-low.html
API doc ：https://artifacts.elastic.co/javadoc/org/elasticsearch/client/elasticsearch-rest-client/6.2.4/index.html.
3. Java High Level REST Client 說明
從6.0.0開始加入的，目的是以java面向對象的方式來進行請求、響應處理。
每個API 支持 同步/異步 兩種方式，同步方法直接返回一個結果對象。異步的方法以async為后綴，通過listener參數來通知結果。
高級java REST 客戶端依賴Elasticsearch core project
兼容性說明：
依賴 java1.8 和 Elasticsearch core project
請使用與服務端ES版本一致的客戶端版本
4. Java High Level REST Client maven 集成
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>6.2.4</version>
</dependency>
5. Java High Level REST Client 初始化
RestHighLevelClient client = new RestHighLevelClient(
RestClient.builder(
new HttpHost("localhost", 9200, "http"),
new HttpHost("localhost", 9201, "http")));
給定集群的多個節點地址，將客戶端負載均衡地向這個節點地址集發請求
Client 不再使用了，記得關閉它：
client.close();
API及用法示例，請參考：
https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high-supported-apis.html
三、Java High Level REST Client 使用示例
准備：
編寫示例之前首先在maven工程里面引入和ES服務端版本一樣的Java客戶端
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>6.2.4</version>
</dependency>
給定集群的多個節點地址，將客戶端負載均衡地向這個節點地址集發請求：
InitDemo.java
package com.study.es_hrset_client;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
/**
*
* @Description: 獲取Java High Level REST Client客戶端
* @author lgs
* @date 2018年6月23日
*
*/
public class InitDemo {
public static RestHighLevelClient getClient() {
RestHighLevelClient client = new RestHighLevelClient(
RestClient.builder(new HttpHost("localhost", 9200, "http"),
new HttpHost("localhost", 9201, "http")));
return client;
}
}
注意使用ES的客戶端時類比之前我們在Kibana進行的ES的相關操作，這樣使用起來更加有效果
1. Create index 創建索引
CreateIndexDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
/**
*
* @Description: 創建索引
* @author lgs
* @date 2018年6月23日
*
*/
public class CreateIndexDemo {
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建 創建索引request 參數：索引名mess
CreateIndexRequest request = new CreateIndexRequest("mess");
// 2、設置索引的settings
request.settings(Settings.builder().put("index.number_of_shards", 3) // 分片數
.put("index.number_of_replicas", 2) // 副本數
.put("analysis.analyzer.default.tokenizer", "ik_smart") // 默認分詞器
);
// 3、設置索引的mappings
request.mapping("_doc",
" {\n" +
" \"_doc\": {\n" +
" \"properties\": {\n" +
" \"message\": {\n" +
" \"type\": \"text\"\n" +
" }\n" +
" }\n" +
" }\n" +
" }",
XContentType.JSON);
// 4、 設置索引的別名
request.alias(new Alias("mmm"));
// 5、 發送請求
// 5.1 同步方式發送請求
CreateIndexResponse createIndexResponse = client.indices()
.create(request);
// 6、處理響應
boolean acknowledged = createIndexResponse.isAcknowledged();
boolean shardsAcknowledged = createIndexResponse
.isShardsAcknowledged();
System.out.println("acknowledged = " + acknowledged);
System.out.println("shardsAcknowledged = " + shardsAcknowledged);
// 5.1 異步方式發送請求
/*ActionListener<CreateIndexResponse> listener = new ActionListener<CreateIndexResponse>() {
@Override
public void onResponse(
CreateIndexResponse createIndexResponse) {
// 6、處理響應
boolean acknowledged = createIndexResponse.isAcknowledged();
boolean shardsAcknowledged = createIndexResponse
.isShardsAcknowledged();
System.out.println("acknowledged = " + acknowledged);
System.out.println(
"shardsAcknowledged = " + shardsAcknowledged);
}
@Override
public void onFailure(Exception e) {
System.out.println("創建索引異常：" + e.getMessage());
}
};
client.indices().createAsync(request, listener);
*/
} catch (IOException e) {
e.printStackTrace();
}
}
}
運行結果：
acknowledged = true
shardsAcknowledged = true
2. index document
索引文檔，即往索引里面放入文檔數據.類似於數據庫里面向表里面插入一行數據，一行數據就是一個文檔
IndexDocumentDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.RestStatus;
/**
*
* @Description: 索引文檔，即往索引里面放入文檔數據.類似於數據庫里面向表里面插入一行數據，一行數據就是一個文檔
* @author lgs
* @date 2018年6月23日
*
*/
public class IndexDocumentDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建索引請求
IndexRequest request = new IndexRequest(
"mess", //索引
"_doc", // mapping type
"1"); //文檔id
// 2、准備文檔數據
// 方式一：直接給JSON串
String jsonString = "{" +
"\"user\":\"kimchy\"," +
"\"postDate\":\"2013-01-30\"," +
"\"message\":\"trying out Elasticsearch\"" +
"}";
request.source(jsonString, XContentType.JSON);
// 方式二：以map對象來表示文檔
/*
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("user", "kimchy");
jsonMap.put("postDate", new Date());
jsonMap.put("message", "trying out Elasticsearch");
request.source(jsonMap);
*/
// 方式三：用XContentBuilder來構建文檔
/*
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
builder.field("user", "kimchy");
builder.field("postDate", new Date());
builder.field("message", "trying out Elasticsearch");
}
builder.endObject();
request.source(builder);
*/
// 方式四：直接用key-value對給出
/*
request.source("user", "kimchy",
"postDate", new Date(),
"message", "trying out Elasticsearch");
*/
//3、其他的一些可選設置
/*
request.routing("routing"); //設置routing值
request.timeout(TimeValue.timeValueSeconds(1)); //設置主分片等待時長
request.setRefreshPolicy("wait_for"); //設置重刷新策略
request.version(2); //設置版本號
request.opType(DocWriteRequest.OpType.CREATE); //操作類別
*/
//4、發送請求
IndexResponse indexResponse = null;
try {
// 同步方式
indexResponse = client.index(request);
} catch(ElasticsearchException e) {
// 捕獲，並處理異常
//判斷是否版本沖突、create但文檔已存在沖突
if (e.status() == RestStatus.CONFLICT) {
logger.error("沖突了，請在此寫沖突處理邏輯！\n" + e.getDetailedMessage());
}
logger.error("索引異常", e);
}
//5、處理響應
if(indexResponse != null) {
String index = indexResponse.getIndex();
String type = indexResponse.getType();
String id = indexResponse.getId();
long version = indexResponse.getVersion();
if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
System.out.println("新增文檔成功，處理邏輯代碼寫到這里。");
} else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
System.out.println("修改文檔成功，處理邏輯代碼寫到這里。");
}
// 分片處理信息
ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
}
// 如果有分片副本失敗，可以獲得失敗原因信息
if (shardInfo.getFailed() > 0) {
for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
String reason = failure.reason();
System.out.println("副本失敗原因：" + reason);
}
}
}
//異步方式發送索引請求
/*ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
@Override
public void onResponse(IndexResponse indexResponse) {
}
@Override
public void onFailure(Exception e) {
}
};
client.indexAsync(request, listener);
*/
} catch (IOException e) {
e.printStackTrace();
}
}
}
運行結果：
新增文檔成功，處理邏輯代碼寫到這里。
3. get document
獲取文檔數據
GetDocumentDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import java.util.Map;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
/**
*
* @Description: 獲取文檔數據
* @author lgs
* @date 2018年6月23日
*
*/
public class GetDocumentDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建獲取文檔請求
GetRequest request = new GetRequest(
"mess", //索引
"_doc", // mapping type
"1"); //文檔id
// 2、可選的設置
//request.routing("routing");
//request.version(2);
//request.fetchSourceContext(new FetchSourceContext(false)); //是否獲取_source字段
//選擇返回的字段
String[] includes = new String[]{"message", "*Date"};
String[] excludes = Strings.EMPTY_ARRAY;
FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
request.fetchSourceContext(fetchSourceContext);
//也可寫成這樣
/*String[] includes = Strings.EMPTY_ARRAY;
String[] excludes = new String[]{"message"};
FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
request.fetchSourceContext(fetchSourceContext);*/
// 取stored字段
/*request.storedFields("message");
GetResponse getResponse = client.get(request);
String message = getResponse.getField("message").getValue();*/
//3、發送請求
GetResponse getResponse = null;
try {
// 同步請求
getResponse = client.get(request);
} catch (ElasticsearchException e) {
if (e.status() == RestStatus.NOT_FOUND) {
logger.error("沒有找到該id的文檔" );
}
if (e.status() == RestStatus.CONFLICT) {
logger.error("獲取時版本沖突了，請在此寫沖突處理邏輯！" );
}
logger.error("獲取文檔異常", e);
}
//4、處理響應
if(getResponse != null) {
String index = getResponse.getIndex();
String type = getResponse.getType();
String id = getResponse.getId();
if (getResponse.isExists()) { // 文檔存在
long version = getResponse.getVersion();
String sourceAsString = getResponse.getSourceAsString(); //結果取成 String
Map<String, Object> sourceAsMap = getResponse.getSourceAsMap(); // 結果取成Map
byte[] sourceAsBytes = getResponse.getSourceAsBytes(); //結果取成字節數組
logger.info("index:" + index + " type:" + type + " id:" + id);
logger.info(sourceAsString);
} else {
logger.error("沒有找到該id的文檔" );
}
}
//異步方式發送獲取文檔請求
/*
ActionListener<GetResponse> listener = new ActionListener<GetResponse>() {
@Override
public void onResponse(GetResponse getResponse) {
}
@Override
public void onFailure(Exception e) {
}
};
client.getAsync(request, listener);
*/
} catch (IOException e) {
e.printStackTrace();
}
}
}
4. Bulk
批量索引文檔，即批量往索引里面放入文檔數據.類似於數據庫里面批量向表里面插入多行數據，一行數據就是一個文檔
BulkDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
/**
*
* @Description: 批量索引文檔，即批量往索引里面放入文檔數據.類似於數據庫里面批量向表里面插入多行數據，一行數據就是一個文檔
* @author lgs
* @date 2018年6月23日
*
*/
public class BulkDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建批量操作請求
BulkRequest request = new BulkRequest();
request.add(new IndexRequest("mess", "_doc", "1")
.source(XContentType.JSON,"field", "foo"));
request.add(new IndexRequest("mess", "_doc", "2")
.source(XContentType.JSON,"field", "bar"));
request.add(new IndexRequest("mess", "_doc", "3")
.source(XContentType.JSON,"field", "baz"));
/*
request.add(new DeleteRequest("mess", "_doc", "3"));
request.add(new UpdateRequest("mess", "_doc", "2")
.doc(XContentType.JSON,"other", "test"));
request.add(new IndexRequest("mess", "_doc", "4")
.source(XContentType.JSON,"field", "baz"));
*/
// 2、可選的設置
/*
request.timeout("2m");
request.setRefreshPolicy("wait_for");
request.waitForActiveShards(2);
*/
//3、發送請求
// 同步請求
BulkResponse bulkResponse = client.bulk(request);
//4、處理響應
if(bulkResponse != null) {
for (BulkItemResponse bulkItemResponse : bulkResponse) {
DocWriteResponse itemResponse = bulkItemResponse.getResponse();
if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
|| bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
IndexResponse indexResponse = (IndexResponse) itemResponse;
//TODO 新增成功的處理
} else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
UpdateResponse updateResponse = (UpdateResponse) itemResponse;
//TODO 修改成功的處理
} else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
//TODO 刪除成功的處理
}
}
}
//異步方式發送批量操作請求
/*
ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
@Override
public void onResponse(BulkResponse bulkResponse) {
}
@Override
public void onFailure(Exception e) {
}
};
client.bulkAsync(request, listener);
*/
} catch (IOException e) {
e.printStackTrace();
}
}
}
5. search
搜索數據
SearchDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.term.TermSuggestion;
/**
*
* @Description: 搜索數據
* @author lgs
* @date 2018年6月23日
*
*/
public class SearchDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("bank");
searchRequest.types("_doc");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//構造QueryBuilder
/*QueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("user", "kimchy")
.fuzziness(Fuzziness.AUTO)
.prefixLength(3)
.maxExpansions(10);
sourceBuilder.query(matchQueryBuilder);*/
sourceBuilder.query(QueryBuilders.termQuery("age", 24));
sourceBuilder.from(0);
sourceBuilder.size(10);
sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
//是否返回_source字段
//sourceBuilder.fetchSource(false);
//設置返回哪些字段
/*String[] includeFields = new String[] {"title", "user", "innerObject.*"};
String[] excludeFields = new String[] {"_type"};
sourceBuilder.fetchSource(includeFields, excludeFields);*/
//指定排序
//sourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));
//sourceBuilder.sort(new FieldSortBuilder("_uid").order(SortOrder.ASC));
// 設置返回 profile
//sourceBuilder.profile(true);
//將請求體加入到請求中
searchRequest.source(sourceBuilder);
// 可選的設置
//searchRequest.routing("routing");
// 高亮設置
/*
HighlightBuilder highlightBuilder = new HighlightBuilder();
HighlightBuilder.Field highlightTitle =
new HighlightBuilder.Field("title");
highlightTitle.highlighterType("unified");
highlightBuilder.field(highlightTitle);
HighlightBuilder.Field highlightUser = new HighlightBuilder.Field("user");
highlightBuilder.field(highlightUser);
sourceBuilder.highlighter(highlightBuilder);*/
//加入聚合
/*TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_company")
.field("company.keyword");
aggregation.subAggregation(AggregationBuilders.avg("average_age")
.field("age"));
sourceBuilder.aggregation(aggregation);*/
//做查詢建議
/*SuggestionBuilder termSuggestionBuilder =
SuggestBuilders.termSuggestion("user").text("kmichy");
SuggestBuilder suggestBuilder = new SuggestBuilder();
suggestBuilder.addSuggestion("suggest_user", termSuggestionBuilder);
sourceBuilder.suggest(suggestBuilder);*/
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest);
//4、處理響應
//搜索結果狀態信息
RestStatus status = searchResponse.status();
TimeValue took = searchResponse.getTook();
Boolean terminatedEarly = searchResponse.isTerminatedEarly();
boolean timedOut = searchResponse.isTimedOut();
//分片搜索情況
int totalShards = searchResponse.getTotalShards();
int successfulShards = searchResponse.getSuccessfulShards();
int failedShards = searchResponse.getFailedShards();
for (ShardSearchFailure failure : searchResponse.getShardFailures()) {
// failures should be handled here
}
//處理搜索命中文檔結果
SearchHits hits = searchResponse.getHits();
long totalHits = hits.getTotalHits();
float maxScore = hits.getMaxScore();
SearchHit[] searchHits = hits.getHits();
for (SearchHit hit : searchHits) {
// do something with the SearchHit
String index = hit.getIndex();
String type = hit.getType();
String id = hit.getId();
float score = hit.getScore();
//取_source字段值
String sourceAsString = hit.getSourceAsString(); //取成json串
Map<String, Object> sourceAsMap = hit.getSourceAsMap(); // 取成map對象
//從map中取字段值
/*
String documentTitle = (String) sourceAsMap.get("title");
List<Object> users = (List<Object>) sourceAsMap.get("user");
Map<String, Object> innerObject = (Map<String, Object>) sourceAsMap.get("innerObject");
*/
logger.info("index:" + index + " type:" + type + " id:" + id);
logger.info(sourceAsString);
//取高亮結果
/*Map<String, HighlightField> highlightFields = hit.getHighlightFields();
HighlightField highlight = highlightFields.get("title");
Text[] fragments = highlight.fragments();
String fragmentString = fragments[0].string();*/
}
// 獲取聚合結果
/*
Aggregations aggregations = searchResponse.getAggregations();
Terms byCompanyAggregation = aggregations.get("by_company");
Bucket elasticBucket = byCompanyAggregation.getBucketByKey("Elastic");
Avg averageAge = elasticBucket.getAggregations().get("average_age");
double avg = averageAge.getValue();
*/
// 獲取建議結果
/*Suggest suggest = searchResponse.getSuggest();
TermSuggestion termSuggestion = suggest.getSuggestion("suggest_user");
for (TermSuggestion.Entry entry : termSuggestion.getEntries()) {
for (TermSuggestion.Entry.Option option : entry) {
String suggestText = option.getText().string();
}
}
*/
//異步方式發送獲查詢請求
/*
ActionListener<SearchResponse> listener = new ActionListener<SearchResponse>() {
@Override
public void onResponse(SearchResponse getResponse) {
//結果獲取
}
@Override
public void onFailure(Exception e) {
//失敗處理
}
};
client.searchAsync(searchRequest, listener);
*/
} catch (IOException e) {
logger.error(e);
}
}
}
6. highlight 高亮
HighlightDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import java.util.Map;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
/**
*
* @Description: 高亮
* @author lgs
* @date 2018年6月23日
*
*/
public class HighlightDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建search請求
SearchRequest searchRequest = new SearchRequest("hl_test");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//構造QueryBuilder
QueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("title", "lucene solr");
sourceBuilder.query(matchQueryBuilder);
//分頁設置
/*sourceBuilder.from(0);
sourceBuilder.size(5); ;*/
// 高亮設置
HighlightBuilder highlightBuilder = new HighlightBuilder();
highlightBuilder.requireFieldMatch(false).field("title").field("content")
.preTags("<strong>").postTags("</strong>");
//不同字段可有不同設置，如不同標簽
/*HighlightBuilder.Field highlightTitle = new HighlightBuilder.Field("title");
highlightTitle.preTags("<strong>").postTags("</strong>");
highlightBuilder.field(highlightTitle);
HighlightBuilder.Field highlightContent = new HighlightBuilder.Field("content");
highlightContent.preTags("<b>").postTags("</b>");
highlightBuilder.field(highlightContent).requireFieldMatch(false);*/
sourceBuilder.highlighter(highlightBuilder);
searchRequest.source(sourceBuilder);
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest);
//4、處理響應
if(RestStatus.OK.equals(searchResponse.status())) {
//處理搜索命中文檔結果
SearchHits hits = searchResponse.getHits();
long totalHits = hits.getTotalHits();
SearchHit[] searchHits = hits.getHits();
for (SearchHit hit : searchHits) {
String index = hit.getIndex();
String type = hit.getType();
String id = hit.getId();
float score = hit.getScore();
//取_source字段值
//String sourceAsString = hit.getSourceAsString(); //取成json串
Map<String, Object> sourceAsMap = hit.getSourceAsMap(); // 取成map對象
//從map中取字段值
/*String title = (String) sourceAsMap.get("title");
String content = (String) sourceAsMap.get("content"); */
logger.info("index:" + index + " type:" + type + " id:" + id);
logger.info("sourceMap : " + sourceAsMap);
//取高亮結果
Map<String, HighlightField> highlightFields = hit.getHighlightFields();
HighlightField highlight = highlightFields.get("title");
if(highlight != null) {
Text[] fragments = highlight.fragments(); //多值的字段會有多個值
if(fragments != null) {
String fragmentString = fragments[0].string();
logger.info("title highlight : " + fragmentString);
//可用高亮字符串替換上面sourceAsMap中的對應字段返回到上一級調用
//sourceAsMap.put("title", fragmentString);
}
}
highlight = highlightFields.get("content");
if(highlight != null) {
Text[] fragments = highlight.fragments(); //多值的字段會有多個值
if(fragments != null) {
String fragmentString = fragments[0].string();
logger.info("content highlight : " + fragmentString);
//可用高亮字符串替換上面sourceAsMap中的對應字段返回到上一級調用
//sourceAsMap.put("content", fragmentString);
}
}
}
}
} catch (IOException e) {
logger.error(e);
}
}
}
7. suggest 查詢建議
SuggestDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.SuggestionBuilder;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.term.TermSuggestion;
/**
*
* @Description: 查詢建議
* @author lgs
* @date 2018年6月23日
*
*/
public class SuggestDemo {
private static Logger logger = LogManager.getRootLogger();
//詞項建議拼寫檢查，檢查用戶的拼寫是否錯誤，如果有錯給用戶推薦正確的詞，appel->apple
public static void termSuggest() {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("mess");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.size(0);
//做查詢建議
//詞項建議
SuggestionBuilder termSuggestionBuilder =
SuggestBuilders.termSuggestion("user").text("kmichy");
SuggestBuilder suggestBuilder = new SuggestBuilder();
suggestBuilder.addSuggestion("suggest_user", termSuggestionBuilder);
sourceBuilder.suggest(suggestBuilder);
searchRequest.source(sourceBuilder);
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest);
//4、處理響應
//搜索結果狀態信息
if(RestStatus.OK.equals(searchResponse.status())) {
// 獲取建議結果
Suggest suggest = searchResponse.getSuggest();
TermSuggestion termSuggestion = suggest.getSuggestion("suggest_user");
for (TermSuggestion.Entry entry : termSuggestion.getEntries()) {
logger.info("text: " + entry.getText().string());
for (TermSuggestion.Entry.Option option : entry) {
String suggestText = option.getText().string();
logger.info(" suggest option : " + suggestText);
}
}
}
/*
"suggest": {
"my-suggestion": [
{
"text": "tring",
"offset": 0,
"length": 5,
"options": [
{
"text": "trying",
"score": 0.8,
"freq": 1
}
]
},
{
"text": "out",
"offset": 6,
"length": 3,
"options": []
},
{
"text": "elasticsearch",
"offset": 10,
"length": 13,
"options": []
}
]
}*/
} catch (IOException e) {
logger.error(e);
}
}
//自動補全，根據用戶的輸入聯想到可能的詞或者短語
public static void completionSuggester() {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("music");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.size(0);
//做查詢建議
//自動補全
/*POST music/_search?pretty
{
"suggest": {
"song-suggest" : {
"prefix" : "lucene s",
"completion" : {
"field" : "suggest" ,
"skip_duplicates": true
}
}
}
}*/
SuggestionBuilder termSuggestionBuilder =
SuggestBuilders.completionSuggestion("suggest").prefix("lucene s")
.skipDuplicates(true);
SuggestBuilder suggestBuilder = new SuggestBuilder();
suggestBuilder.addSuggestion("song-suggest", termSuggestionBuilder);
sourceBuilder.suggest(suggestBuilder);
searchRequest.source(sourceBuilder);
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest);
//4、處理響應
//搜索結果狀態信息
if(RestStatus.OK.equals(searchResponse.status())) {
// 獲取建議結果
Suggest suggest = searchResponse.getSuggest();
CompletionSuggestion termSuggestion = suggest.getSuggestion("song-suggest");
for (CompletionSuggestion.Entry entry : termSuggestion.getEntries()) {
logger.info("text: " + entry.getText().string());
for (CompletionSuggestion.Entry.Option option : entry) {
String suggestText = option.getText().string();
logger.info(" suggest option : " + suggestText);
}
}
}
} catch (IOException e) {
logger.error(e);
}
}
public static void main(String[] args) {
termSuggest();
logger.info("--------------------------------------");
completionSuggester();
}
}
8. aggregation 聚合分析
AggregationDemo.java
package com.study.es_hrset_client;
import java.io.IOException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.Avg;
import org.elasticsearch.search.builder.SearchSourceBuilder;
/**
*
* @Description: 聚合分析
* @author lgs
* @date 2018年6月23日
*
*/
public class AggregationDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (RestHighLevelClient client = InitDemo.getClient();) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("bank");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.size(0);
//加入聚合
//字段值項分組聚合
TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_age")
.field("age").order(BucketOrder.aggregation("average_balance", true));
//計算每組的平均balance指標
aggregation.subAggregation(AggregationBuilders.avg("average_balance")
.field("balance"));
sourceBuilder.aggregation(aggregation);
searchRequest.source(sourceBuilder);
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest);
//4、處理響應
//搜索結果狀態信息
if(RestStatus.OK.equals(searchResponse.status())) {
// 獲取聚合結果
Aggregations aggregations = searchResponse.getAggregations();
Terms byAgeAggregation = aggregations.get("by_age");
logger.info("aggregation by_age 結果");
logger.info("docCountError: " + byAgeAggregation.getDocCountError());
logger.info("sumOfOtherDocCounts: " + byAgeAggregation.getSumOfOtherDocCounts());
logger.info("------------------------------------");
for(Bucket buck : byAgeAggregation.getBuckets()) {
logger.info("key: " + buck.getKeyAsNumber());
logger.info("docCount: " + buck.getDocCount());
logger.info("docCountError: " + buck.getDocCountError());
//取子聚合
Avg averageBalance = buck.getAggregations().get("average_balance");
logger.info("average_balance: " + averageBalance.getValue());
logger.info("------------------------------------");
}
//直接用key 來去分組
/*Bucket elasticBucket = byCompanyAggregation.getBucketByKey("24");
Avg averageAge = elasticBucket.getAggregations().get("average_age");
double avg = averageAge.getValue();*/
}
} catch (IOException e) {
logger.error(e);
}
}
}
9. 官網資料
各種查詢對應的QueryBuilder：
https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high-query-builders.html
各種聚合對應的AggregationBuilder：
https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high-aggregation-builders.html
四、Java Client
1. Java Client 說明
java client 使用 TransportClient，各種操作本質上都是異步的(可以用 listener，或返回 Future ）。
注意：ES的發展規划中在7.0版本開始將廢棄 TransportClient，8.0版本中將完全移除 TransportClient，取而代之的是High Level REST Client。
High Level REST Client 中的操作API和java client 大多是一樣的。
2. 官方學習鏈接
https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/index.html
3. 兼容性說明
請使用與服務端ES版本一致的客戶端版本
4. Java Client maven 集成
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>6.2.4</version>
</dependency>
5. Java Client logger 日志器說明
使用的是 log4j2 日志組件。
如果要使用其他的日志組件，可使用slf4j作橋
6. Init Client
Init Client setting 可用參數說明：
cluster.name
指定集群的名字，如果集群的名字不是默認的elasticsearch，需指定。
client.transport.sniff
設置為true,將自動嗅探整個集群，自動加入集群的節點到連接列表中。
client.transport.ignore_cluster_name
Set to true to ignore cluster name validation of connected nodes. (since 0.19.4)
client.transport.ping_timeout
The time to wait for a ping response from a node. Defaults to 5s.
client.transport.nodes_sampler_interval
How often to sample / ping the nodes listed and connected. Defaults to 5s.
五、Java Client使用示例
注意：TransPort客戶端的使用和RESTful風格的使用基本一致，除了獲取客戶端不一樣，還有發送請求有的不一樣外
准備：
編寫示例之前首先在maven工程里面引入和ES服務端版本一樣的Java客戶端
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>6.2.4</version>
</dependency>
給定集群的多個節點地址，將客戶端負載均衡地向這個節點地址集發請求：
InitDemo.java
package com.study.es_java_client;
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.TransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
public class InitDemo {
private static TransportClient client;
public static TransportClient getClient() throws UnknownHostException {
if(client == null) {
//client = new PreBuiltTransportClient(Settings.EMPTY)
// 連接集群的設置
Settings settings = Settings.builder()
//.put("cluster.name", "myClusterName") //如果集群的名字不是默認的elasticsearch，需指定
.put("client.transport.sniff", true) //自動嗅探
.build();
client = new PreBuiltTransportClient(settings)
//.addTransportAddress(new TransportAddress(InetAddress.getByName("localhost"), 9300));
.addTransportAddress(new TransportAddress(InetAddress.getByName("127.0.0.1"), 9300));
//可用連接設置參數說明
/*
cluster.name
指定集群的名字，如果集群的名字不是默認的elasticsearch，需指定。
client.transport.sniff
設置為true,將自動嗅探整個集群，自動加入集群的節點到連接列表中。
client.transport.ignore_cluster_name
Set to true to ignore cluster name validation of connected nodes. (since 0.19.4)
client.transport.ping_timeout
The time to wait for a ping response from a node. Defaults to 5s.
client.transport.nodes_sampler_interval
How often to sample / ping the nodes listed and connected. Defaults to 5s.
*/
}
return client;
}
}
注意使用ES的客戶端時類比之前我們在Kibana進行的ES的相關操作，這樣使用起來更加有效果
1. Create index 創建索引
CreateIndexDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
public class CreateIndexDemo {
public static void main(String[] args) {
//這里和RESTful風格不同
try (TransportClient client = InitDemo.getClient();) {
// 1、創建 創建索引request
CreateIndexRequest request = new CreateIndexRequest("mess");
// 2、設置索引的settings
request.settings(Settings.builder().put("index.number_of_shards", 3) // 分片數
.put("index.number_of_replicas", 2) // 副本數
.put("analysis.analyzer.default.tokenizer", "ik_smart") // 默認分詞器
);
// 3、設置索引的mappings
request.mapping("_doc",
" {\n" +
" \"_doc\": {\n" +
" \"properties\": {\n" +
" \"message\": {\n" +
" \"type\": \"text\"\n" +
" }\n" +
" }\n" +
" }\n" +
" }",
XContentType.JSON);
// 4、 設置索引的別名
request.alias(new Alias("mmm"));
// 5、 發送請求 這里和RESTful風格不同
CreateIndexResponse createIndexResponse = client.admin().indices()
.create(request).get();
// 6、處理響應
boolean acknowledged = createIndexResponse.isAcknowledged();
boolean shardsAcknowledged = createIndexResponse
.isShardsAcknowledged();
System.out.println("acknowledged = " + acknowledged);
System.out.println("shardsAcknowledged = " + shardsAcknowledged);
// listener方式發送請求
/*ActionListener<CreateIndexResponse> listener = new ActionListener<CreateIndexResponse>() {
@Override
public void onResponse(
CreateIndexResponse createIndexResponse) {
// 6、處理響應
boolean acknowledged = createIndexResponse.isAcknowledged();
boolean shardsAcknowledged = createIndexResponse
.isShardsAcknowledged();
System.out.println("acknowledged = " + acknowledged);
System.out.println(
"shardsAcknowledged = " + shardsAcknowledged);
}
@Override
public void onFailure(Exception e) {
System.out.println("創建索引異常：" + e.getMessage());
}
};
client.admin().indices().create(request, listener);
*/
} catch (IOException | InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
2. index document
索引文檔，即往索引里面放入文檔數據.類似於數據庫里面向表里面插入一行數據，一行數據就是一個文檔
IndexDocumentDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.RestStatus;
public class IndexDocumentDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
//這里和RESTful風格不同
try (TransportClient client = InitDemo.getClient();) {
// 1、創建索引請求
IndexRequest request = new IndexRequest(
"mess", //索引
"_doc", // mapping type
"11"); //文檔id
// 2、准備文檔數據
// 方式一：直接給JSON串
String jsonString = "{" +
"\"user\":\"kimchy\"," +
"\"postDate\":\"2013-01-30\"," +
"\"message\":\"trying out Elasticsearch\"" +
"}";
request.source(jsonString, XContentType.JSON);
// 方式二：以map對象來表示文檔
/*
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("user", "kimchy");
jsonMap.put("postDate", new Date());
jsonMap.put("message", "trying out Elasticsearch");
request.source(jsonMap);
*/
// 方式三：用XContentBuilder來構建文檔
/*
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
builder.field("user", "kimchy");
builder.field("postDate", new Date());
builder.field("message", "trying out Elasticsearch");
}
builder.endObject();
request.source(builder);
*/
// 方式四：直接用key-value對給出
/*
request.source("user", "kimchy",
"postDate", new Date(),
"message", "trying out Elasticsearch");
*/
//3、其他的一些可選設置
/*
request.routing("routing"); //設置routing值
request.timeout(TimeValue.timeValueSeconds(1)); //設置主分片等待時長
request.setRefreshPolicy("wait_for"); //設置重刷新策略
request.version(2); //設置版本號
request.opType(DocWriteRequest.OpType.CREATE); //操作類別
*/
//4、發送請求
IndexResponse indexResponse = null;
try {
//方式一： 用client.index 方法，返回是 ActionFuture<IndexResponse>，再調用get獲取響應結果
indexResponse = client.index(request).get();
//方式二：client提供了一個 prepareIndex方法，內部為我們創建IndexRequest
/*IndexResponse indexResponse = client.prepareIndex("mess","_doc","11")
.setSource(jsonString, XContentType.JSON)
.get();*/
//方式三：request + listener
//client.index(request, listener);
} catch(ElasticsearchException e) {
// 捕獲，並處理異常
//判斷是否版本沖突、create但文檔已存在沖突
if (e.status() == RestStatus.CONFLICT) {
logger.error("沖突了，請在此寫沖突處理邏輯！\n" + e.getDetailedMessage());
}
logger.error("索引異常", e);
}catch (InterruptedException | ExecutionException e) {
logger.error("索引異常", e);
}
//5、處理響應
if(indexResponse != null) {
String index = indexResponse.getIndex();
String type = indexResponse.getType();
String id = indexResponse.getId();
long version = indexResponse.getVersion();
if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
System.out.println("新增文檔成功，處理邏輯代碼寫到這里。");
} else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
System.out.println("修改文檔成功，處理邏輯代碼寫到這里。");
}
// 分片處理信息
ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
}
// 如果有分片副本失敗，可以獲得失敗原因信息
if (shardInfo.getFailed() > 0) {
for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
String reason = failure.reason();
System.out.println("副本失敗原因：" + reason);
}
}
}
//listener 方式
/*ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
@Override
public void onResponse(IndexResponse indexResponse) {
}
@Override
public void onFailure(Exception e) {
}
};
client.index(request, listener);
*/
} catch (IOException e) {
e.printStackTrace();
}
}
}
3. get document
獲取文檔數據
GetDocumentDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
public class GetDocumentDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
//這里和RESTful風格不同
try (TransportClient client = InitDemo.getClient();) {
// 1、創建獲取文檔請求
GetRequest request = new GetRequest(
"mess", //索引
"_doc", // mapping type
"11"); //文檔id
// 2、可選的設置
//request.routing("routing");
//request.version(2);
//request.fetchSourceContext(new FetchSourceContext(false)); //是否獲取_source字段
//選擇返回的字段
String[] includes = new String[]{"message", "*Date"};
String[] excludes = Strings.EMPTY_ARRAY;
FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
request.fetchSourceContext(fetchSourceContext);
//也可寫成這樣
/*String[] includes = Strings.EMPTY_ARRAY;
String[] excludes = new String[]{"message"};
FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
request.fetchSourceContext(fetchSourceContext);*/
// 取stored字段
/*request.storedFields("message");
GetResponse getResponse = client.get(request);
String message = getResponse.getField("message").getValue();*/
//3、發送請求
GetResponse getResponse = null;
try {
getResponse = client.get(request).get();
} catch (ElasticsearchException e) {
if (e.status() == RestStatus.NOT_FOUND) {
logger.error("沒有找到該id的文檔" );
}
if (e.status() == RestStatus.CONFLICT) {
logger.error("獲取時版本沖突了，請在此寫沖突處理邏輯！" );
}
logger.error("獲取文檔異常", e);
}catch (InterruptedException | ExecutionException e) {
logger.error("索引異常", e);
}
//4、處理響應
if(getResponse != null) {
String index = getResponse.getIndex();
String type = getResponse.getType();
String id = getResponse.getId();
if (getResponse.isExists()) { // 文檔存在
long version = getResponse.getVersion();
String sourceAsString = getResponse.getSourceAsString(); //結果取成 String
Map<String, Object> sourceAsMap = getResponse.getSourceAsMap(); // 結果取成Map
byte[] sourceAsBytes = getResponse.getSourceAsBytes(); //結果取成字節數組
logger.info("index:" + index + " type:" + type + " id:" + id);
logger.info(sourceAsString);
} else {
logger.error("沒有找到該id的文檔" );
}
}
//異步方式發送獲取文檔請求
/*
ActionListener<GetResponse> listener = new ActionListener<GetResponse>() {
@Override
public void onResponse(GetResponse getResponse) {
}
@Override
public void onFailure(Exception e) {
}
};
client.getAsync(request, listener);
*/
} catch (IOException e) {
e.printStackTrace();
}
}
}
4. Bulk
批量索引文檔，即批量往索引里面放入文檔數據.類似於數據庫里面批量向表里面插入多行數據，一行數據就是一個文檔
BulkDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
public class BulkDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
//這里和RESTful風格不同
try (TransportClient client = InitDemo.getClient();) {
// 1、創建批量操作請求
BulkRequest request = new BulkRequest();
request.add(new IndexRequest("mess", "_doc", "1")
.source(XContentType.JSON,"field", "foo"));
request.add(new IndexRequest("mess", "_doc", "2")
.source(XContentType.JSON,"field", "bar"));
request.add(new IndexRequest("mess", "_doc", "3")
.source(XContentType.JSON,"field", "baz"));
/*
request.add(new DeleteRequest("mess", "_doc", "3"));
request.add(new UpdateRequest("mess", "_doc", "2")
.doc(XContentType.JSON,"other", "test"));
request.add(new IndexRequest("mess", "_doc", "4")
.source(XContentType.JSON,"field", "baz"));
*/
// 2、可選的設置
/*
request.timeout("2m");
request.setRefreshPolicy("wait_for");
request.waitForActiveShards(2);
*/
//3、發送請求
// 同步請求
BulkResponse bulkResponse = client.bulk(request).get();
//4、處理響應
if(bulkResponse != null) {
for (BulkItemResponse bulkItemResponse : bulkResponse) {
DocWriteResponse itemResponse = bulkItemResponse.getResponse();
if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
|| bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
IndexResponse indexResponse = (IndexResponse) itemResponse;
//TODO 新增成功的處理
} else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
UpdateResponse updateResponse = (UpdateResponse) itemResponse;
//TODO 修改成功的處理
} else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
//TODO 刪除成功的處理
}
}
}
//異步方式發送批量操作請求
/*
ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
@Override
public void onResponse(BulkResponse bulkResponse) {
}
@Override
public void onFailure(Exception e) {
}
};
client.bulkAsync(request, listener);
*/
} catch (IOException | InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
5. search
搜索數據
SearchDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
public class SearchDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (TransportClient client = InitDemo.getClient();) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("bank");
searchRequest.types("_doc");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//構造QueryBuilder
/*QueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("user", "kimchy")
.fuzziness(Fuzziness.AUTO)
.prefixLength(3)
.maxExpansions(10);
sourceBuilder.query(matchQueryBuilder);*/
sourceBuilder.query(QueryBuilders.termQuery("age", 24));
sourceBuilder.from(0);
sourceBuilder.size(10);
sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
//是否返回_source字段
//sourceBuilder.fetchSource(false);
//設置返回哪些字段
/*String[] includeFields = new String[] {"title", "user", "innerObject.*"};
String[] excludeFields = new String[] {"_type"};
sourceBuilder.fetchSource(includeFields, excludeFields);*/
//指定排序
//sourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));
//sourceBuilder.sort(new FieldSortBuilder("_uid").order(SortOrder.ASC));
// 設置返回 profile
//sourceBuilder.profile(true);
//將請求體加入到請求中
searchRequest.source(sourceBuilder);
// 可選的設置
//searchRequest.routing("routing");
// 高亮設置
/*
HighlightBuilder highlightBuilder = new HighlightBuilder();
HighlightBuilder.Field highlightTitle =
new HighlightBuilder.Field("title");
highlightTitle.highlighterType("unified");
highlightBuilder.field(highlightTitle);
HighlightBuilder.Field highlightUser = new HighlightBuilder.Field("user");
highlightBuilder.field(highlightUser);
sourceBuilder.highlighter(highlightBuilder);*/
//加入聚合
/*TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_company")
.field("company.keyword");
aggregation.subAggregation(AggregationBuilders.avg("average_age")
.field("age"));
sourceBuilder.aggregation(aggregation);*/
//做查詢建議
/*SuggestionBuilder termSuggestionBuilder =
SuggestBuilders.termSuggestion("user").text("kmichy");
SuggestBuilder suggestBuilder = new SuggestBuilder();
suggestBuilder.addSuggestion("suggest_user", termSuggestionBuilder);
sourceBuilder.suggest(suggestBuilder);*/
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest).get();
//4、處理響應
//搜索結果狀態信息
RestStatus status = searchResponse.status();
TimeValue took = searchResponse.getTook();
Boolean terminatedEarly = searchResponse.isTerminatedEarly();
boolean timedOut = searchResponse.isTimedOut();
//分片搜索情況
int totalShards = searchResponse.getTotalShards();
int successfulShards = searchResponse.getSuccessfulShards();
int failedShards = searchResponse.getFailedShards();
for (ShardSearchFailure failure : searchResponse.getShardFailures()) {
// failures should be handled here
}
//處理搜索命中文檔結果
SearchHits hits = searchResponse.getHits();
long totalHits = hits.getTotalHits();
float maxScore = hits.getMaxScore();
SearchHit[] searchHits = hits.getHits();
for (SearchHit hit : searchHits) {
// do something with the SearchHit
String index = hit.getIndex();
String type = hit.getType();
String id = hit.getId();
float score = hit.getScore();
//取_source字段值
String sourceAsString = hit.getSourceAsString(); //取成json串
Map<String, Object> sourceAsMap = hit.getSourceAsMap(); // 取成map對象
//從map中取字段值
/*
String documentTitle = (String) sourceAsMap.get("title");
List<Object> users = (List<Object>) sourceAsMap.get("user");
Map<String, Object> innerObject = (Map<String, Object>) sourceAsMap.get("innerObject");
*/
logger.info("index:" + index + " type:" + type + " id:" + id);
logger.info(sourceAsString);
//取高亮結果
/*Map<String, HighlightField> highlightFields = hit.getHighlightFields();
HighlightField highlight = highlightFields.get("title");
Text[] fragments = highlight.fragments();
String fragmentString = fragments[0].string();*/
}
// 獲取聚合結果
/*
Aggregations aggregations = searchResponse.getAggregations();
Terms byCompanyAggregation = aggregations.get("by_company");
Bucket elasticBucket = byCompanyAggregation.getBucketByKey("Elastic");
Avg averageAge = elasticBucket.getAggregations().get("average_age");
double avg = averageAge.getValue();
*/
// 獲取建議結果
/*Suggest suggest = searchResponse.getSuggest();
TermSuggestion termSuggestion = suggest.getSuggestion("suggest_user");
for (TermSuggestion.Entry entry : termSuggestion.getEntries()) {
for (TermSuggestion.Entry.Option option : entry) {
String suggestText = option.getText().string();
}
}
*/
//異步方式發送獲查詢請求
/*
ActionListener<SearchResponse> listener = new ActionListener<SearchResponse>() {
@Override
public void onResponse(SearchResponse getResponse) {
//結果獲取
}
@Override
public void onFailure(Exception e) {
//失敗處理
}
};
client.searchAsync(searchRequest, listener);
*/
} catch (IOException | InterruptedException | ExecutionException e) {
logger.error(e);
}
}
}
6. highlight 高亮
HighlightDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
public class HighlightDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (TransportClient client = InitDemo.getClient();) {
// 1、創建search請求
SearchRequest searchRequest = new SearchRequest("hl_test");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//構造QueryBuilder
QueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("title", "lucene solr");
sourceBuilder.query(matchQueryBuilder);
//分頁設置
/*sourceBuilder.from(0);
sourceBuilder.size(5); ;*/
// 高亮設置
HighlightBuilder highlightBuilder = new HighlightBuilder();
highlightBuilder.requireFieldMatch(false).field("title").field("content")
.preTags("<strong>").postTags("</strong>");
//不同字段可有不同設置，如不同標簽
/*HighlightBuilder.Field highlightTitle = new HighlightBuilder.Field("title");
highlightTitle.preTags("<strong>").postTags("</strong>");
highlightBuilder.field(highlightTitle);
HighlightBuilder.Field highlightContent = new HighlightBuilder.Field("content");
highlightContent.preTags("<b>").postTags("</b>");
highlightBuilder.field(highlightContent).requireFieldMatch(false);*/
sourceBuilder.highlighter(highlightBuilder);
searchRequest.source(sourceBuilder);
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest).get();
//4、處理響應
if(RestStatus.OK.equals(searchResponse.status())) {
//處理搜索命中文檔結果
SearchHits hits = searchResponse.getHits();
long totalHits = hits.getTotalHits();
SearchHit[] searchHits = hits.getHits();
for (SearchHit hit : searchHits) {
String index = hit.getIndex();
String type = hit.getType();
String id = hit.getId();
float score = hit.getScore();
//取_source字段值
//String sourceAsString = hit.getSourceAsString(); //取成json串
Map<String, Object> sourceAsMap = hit.getSourceAsMap(); // 取成map對象
//從map中取字段值
/*String title = (String) sourceAsMap.get("title");
String content = (String) sourceAsMap.get("content"); */
logger.info("index:" + index + " type:" + type + " id:" + id);
logger.info("sourceMap : " + sourceAsMap);
//取高亮結果
Map<String, HighlightField> highlightFields = hit.getHighlightFields();
HighlightField highlight = highlightFields.get("title");
if(highlight != null) {
Text[] fragments = highlight.fragments(); //多值的字段會有多個值
if(fragments != null) {
String fragmentString = fragments[0].string();
logger.info("title highlight : " + fragmentString);
//可用高亮字符串替換上面sourceAsMap中的對應字段返回到上一級調用
//sourceAsMap.put("title", fragmentString);
}
}
highlight = highlightFields.get("content");
if(highlight != null) {
Text[] fragments = highlight.fragments(); //多值的字段會有多個值
if(fragments != null) {
String fragmentString = fragments[0].string();
logger.info("content highlight : " + fragmentString);
//可用高亮字符串替換上面sourceAsMap中的對應字段返回到上一級調用
//sourceAsMap.put("content", fragmentString);
}
}
}
}
} catch (IOException | InterruptedException | ExecutionException e) {
logger.error(e);
}
}
}
7. suggest 查詢建議
SuggestDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.SuggestionBuilder;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.term.TermSuggestion;
public class SuggestDemo {
private static Logger logger = LogManager.getRootLogger();
//拼寫檢查
public static void termSuggest(TransportClient client) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("mess");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.size(0);
//做查詢建議
//詞項建議
SuggestionBuilder termSuggestionBuilder =
SuggestBuilders.termSuggestion("user").text("kmichy");
SuggestBuilder suggestBuilder = new SuggestBuilder();
suggestBuilder.addSuggestion("suggest_user", termSuggestionBuilder);
sourceBuilder.suggest(suggestBuilder);
searchRequest.source(sourceBuilder);
try{
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest).get();
//4、處理響應
//搜索結果狀態信息
if(RestStatus.OK.equals(searchResponse.status())) {
// 獲取建議結果
Suggest suggest = searchResponse.getSuggest();
TermSuggestion termSuggestion = suggest.getSuggestion("suggest_user");
for (TermSuggestion.Entry entry : termSuggestion.getEntries()) {
logger.info("text: " + entry.getText().string());
for (TermSuggestion.Entry.Option option : entry) {
String suggestText = option.getText().string();
logger.info(" suggest option : " + suggestText);
}
}
}
} catch (InterruptedException | ExecutionException e) {
logger.error(e);
}
/*
"suggest": {
"my-suggestion": [
{
"text": "tring",
"offset": 0,
"length": 5,
"options": [
{
"text": "trying",
"score": 0.8,
"freq": 1
}
]
},
{
"text": "out",
"offset": 6,
"length": 3,
"options": []
},
{
"text": "elasticsearch",
"offset": 10,
"length": 13,
"options": []
}
]
}*/
}
//自動補全
public static void completionSuggester(TransportClient client) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("music");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.size(0);
//做查詢建議
//自動補全
/*POST music/_search?pretty
{
"suggest": {
"song-suggest" : {
"prefix" : "lucene s",
"completion" : {
"field" : "suggest" ,
"skip_duplicates": true
}
}
}
}*/
SuggestionBuilder termSuggestionBuilder =
SuggestBuilders.completionSuggestion("suggest").prefix("lucene s")
.skipDuplicates(true);
SuggestBuilder suggestBuilder = new SuggestBuilder();
suggestBuilder.addSuggestion("song-suggest", termSuggestionBuilder);
sourceBuilder.suggest(suggestBuilder);
searchRequest.source(sourceBuilder);
try {
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest).get();
//4、處理響應
//搜索結果狀態信息
if(RestStatus.OK.equals(searchResponse.status())) {
// 獲取建議結果
Suggest suggest = searchResponse.getSuggest();
CompletionSuggestion termSuggestion = suggest.getSuggestion("song-suggest");
for (CompletionSuggestion.Entry entry : termSuggestion.getEntries()) {
logger.info("text: " + entry.getText().string());
for (CompletionSuggestion.Entry.Option option : entry) {
String suggestText = option.getText().string();
logger.info(" suggest option : " + suggestText);
}
}
}
} catch (InterruptedException | ExecutionException e) {
logger.error(e);
}
}
public static void main(String[] args) {
try (TransportClient client = InitDemo.getClient();) {
termSuggest(client);
logger.info("--------------------------------------");
completionSuggester(client);
} catch (IOException e) {
logger.error(e);
}
}
}
8. aggregation 聚合分析
AggregationDemo.java
package com.study.es_java_client;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.Avg;
import org.elasticsearch.search.builder.SearchSourceBuilder;
public class AggregationDemo {
private static Logger logger = LogManager.getRootLogger();
public static void main(String[] args) {
try (TransportClient client = InitDemo.getClient();) {
// 1、創建search請求
//SearchRequest searchRequest = new SearchRequest();
SearchRequest searchRequest = new SearchRequest("bank");
// 2、用SearchSourceBuilder來構造查詢請求體 ,請仔細查看它的方法，構造各種查詢的方法都在這。
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.size(0);
//加入聚合
//字段值項分組聚合
TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_age")
.field("age").order(BucketOrder.aggregation("average_balance", true));
//計算每組的平均balance指標
aggregation.subAggregation(AggregationBuilders.avg("average_balance")
.field("balance"));
sourceBuilder.aggregation(aggregation);
searchRequest.source(sourceBuilder);
//3、發送請求
SearchResponse searchResponse = client.search(searchRequest).get();
//4、處理響應
//搜索結果狀態信息
if(RestStatus.OK.equals(searchResponse.status())) {
// 獲取聚合結果
Aggregations aggregations = searchResponse.getAggregations();
Terms byAgeAggregation = aggregations.get("by_age");
logger.info("aggregation by_age 結果");
logger.info("docCountError: " + byAgeAggregation.getDocCountError());
logger.info("sumOfOtherDocCounts: " + byAgeAggregation.getSumOfOtherDocCounts());
logger.info("------------------------------------");
for(Bucket buck : byAgeAggregation.getBuckets()) {
logger.info("key: " + buck.getKeyAsNumber());
logger.info("docCount: " + buck.getDocCount());
//logger.info("docCountError: " + buck.getDocCountError());
//取子聚合
Avg averageBalance = buck.getAggregations().get("average_balance");
logger.info("average_balance: " + averageBalance.getValue());
logger.info("------------------------------------");
}
//直接用key 來去分組
/*Bucket elasticBucket = byCompanyAggregation.getBucketByKey("24");
Avg averageAge = elasticBucket.getAggregations().get("average_age");
double avg = averageAge.getValue();*/
}
} catch (IOException | InterruptedException | ExecutionException e) {
logger.error(e);
}
}
}
9. 官網文檔
Document API 文檔操作API：
https://www.elastic.co/guide/en/elasticsearch/client/java-api/6.2/java-docs.html
Search API：
https://www.elastic.co/guide/en/elasticsearch/client/java-api/6.2/java-search.html
六、Spring Data Elasticsearch
ES與Spring集成使用，可以作為了解，個人建議還是使用原生的ES的java客戶端
官網鏈接：
https://docs.spring.io/spring-data/elasticsearch/docs/current/reference/html/
代碼庫：
https://github.com/spring-projects/spring-data-elasticsearch
七、源代碼獲取地址
https://github.com/leeSmall/Elasticsearch-Java-client-api
轉自：https://www.cnblogs.com/leeSmall/p/9218779.html
注:想學習es相關知識，請到https://www.cnblogs.com/leeSmall/p/9218779.html連接下查找詳細信息，該博客寫的很好