溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
基于員工信息的CRUD操作
/**
* 員工增刪改查的應用程序
*
* @author Administrator
*
*/
public class EmployeeCRUDApp {
@SuppressWarnings({ "unchecked", "resource" })
public static void main(String[] args) throws Exception {
// 先構建client
Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();
TransportClient client = new PreBuiltTransportClient(settings)
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
// createEmployee(client);
// getEmployee(client);
// updateEmployee(client);
// deleteEmployee(client);
client.close();
}
/**
* 創建員工信息（創建一個document）
*
* @param client
*/
private static void createEmployee(TransportClient client) throws Exception {
IndexResponse response = client.prepareIndex("company", "employee", "1")
.setSource(XContentFactory.jsonBuilder().startObject().field("name", "jack").field("age", 27)
.field("position", "technique").field("country", "china").field("join_date", "2017-01-01")
.field("salary", 10000).endObject())
.get();
System.out.println(response.getResult());
}
/**
* 獲取員工信息
*
* @param client
* @throws Exception
*/
private static void getEmployee(TransportClient client) throws Exception {
GetResponse response = client.prepareGet("company", "employee", "1").get();
System.out.println(response.getSourceAsString());
}
/**
* 修改員工信息
*
* @param client
* @throws Exception
*/
private static void updateEmployee(TransportClient client) throws Exception {
UpdateResponse response = client.prepareUpdate("company", "employee", "1")
.setDoc(XContentFactory.jsonBuilder().startObject().field("position", "technique manager").endObject())
.get();
System.out.println(response.getResult());
}
/**
* 刪除 員工信息
*
* @param client
* @throws Exception
*/
private static void deleteEmployee(TransportClient client) throws Exception {
DeleteResponse response = client.prepareDelete("company", "employee", "1").get();
System.out.println(response.getResult());
}
}基于員工信息的查詢操作
/**
* 員工搜索應用程序
*
* @author Administrator
*
*/
public class EmployeeSearchApp {
@SuppressWarnings({ "unchecked", "resource" })
public static void main(String[] args) throws Exception {
Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();
TransportClient client = new PreBuiltTransportClient(settings)
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
prepareData(client);
// executeSearch(client);
client.close();
}
/**
* 執行搜索操作
*
* @param client
*/
private static void executeSearch(TransportClient client) {
SearchResponse response = client.prepareSearch("company").setTypes("employee")
.setQuery(QueryBuilders.matchQuery("position", "technique"))
.setPostFilter(QueryBuilders.rangeQuery("age").from(30).to(40)).setFrom(0).setSize(1).get();
SearchHit[] searchHits = response.getHits().getHits();
for (int i = 0; i < searchHits.length; i++) {
System.out.println(searchHits[i].getSourceAsString());
}
}
/**
* 準備數據
*
* @param client
*/
private static void prepareData(TransportClient client) throws Exception {
client.prepareIndex("company", "employee", "1")
.setSource(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())
.get();
client.prepareIndex("company", "employee", "2")
.setSource(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())
.get();
client.prepareIndex("company", "employee", "3")
.setSource(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())
.get();
client.prepareIndex("company", "employee", "4")
.setSource(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())
.get();
client.prepareIndex("company", "employee", "5")
.setSource(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())
.get();
}
}基于員工新的聚合查詢操作
/**
* 員工聚合分析應用程序
*
* @author Administrator
*
*/
public class EmployeeAggrApp {
@SuppressWarnings({ "unchecked", "resource" })
public static void main(String[] args) throws Exception {
Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();
TransportClient client = new PreBuiltTransportClient(settings)
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
SearchResponse searchResponse = client.prepareSearch("company")
.addAggregation(AggregationBuilders.terms("group_by_country").field("country")
.subAggregation(AggregationBuilders.dateHistogram("group_by_join_date").field("join_date")
.dateHistogramInterval(DateHistogramInterval.YEAR)
.subAggregation(AggregationBuilders.avg("avg_salary").field("salary"))))
.execute().actionGet();
Map<String, Aggregation> aggrMap = searchResponse.getAggregations().asMap();
StringTerms groupByCountry = (StringTerms) aggrMap.get("group_by_country");
Iterator<Bucket> groupByCountryBucketIterator = groupByCountry.getBuckets().iterator();
while (groupByCountryBucketIterator.hasNext()) {
Bucket groupByCountryBucket = groupByCountryBucketIterator.next();
System.out.println(groupByCountryBucket.getKey() + ":" + groupByCountryBucket.getDocCount());
Histogram groupByJoinDate = (Histogram) groupByCountryBucket.getAggregations().asMap()
.get("group_by_join_date");
Iterator<org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Bucket> groupByJoinDateBucketIterator = groupByJoinDate
.getBuckets().iterator();
while (groupByJoinDateBucketIterator.hasNext()) {
org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Bucket groupByJoinDateBucket = groupByJoinDateBucketIterator
.next();
System.out.println(groupByJoinDateBucket.getKey() + ":" + groupByJoinDateBucket.getDocCount());
Avg avg = (Avg) groupByJoinDateBucket.getAggregations().asMap().get("avg_salary");
System.out.println(avg.getValue());
}
}
client.close();
}
}注：聚合查詢的時候可能出現問題：fielddata需要變為true，這個時候需要手動添加mapping
GET /company/_mapping/employee
{
"company": {
"mappings": {
"employee": {
"properties": {
"age": {
"type": "long"
},
"country": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"join_date": {
"type": "date"
},
"name": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"position": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"salary": {
"type": "long"
}
}
}
}
}
}
Delete /company
上面查出來以后進行截取，修改"fielddata": true
PUT /company
{
"mappings": {
"employee": {
"properties": {
"age": {
"type": "long"
},
"country": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
},
"fielddata": true
},
"join_date": {
"type": "date"
},
"name": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"position": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"salary": {
"type": "long"
}
}
}
}
}elasticsearch-5.2.0獲取代碼：
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.springframework.stereotype.Component;
import com.ad.utils.ConfigUtil;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Created by wangyunpeng on 2017/8/5.
*/
@Component
public class ElasticSearchClient {
@Resource(type=ConfigUtil.class)
private ConfigUtil configUtil;
private Map<String, Client> clientMap = new ConcurrentHashMap<String, Client>();
/*1.@PostConstruct說明
被@PostConstruct修飾的方法會在服務器加載Servlet的時候運行，并且只會被服務器調用一次，類似于Serclet的inti()方法。被@PostConstruct修飾的方法會在構造函數之后，init()方法之前運行。
2.@PreConstruct說明
被@PreConstruct修飾的方法會在服務器卸載Servlet的時候運行，并且只會被服務器調用一次，類似于Servlet的destroy()方法。被@PreConstruct修飾的方法會在destroy()方法之后運行，在Servlet被徹底卸載之前。
*/
@PostConstruct
public void init() {
init(configUtil.clusterName, configUtil.clusterIpAddress);
//init(configUtil.clusterName2, configUtil.clusterIpAddress2);
}
private void init(String clusterName, String clusterIpAddress){
try {
Settings settings = Settings.builder()
.put("cluster.name", clusterName)
.build();
addClient(settings, getAllAddress(clusterIpAddress));
}catch(Exception e){
e.printStackTrace();
}
}
/**
* 得所有的地址端口
* @param ips
* @return
* @throws Exception
*/
public List<TransportAddress> getAllAddress(String ips)throws Exception {
List<TransportAddress> addressList = new ArrayList<TransportAddress>();
if(StringUtils.isNotBlank(ips)&&ips.contains(",")){
String[] ipaddr=ips.split(",");
for (int i=0;i<ipaddr.length;i++) {
addressList.add(new TransportAddress(InetAddress.getByName(ipaddr[i]),configUtil.clusterPort));
}
}else{
addressList.add(new TransportAddress(InetAddress.getByName(ips),configUtil.clusterPort));
}
return addressList;
}
/**
* 添加es客戶端
* @param setting
* @param transportAddress
* @throws Exception
*/
public void addClient(Settings setting, List<TransportAddress> transportAddress) throws Exception{
TransportClient client = new PreBuiltTransportClient(setting);
for(int i=0;i<transportAddress.size();i++){
client.addTransportAddress(transportAddress.get(i));
}
clientMap.put(setting.get("cluster.name"), client);
}
public Client getClient(String clusterName) {
return clientMap.get(clusterName);
}
}elasticsearch-6.0.0獲取代碼：
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.springframework.stereotype.Component;
import com.ad.utils.ConfigUtil;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Created by wangyunpeng on 2017/8/5.
*/
@Component
public class ElasticSearchClient {
@Resource(type=ConfigUtil.class)
private ConfigUtil configUtil;
private Map<String, Client> clientMap = new ConcurrentHashMap<String, Client>();
/*1.@PostConstruct說明
被@PostConstruct修飾的方法會在服務器加載Servlet的時候運行，并且只會被服務器調用一次，類似于Serclet的inti()方法。被@PostConstruct修飾的方法會在構造函數之后，init()方法之前運行。
2.@PreConstruct說明
被@PreConstruct修飾的方法會在服務器卸載Servlet的時候運行，并且只會被服務器調用一次，類似于Servlet的destroy()方法。被@PreConstruct修飾的方法會在destroy()方法之后運行，在Servlet被徹底卸載之前。
*/ @PostConstruct
public void init() {
init(configUtil.clusterName, configUtil.clusterIpAddress);
//init(configUtil.clusterName2, configUtil.clusterIpAddress2);
}
private void init(String clusterName, String clusterIpAddress){
try {
Settings settings = Settings.builder()
.put("cluster.name", clusterName)
.build();
addClient(settings, getAllAddress(clusterIpAddress));
}catch(Exception e){
e.printStackTrace();
}
/*try {
Settings settings = TransportClient.builder().build().settings().builder().put("cluster.name", clusterName).build();
addClient(settings, getAllAddress(clusterIpAddress));
}catch(Exception e){
e.printStackTrace();
}*/
}
/**
* 得所有的地址端口
* @param ips
* @return
* @throws Exception
*/
public List<InetSocketTransportAddress> getAllAddress(String ips)throws Exception {
List<InetSocketTransportAddress> addressList = new ArrayList<InetSocketTransportAddress>();
if(StringUtils.isNotBlank(ips)&&ips.contains(",")){
String[] ipaddr=ips.split(",");
for (int i=0;i<ipaddr.length;i++) {
addressList.add(new InetSocketTransportAddress(InetAddress.getByName(ipaddr[i]),configUtil.clusterPort));
}
}else{
addressList.add(new InetSocketTransportAddress(InetAddress.getByName(ips),configUtil.clusterPort));
}
return addressList;
}
/**
* 添加es客戶端
* @param setting
* @param transportAddress
* @throws Exception
*/
/*TransportClient client = new PreBuiltTransportClient(settings)
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));*/
public void addClient(Settings setting, List<InetSocketTransportAddress> transportAddress) throws Exception{
TransportClient client = new PreBuiltTransportClient(setting);
//TransportClient client = TransportClient.builder().settings(setting).build();
for(int i=0;i<transportAddress.size();i++){
client.addTransportAddress(transportAddress.get(i));
}
clientMap.put(setting.get("cluster.name"), client);
}
public Client getClient(String clusterName) {
return clientMap.get(clusterName);
}
}maven依賴：
<!-- elasticsearch package --> <!-- <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>transport</artifactId> <version>5.2.2</version> </dependency> --> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>transport</artifactId> <version>6.0.0</version> </dependency>
測試代碼：
import java.io.IOException;
import java.util.UUID;
import javax.annotation.Resource;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import com.ad.base.AlbumIndexField;
import com.ad.utils.ConfigUtil;
import com.ad.utils.ExceptionUtil;
import net.sf.json.JSONObject;
@Component
public class ElasticSaveServiceImpl implements ElasticSaveService {
private static final Logger logger = LoggerFactory.getLogger(ElasticSearchServiceImpl.class);
@Resource(type = ElasticSearchClient.class)
private ElasticSearchClient elasticSearchClient;
@Resource(type = ConfigUtil.class)
private ConfigUtil configUtil;
/**
* @return
*/
public Client getClient(String clusterName) {
try {
return elasticSearchClient.getClient(clusterName);
} catch (Exception e) {
logger.error("ES獲取client失敗 :" + ExceptionUtil.stackTrace(e));
return null;
}
}
@Override
public void executeSave(String clusterName, String json1) {
JSONObject json = JSONObject.fromObject(json1);
String documentId = (UUID.randomUUID().toString().replaceAll("-", ""));
IndexResponse response;
try {
response = this.getClient(clusterName).prepareIndex(configUtil.indexName, configUtil.indexType, documentId)
.setSource(XContentFactory.jsonBuilder()
.startObject()
.field(AlbumIndexField.FID, json.getString("fid"))
.endObject())
.get();
} catch (IOException e) {
logger.error("===AdInfoConsumer consumer is exception", e);
//e.printStackTrace();
}
}
}免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
