使用java操作elasticsearch的具體方法
更新時間：2018年01月03日 09:31:33 作者：bronk
本篇文章主要介紹了使用java操作elasticsearch的具體方法，小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
系統(tǒng)環(huán)境: vm12 下的centos 7.2
當前安裝版本: elasticsearch-2.4.0.tar.gz
Java操作es集群步驟1：配置集群對象信息；2：創(chuàng)建客戶端；3：查看集群信息
1：集群名稱
默認集群名為elasticsearch，如果集群名稱和指定的不一致則在使用節(jié)點資源時會報錯。
2：嗅探功能
通過client.transport.sniff啟動嗅探功能，這樣只需要指定集群中的某一個節(jié)點(不一定是主節(jié)點)，然后會加載集群中的其他節(jié)點，這樣只要程序不停即使此節(jié)點宕機仍然可以連接到其他節(jié)點。
3：查詢類型SearchType.QUERY_THEN_FETCH
es 查詢共有4種查詢類型
QUERY_AND_FETCH:
主節(jié)點將查詢請求分發(fā)到所有的分片中，各個分片按照自己的查詢規(guī)則即詞頻文檔頻率進行打分排序，然后將結果返回給主節(jié)點，主節(jié)點對所有數據進行匯總排序然后再返回給客戶端，此種方式只需要和es交互一次。
這種查詢方式存在數據量和排序問題，主節(jié)點會匯總所有分片返回的數據這樣數據量會比較大，二是各個分片上的規(guī)則可能不一致。
QUERY_THEN_FETCH:
主節(jié)點將請求分發(fā)給所有分片，各個分片打分排序后將數據的id和分值返回給主節(jié)點，主節(jié)點收到后進行匯總排序再根據排序后的id到對應的節(jié)點讀取對應的數據再返回給客戶端，此種方式需要和es交互兩次。
這種方式解決了數據量問題但是排序問題依然存在而且是es的默認查詢方式
DEF_QUERY_AND_FETCH: 和 DFS_QUERY_THEN_FETCH:
將各個分片的規(guī)則統(tǒng)一起來進行打分。解決了排序問題但是DFS_QUERY_AND_FETCH仍然存在數據量問題，DFS_QUERY_THEN_FETCH兩種噢乖你問題都解決但是效率是最差的。
1, 獲取client, 兩種方式獲取
@Before
public void before() throws Exception {
Map<String, String> map = new HashMap<String, String>();
map.put("cluster.name", "elasticsearch_wenbronk");
Settings.Builder settings = Settings.builder().put(map);
client = TransportClient.builder().settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("www.wenbronk.com"), Integer.parseInt("9300")));
}
@Before
public void before11() throws Exception {
// 創(chuàng)建客戶端, 使用的默認集群名, "elasticSearch"
// client = TransportClient.builder().build()
// .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("www.wenbronk.com"), 9300));
// 通過setting對象指定集群配置信息, 配置的集群名
Settings settings = Settings.settingsBuilder().put("cluster.name", "elasticsearch_wenbronk") // 設置集群名
// .put("client.transport.sniff", true) // 開啟嗅探 , 開啟后會一直連接不上, 原因未知
// .put("network.host", "192.168.50.37")
.put("client.transport.ignore_cluster_name", true) // 忽略集群名字驗證, 打開后集群名字不對也能連接上
// .put("client.transport.nodes_sampler_interval", 5) //報錯,
// .put("client.transport.ping_timeout", 5) // 報錯, ping等待時間,
.build();
client = TransportClient.builder().settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress("192.168.50.37", 9300)));
// 默認5s
// 多久打開連接, 默認5s
System.out.println("success connect");
}
PS: 官網給的2種方式都不能用, 需要合起來才能用, 浪費老子一下午...
其他參數的意義:
代碼:
package com.wenbronk.javaes;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkProcessor.Listener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
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.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.script.Script;
import org.junit.Before;
import org.junit.Test;
import com.alibaba.fastjson.JSONObject;
/**
* 使用java API操作elasticSearch
*
* @author 231
*
*/
public class JavaESTest {
private TransportClient client;
private IndexRequest source;
/**
* 獲取連接, 第一種方式
* @throws Exception
*/
// @Before
public void before() throws Exception {
Map<String, String> map = new HashMap<String, String>();
map.put("cluster.name", "elasticsearch_wenbronk");
Settings.Builder settings = Settings.builder().put(map);
client = TransportClient.builder().settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("www.wenbronk.com"), Integer.parseInt("9300")));
}
/**
* 查看集群信息
*/
@Test
public void testInfo() {
List<DiscoveryNode> nodes = client.connectedNodes();
for (DiscoveryNode node : nodes) {
System.out.println(node.getHostAddress());
}
}
/**
* 組織json串, 方式1,直接拼接
*/
public String createJson1() {
String json = "{" +
"\"user\":\"kimchy\"," +
"\"postDate\":\"2013-01-30\"," +
"\"message\":\"trying out Elasticsearch\"" +
"}";
return json;
}
/**
* 使用map創(chuàng)建json
*/
public Map<String, Object> createJson2() {
Map<String,Object> json = new HashMap<String, Object>();
json.put("user", "kimchy");
json.put("postDate", new Date());
json.put("message", "trying out elasticsearch");
return json;
}
/**
* 使用fastjson創(chuàng)建
*/
public JSONObject createJson3() {
JSONObject json = new JSONObject();
json.put("user", "kimchy");
json.put("postDate", new Date());
json.put("message", "trying out elasticsearch");
return json;
}
/**
* 使用es的幫助類
*/
public XContentBuilder createJson4() throws Exception {
// 創(chuàng)建json對象, 其中一個創(chuàng)建json的方式
XContentBuilder source = XContentFactory.jsonBuilder()
.startObject()
.field("user", "kimchy")
.field("postDate", new Date())
.field("message", "trying to out ElasticSearch")
.endObject();
return source;
}
/**
* 存入索引中
* @throws Exception
*/
@Test
public void test1() throws Exception {
XContentBuilder source = createJson4();
// 存json入索引中
IndexResponse response = client.prepareIndex("twitter", "tweet", "1").setSource(source).get();
// // 結果獲取
String index = response.getIndex();
String type = response.getType();
String id = response.getId();
long version = response.getVersion();
boolean created = response.isCreated();
System.out.println(index + " : " + type + ": " + id + ": " + version + ": " + created);
}
/**
* get API 獲取指定文檔信息
*/
@Test
public void testGet() {
// GetResponse response = client.prepareGet("twitter", "tweet", "1")
// .get();
GetResponse response = client.prepareGet("twitter", "tweet", "1")
.setOperationThreaded(false) // 線程安全
.get();
System.out.println(response.getSourceAsString());
}
/**
* 測試 delete api
*/
@Test
public void testDelete() {
DeleteResponse response = client.prepareDelete("twitter", "tweet", "1")
.get();
String index = response.getIndex();
String type = response.getType();
String id = response.getId();
long version = response.getVersion();
System.out.println(index + " : " + type + ": " + id + ": " + version);
}
/**
* 測試更新 update API
* 使用 updateRequest 對象
* @throws Exception
*/
@Test
public void testUpdate() throws Exception {
UpdateRequest updateRequest = new UpdateRequest();
updateRequest.index("twitter");
updateRequest.type("tweet");
updateRequest.id("1");
updateRequest.doc(XContentFactory.jsonBuilder()
.startObject()
// 對沒有的字段添加, 對已有的字段替換
.field("gender", "male")
.field("message", "hello")
.endObject());
UpdateResponse response = client.update(updateRequest).get();
// 打印
String index = response.getIndex();
String type = response.getType();
String id = response.getId();
long version = response.getVersion();
System.out.println(index + " : " + type + ": " + id + ": " + version);
}
/**
* 測試update api, 使用client
* @throws Exception
*/
@Test
public void testUpdate2() throws Exception {
// 使用Script對象進行更新
// UpdateResponse response = client.prepareUpdate("twitter", "tweet", "1")
// .setScript(new Script("hits._source.gender = \"male\""))
// .get();
// 使用XContFactory.jsonBuilder() 進行更新
// UpdateResponse response = client.prepareUpdate("twitter", "tweet", "1")
// .setDoc(XContentFactory.jsonBuilder()
// .startObject()
// .field("gender", "malelelele")
// .endObject()).get();
// 使用updateRequest對象及script
// UpdateRequest updateRequest = new UpdateRequest("twitter", "tweet", "1")
// .script(new Script("ctx._source.gender=\"male\""));
// UpdateResponse response = client.update(updateRequest).get();
// 使用updateRequest對象及documents進行更新
UpdateResponse response = client.update(new UpdateRequest("twitter", "tweet", "1")
.doc(XContentFactory.jsonBuilder()
.startObject()
.field("gender", "male")
.endObject()
)).get();
System.out.println(response.getIndex());
}
/**
* 測試update
* 使用updateRequest
* @throws Exception
* @throws InterruptedException
*/
@Test
public void testUpdate3() throws InterruptedException, Exception {
UpdateRequest updateRequest = new UpdateRequest("twitter", "tweet", "1")
.script(new Script("ctx._source.gender=\"male\""));
UpdateResponse response = client.update(updateRequest).get();
}
/**
* 測試upsert方法
* @throws Exception
*
*/
@Test
public void testUpsert() throws Exception {
// 設置查詢條件, 查找不到則添加生效
IndexRequest indexRequest = new IndexRequest("twitter", "tweet", "2")
.source(XContentFactory.jsonBuilder()
.startObject()
.field("name", "214")
.field("gender", "gfrerq")
.endObject());
// 設置更新, 查找到更新下面的設置
UpdateRequest upsert = new UpdateRequest("twitter", "tweet", "2")
.doc(XContentFactory.jsonBuilder()
.startObject()
.field("user", "wenbronk")
.endObject())
.upsert(indexRequest);
client.update(upsert).get();
}
/**
* 測試multi get api
* 從不同的index, type, 和id中獲取
*/
@Test
public void testMultiGet() {
MultiGetResponse multiGetResponse = client.prepareMultiGet()
.add("twitter", "tweet", "1")
.add("twitter", "tweet", "2", "3", "4")
.add("anothoer", "type", "foo")
.get();
for (MultiGetItemResponse itemResponse : multiGetResponse) {
GetResponse response = itemResponse.getResponse();
if (response.isExists()) {
String sourceAsString = response.getSourceAsString();
System.out.println(sourceAsString);
}
}
}
/**
* bulk 批量執(zhí)行
* 一次查詢可以update 或 delete多個document
*/
@Test
public void testBulk() throws Exception {
BulkRequestBuilder bulkRequest = client.prepareBulk();
bulkRequest.add(client.prepareIndex("twitter", "tweet", "1")
.setSource(XContentFactory.jsonBuilder()
.startObject()
.field("user", "kimchy")
.field("postDate", new Date())
.field("message", "trying out Elasticsearch")
.endObject()));
bulkRequest.add(client.prepareIndex("twitter", "tweet", "2")
.setSource(XContentFactory.jsonBuilder()
.startObject()
.field("user", "kimchy")
.field("postDate", new Date())
.field("message", "another post")
.endObject()));
BulkResponse response = bulkRequest.get();
System.out.println(response.getHeaders());
}
/**
* 使用bulk processor
* @throws Exception
*/
@Test
public void testBulkProcessor() throws Exception {
// 創(chuàng)建BulkPorcessor對象
BulkProcessor bulkProcessor = BulkProcessor.builder(client, new Listener() {
public void beforeBulk(long paramLong, BulkRequest paramBulkRequest) {
// TODO Auto-generated method stub
}
// 執(zhí)行出錯時執(zhí)行
public void afterBulk(long paramLong, BulkRequest paramBulkRequest, Throwable paramThrowable) {
// TODO Auto-generated method stub
}
public void afterBulk(long paramLong, BulkRequest paramBulkRequest, BulkResponse paramBulkResponse) {
// TODO Auto-generated method stub
}
})
// 1w次請求執(zhí)行一次bulk
.setBulkActions(10000)
// 1gb的數據刷新一次bulk
.setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
// 固定5s必須刷新一次
.setFlushInterval(TimeValue.timeValueSeconds(5))
// 并發(fā)請求數量, 0不并發(fā), 1并發(fā)允許執(zhí)行
.setConcurrentRequests(1)
// 設置退避, 100ms后執(zhí)行, 最大請求3次
.setBackoffPolicy(
BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
.build();
// 添加單次請求
bulkProcessor.add(new IndexRequest("twitter", "tweet", "1"));
bulkProcessor.add(new DeleteRequest("twitter", "tweet", "2"));
// 關閉
bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
// 或者
bulkProcessor.close();
}
}
tes2代碼:
package com.wenbronk.javaes;
import java.net.InetSocketAddress;
import org.apache.lucene.queryparser.xml.FilterBuilderFactory;
import org.elasticsearch.action.search.MultiSearchResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.Settings.Builder;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.sort.SortParseElement;
import org.junit.Before;
import org.junit.Test;
/**
* 使用java API操作elasticSearch
* search API
* @author 231
*
*/
public class JavaESTest2 {
private TransportClient client;
/**
* 獲取client對象
*/
@Before
public void testBefore() {
Builder builder = Settings.settingsBuilder();
builder.put("cluster.name", "wenbronk_escluster");
// .put("client.transport.ignore_cluster_name", true);
Settings settings = builder.build();
org.elasticsearch.client.transport.TransportClient.Builder transportBuild = TransportClient.builder();
TransportClient client1 = transportBuild.settings(settings).build();
client = client1.addTransportAddress((new InetSocketTransportAddress(new InetSocketAddress("192.168.50.37", 9300))));
System.out.println("success connect to escluster");
}
/**
* 測試查詢
*/
@Test
public void testSearch() {
// SearchRequestBuilder searchRequestBuilder = client.prepareSearch("twitter", "tweet", "1");
// SearchResponse response = searchRequestBuilder.setTypes("type1", "type2")
// .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
// .setQuery(QueryBuilders.termQuery("user", "test"))
// .setPostFilter(QueryBuilders.rangeQuery("age").from(0).to(1))
// .setFrom(0).setSize(2).setExplain(true)
// .execute().actionGet();
SearchResponse response = client.prepareSearch()
.execute().actionGet();
// SearchHits hits = response.getHits();
// for (SearchHit searchHit : hits) {
// for(Iterator<SearchHitField> iterator = searchHit.iterator(); iterator.hasNext(); ) {
// SearchHitField next = iterator.next();
// System.out.println(next.getValues());
// }
// }
System.out.println(response);
}
/**
* 測試scroll api
* 對大量數據的處理更有效
*/
@Test
public void testScrolls() {
QueryBuilder queryBuilder = QueryBuilders.termQuery("twitter", "tweet");
SearchResponse response = client.prepareSearch("twitter")
.addSort(SortParseElement.DOC_FIELD_NAME, SortOrder.ASC)
.setScroll(new TimeValue(60000))
.setQuery(queryBuilder)
.setSize(100).execute().actionGet();
while(true) {
for (SearchHit hit : response.getHits().getHits()) {
System.out.println("i am coming");
}
SearchResponse response2 = client.prepareSearchScroll(response.getScrollId())
.setScroll(new TimeValue(60000)).execute().actionGet();
if (response2.getHits().getHits().length == 0) {
System.out.println("oh no=====");
break;
}
}
}
/**
* 測試multiSearch
*/
@Test
public void testMultiSearch() {
QueryBuilder qb1 = QueryBuilders.queryStringQuery("elasticsearch");
SearchRequestBuilder requestBuilder1 = client.prepareSearch().setQuery(qb1).setSize(1);
QueryBuilder qb2 = QueryBuilders.matchQuery("user", "kimchy");
SearchRequestBuilder requestBuilder2 = client.prepareSearch().setQuery(qb2).setSize(1);
MultiSearchResponse multiResponse = client.prepareMultiSearch().add(requestBuilder1).add(requestBuilder2)
.execute().actionGet();
long nbHits = 0;
for (MultiSearchResponse.Item item : multiResponse.getResponses()) {
SearchResponse response = item.getResponse();
nbHits = response.getHits().getTotalHits();
SearchHit[] hits = response.getHits().getHits();
System.out.println(nbHits);
}
}
/**
* 測試聚合查詢
*/
@Test
public void testAggregation() {
SearchResponse response = client.prepareSearch()
.setQuery(QueryBuilders.matchAllQuery()) // 先使用query過濾掉一部分
.addAggregation(AggregationBuilders.terms("term").field("user"))
.addAggregation(AggregationBuilders.dateHistogram("agg2").field("birth")
.interval(DateHistogramInterval.YEAR))
.execute().actionGet();
Aggregation aggregation2 = response.getAggregations().get("term");
Aggregation aggregation = response.getAggregations().get("agg2");
// SearchResponse response2 = client.search(new SearchRequest().searchType(SearchType.QUERY_AND_FETCH)).actionGet();
}
/**
* 測試terminate
*/
@Test
public void testTerminateAfter() {
SearchResponse response = client.prepareSearch("twitter").setTerminateAfter(1000).get();
if (response.isTerminatedEarly()) {
System.out.println("ternimate");
}
}
/**
* 過濾查詢: 大于gt, 小于lt, 小于等于lte, 大于等于gte
*/
@Test
public void testFilter() {
SearchResponse response = client.prepareSearch("twitter")
.setTypes("")
.setQuery(QueryBuilders.matchAllQuery()) //查詢所有
.setSearchType(SearchType.QUERY_THEN_FETCH)
// .setPostFilter(FilterBuilders.rangeFilter("age").from(0).to(19)
// .includeLower(true).includeUpper(true))
// .setPostFilter(FilterBuilderFactory .rangeFilter("age").gte(18).lte(22))
.setExplain(true) //explain為true表示根據數據相關度排序，和關鍵字匹配最高的排在前面
.get();
}
/**
* 分組查詢
*/
@Test
public void testGroupBy() {
client.prepareSearch("twitter").setTypes("tweet")
.setQuery(QueryBuilders.matchAllQuery())
.setSearchType(SearchType.QUERY_THEN_FETCH)
.addAggregation(AggregationBuilders.terms("user")
.field("user").size(0) // 根據user進行分組
// size(0) 也是10
).get();
}
}
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。
相關文章
Java中的數組復制（clone與arraycopy）代碼詳解
這篇文章主要介紹了Java中的數組復制（clone與arraycopy）代碼詳解，本文并未全部介紹數組復制的幾種方式，僅對clone和copy的相關內容進行了解析，具有一定參考價值，需要的朋友可以了解下。2017-11-11
解決Eclipse add external jars運行出現java.lang.NoClassDefFoundErro
本篇文章對Eclipse add external jars導致運行出現java.lang.NoClassDefFoundError的解決方法進行了詳細的分析介紹。需要的朋友參考下2013-05-05
Socket.IO是實現瀏覽器與服務器之間實時、雙向和基于事件的通信的工具庫,本文主要介紹了Springboot 集成 SocketIO的示例代碼,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-10-10
這篇文章主要介紹了Spring的定時任務@Scheduled源碼詳解,@Scheduled注解是包org.springframework.scheduling.annotation中的一個注解,主要是用來開啟定時任務,本文提供了部分實現代碼與思路,需要的朋友可以參考下2023-09-09
今天小編就為大家分享一篇用java實現在txt文本中寫數據和讀數據的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-07-07
在使用Mybatis-Plus新增的時候，我們往往想要id隨著數據庫自增，本文主要介紹了Mybatis-Plus設置全局或者局部ID自增的實現，具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-01-01
這篇文章主要介紹了Springboot前后端分離項目配置跨域實現過程解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-08-08
解決SpringBoot項目使用多線程處理任務時無法通過@Autowired注入bean問題
這篇文章主要介紹了SpringBoot項目使用多線程處理任務時無法通過@Autowired注入bean問題的解決方法,需要的朋友可以參考下2018-09-09
這篇文章主要介紹了java 中clone()的使用方法的相關資料,希望通過本文能幫助大家能掌握clone（）的克隆方法，需要的朋友可以參考下2017-09-09
多線程和并發(fā)問題是Java技術面試中面試官比較喜歡問的問題之一。在這里，從面試的角度列出了大部分重要的問題，但是你仍然應該牢固的掌握Java多線程基礎知識來對應日后碰到的問題2014-09-09