ES練習代碼

  

package elasticsearch;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.elasticsearch.action.bulk.BulkItemResponse;
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.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.collect.ImmutableList;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.metrics.sum.Sum;
import org.elasticsearch.search.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.junit.Before;
import org.junit.Test;

import com.fasterxml.jackson.databind.ObjectMapper;

public class TestEs {
    
    TransportClient transportClient = new TransportClient();
    //before表示在執行每個test方法之前運行
    @Before
    public void test0() throws Exception {
        //指定es集群中的節點信息
        TransportAddress transportAddress = new InetSocketTransportAddress("192.168.1.99",9300);
        //TransportAddress transportAddress1 = new InetSocketTransportAddress("192.168.1.98",9300);
        transportClient.addTransportAddresses(transportAddress);
    }
    
    /**
     * 通過java代碼操作es-1
     * 在實際工作中這樣寫不是很靠譜,需要完善,做測試可以
     * @throws Exception
     */
    @Test
    public void test1() throws Exception {
        //通過TransportClient可以和es集群交互
        //TransportClient transportClient = new TransportClient();
        
        //指定es集群中的節點信息, 這個地方指定的端口是節點和節點之間的通信端口是9300,不是Http請求的端口9200.
        TransportAddress transportAddress = new InetSocketTransportAddress("192.168.1.99",9300);
        TransportAddress transportAddress1 = new InetSocketTransportAddress("192.168.1.98",9300);
        transportClient.addTransportAddresses(transportAddress,transportAddress1);//加入多個地址
        
        //獲取當前transportClient連接到了集群多少個節點
        ImmutableList<DiscoveryNode> connectedNodes = transportClient.connectedNodes();
        for (DiscoveryNode discoveryNode : connectedNodes) {
            System.out.println(discoveryNode.getHostAddress());//打印192.168.1.99;192.168.1.98
            //如果加入transportClient.addTransportAddresses(transportAddress)  只有一個ip,打印的就只有一個.
        }
    }
    
    
    
    /**
     * 通過java代碼操作es-2
     * 實際工作中使用的時候建議加上下面這些配置信息
     * @throws Exception
     */
    @Test
    public void test2() throws Exception {
        //指定es的配置信息   Immutable不可改變的;
        Settings settings = ImmutableSettings.settingsBuilder()
                .put("cluster.name", "elasticsearch")//如果集群名稱在配置文件中被修改了，那么在這需要顯式定義一下
                //es集群名稱默認是 elasticsearch  sniff嗅; 發現;
                .put("client.transport.sniff", true)//開啟集群的嗅探功能，這樣可以保證es會自動把集群中的其他節點信息添加到transportClient里面
                //開啟嗅探功能后 只要指定集群中的任意一個可用節點就可以了.當把代碼運行之后TransportClient里面會把集群中所有節點的信息都拿到,能識別集群中的所有節點.
                .build();
        
        //通過TransportClient可以和es集群交互
        TransportClient transportClient = new TransportClient(settings);
        //指定es集群中的節點信息
        TransportAddress transportAddress = new InetSocketTransportAddress("192.168.1.99",9300);
        transportClient.addTransportAddresses(transportAddress);
        
        //獲取當前transportClient連接到了集群多少個節點
        ImmutableList<DiscoveryNode> connectedNodes = transportClient.connectedNodes();
        for (DiscoveryNode discoveryNode : connectedNodes) {
            System.out.println(discoveryNode.getHostAddress()); //雖然前面只指定了1.99 但是打印192.168.1.99  192.168.1.98
        }
    }
    
    String index = "crxy";
    String type = "emp";    
    /**
     * index-1
     * @throws Exception
     */
    @Test
    public void test3() throws Exception {
        String jsonStr = "{\"name\":\"zs\",\"age\":20}";//向索引庫中傳入一個String字符串,還可以接受其他類型
        IndexResponse indexResponse = transportClient.prepareIndex(index, type, "7")//添加一個id=7的數據
                            .setSource(jsonStr)
                            //.execute().actionGet();   這個和下面的get()方法是一樣的,get()就是對.execute().actionGet() 進行了封裝
                            .get();//執行
        System.out.println(indexResponse.getVersion());//得到這個數據的version,如果version=1代表是新添加的數據
    }
    
    /**
     * index-2
     * 實際工作中使用
     * @throws Exception
     */
    @Test
    public void test4() throws Exception {//把hashmap類型的數據放入index庫
        HashMap<String, Object> hashMap = new HashMap<String, Object>();
        hashMap.put("name", "heeh");
        hashMap.put("age", 20);
        IndexResponse indexResponse = transportClient.prepareIndex(index, type, "8").setSource(hashMap).get();
        System.out.println(indexResponse.getVersion());
    }
    
    /**
     * index -3
     * 實際工作中使用
     * 使用對象的時候需要把對象中的屬性轉化成json字符串
     * 
     * <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.1.3</version>
        </dependency>
     * @throws Exception
     */
    @Test
    public void test5() throws Exception {//傳入一個對象到index索引庫
        Person person = new Person();
        person.setName("lisi");
        person.setAge(30);
        
        //如果直接傳入一個person對象會報錯,java.lang.IllegalArgumentException,必須把對象轉換成一個Json字符串,使用jackson依賴
        //IndexResponse indexResponse = transportClient.prepareIndex(index, type, "9").setSource(person).get();
        
        ObjectMapper objectMapper = new ObjectMapper();
        IndexResponse indexResponse = transportClient.prepareIndex(index, type, "9").setSource(objectMapper.writeValueAsString(person)).get();
        System.out.println(indexResponse.getVersion());
    }
    
    /**
     * index -4
     * 測試數據這樣使用
     * @throws Exception
     */
    @Test
    public void test6() throws Exception {
        XContentBuilder builder = XContentFactory.jsonBuilder()//XContentFactory 這個是ES官方提供的可以構建Json字符串的工具類.
                .startObject()//{
                .field("name", "zs")
                .field("age", 18)
                .endObject();//}
        IndexResponse indexResponse = transportClient.prepareIndex(index, type, "11").setSource(builder).get();
        System.out.println(indexResponse.getVersion());
    }
    
    
    
    /**
     * 查詢 通過id
     * @throws Exception
     */
    @Test
    public void test7() throws Exception {
        GetResponse getResponse = transportClient.prepareGet(index, type, "9").get();//查詢id為9的數據
        System.out.println(getResponse.getSourceAsString());
    }
    
    /**
     * 局部更新
     * @throws Exception
     */
    @Test
    public void test8() throws Exception {
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject().field("age", 19).endObject();
        UpdateResponse updateResponse = transportClient.prepareUpdate(index, type, "9").setDoc(builder).get();
        System.out.println(updateResponse.getVersion());//version打印2 數據更新
    }
    
    /**
     * 刪除-通過id刪除
     * @throws Exception
     */
    @Test
    public void test9() throws Exception {
        DeleteResponse deleteResponse = transportClient.prepareDelete(index, type, "5").get();//刪除比較簡單
    }
    
    /**
     * 求總數
     * 類似於mysql中的select count(*)
     * @throws Exception
     */
    @Test
    public void test10() throws Exception {//查找索引庫中的數據個數
        long count = transportClient.prepareCount(index).get().getCount();
        System.out.println(count);
    }
    
    
    /**
     * 批量操作 bulk
     * @throws Exception
     */
    @Test
    public void test11() throws Exception {
        BulkRequestBuilder bulkBuilder = transportClient.prepareBulk();
        
        IndexRequest indexRequest = new IndexRequest(index, type, "12");
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject().field("name", "haha").field("age", 18).endObject();
        indexRequest.source(builder);
        
        DeleteRequest deleteRequest = new DeleteRequest(index, type, "13");
        
        bulkBuilder.add(indexRequest);//bulkBuilder中可以添加多個操作,這里一個是建立索引的操作,一個是刪除的操作.
        bulkBuilder.add(deleteRequest);
        
        BulkResponse bulkResponse = bulkBuilder.get();
        if(bulkResponse.hasFailures()){//批量操作中可能有的操作會出現問題,這個地方對操作失敗的處理
            //獲取所有錯誤信息，並打印
            BulkItemResponse[] items = bulkResponse.getItems();
            for (BulkItemResponse bulkItemResponse : items) {
                System.out.println(bulkItemResponse.getFailureMessage());
            }
        }else{
            System.out.println("全部OK");    
        }
        
    }
    
    
    /**
     * 查詢
     * lt:小於
     * lte:小於等於
     * gt:大於
     * gte:大於等於
     * 
     * @throws Exception
     */
    @Test
    public void test12() throws Exception {
        SearchResponse searchResponse = transportClient.prepareSearch(index)//指定索引庫
            .setTypes(type)//指定類型
            .setQuery(QueryBuilders.matchQuery("name", "zs"))//指定查詢條件,不支持通配符
            //.setQuery(QueryBuilders.multiMatchQuery("zs", "name","title"))//根據多個屬性進行查詢
            //.setQuery(QueryBuilders.matchAllQuery())//查詢所有
            //.setQuery(QueryBuilders.queryString("name:z* AND age:20"))//支持通配符* ?，可以實現復雜查詢，可以使用AND OR 之類的運算符(運算符要大寫)
            //.setQuery(QueryBuilders.termQuery("name", "zs"))//在查詢的時候不分詞，主要針對 人名 地名等特殊的詞語
            //工作中沒有說明特殊需求，就是用默認的查詢類型，如果對搜索准確度要求非常高，建議使用DFS_QUERY_THEN_FETCH,如果只追求查詢效果，對其他的指標不關心，可以使用QUERY_AND_FETCH
            .setSearchType(SearchType.QUERY_THEN_FETCH)//指定查詢類型，可以指定四種
            .setExplain(true)//返回的數據按照搜索詞的相關度排序
            //分頁參數
            .setFrom(0)
            .setSize(10)
            //根據某一個字段排序
            .addSort("age",SortOrder.DESC)
            //過濾
            //.setPostFilter(FilterBuilders.rangeFilter("age").from(0).to(18).includeLower(true).includeUpper(false))//默認是閉區間
            //.setPostFilter(FilterBuilders.rangeFilter("age").gt(0).lt(18))
            //實現高亮
            .addHighlightedField("name")//設置高亮字段
            .setHighlighterPreTags("<font color='red'>")//設置高亮前綴和后綴
            .setHighlighterPostTags("</font>")
            .get();
        SearchHits hits = searchResponse.getHits();
        long totalHits = hits.getTotalHits();
        System.out.println("總數："+totalHits);
        SearchHit[] hits2 = hits.getHits();
        for (SearchHit searchHit : hits2) {
            //獲取高亮內容
            Map<String, HighlightField> highlightFields = searchHit.getHighlightFields();
            HighlightField highlightField = highlightFields.get("name");
            if(highlightField!=null){
                String name_h = "";
                Text[] fragments = highlightField.fragments();
                for (Text text : fragments) {
                    name_h+=text;
                }
                System.out.println("高亮內容："+name_h);
            }
            System.out.println(searchHit.getSourceAsString());
        }
    }
    
    //ES是一個分布式是搜索引擎,天生就是為分布式而生的,但是分布式有優點,也是有缺點.
    //一個索引庫crxy的數據分布到了5個分片,去哪個分片去查,由於是分布式,可能每個分片中都有數據,所以一定要到所有分片中去查找.
    
    
    
    /**
     * 根據查詢條件刪除數據
     * @throws Exception
     */
    @Test
    public void test13() throws Exception {
        DeleteByQueryResponse deleteByQueryResponse = transportClient.prepareDeleteByQuery(index)
                .setQuery(QueryBuilders.matchAllQuery())
                .get();
    }
    
    
    /**
     * 統計分析-count
     * 根據年齡進行分組，統計相同年齡的數據有多少條
     * 
     * 默認情況下，如果分組個數大於10條的話，默認只會返回前10條分組數據
     * 如果想獲取所有分組數據，或者想要獲取指定數量的分組數據，如何實現呢？
     *     .size(0)
     * 
     * @throws Exception
     */
    @Test
    public void test14() throws Exception {
        SearchResponse searchResponse = transportClient.prepareSearch(index)
            .setTypes(type)
            .addAggregation(AggregationBuilders.terms("age_term").field("age").size(0))//給分組起個名稱，並且指定分組字段
            .get();
        
        Terms terms = searchResponse.getAggregations().get("age_term");//指定分組的名字.
        //獲取分組數據
        List<Bucket> buckets = terms.getBuckets();
        for (Bucket bucket : buckets) {
            System.out.println(bucket.getKey()+"----"+bucket.getDocCount());
        }
    }
    /*打印
     21----3
     18----2
     19----1
     20----1
     */
    
    
    /**
     * 統計分析-sum   select name,sum(score) from table group by name;
     * 需要使用.subAggregation
     * @throws Exception
     */
    /**
         使用數據:
        #aggregations-2
        curl -XPUT 'localhost:9200/crxy/emp/1' -d'{"name":"zs","score":60}'
        curl -XPUT 'localhost:9200/crxy/emp/2' -d'{"name":"zs","score":90}'
        curl -XPUT 'localhost:9200/crxy/emp/3' -d'{"name":"ls","score":80}'
        curl -XPUT 'localhost:9200/crxy/emp/4' -d'{"name":"ls","score":70}'
     */
    @Test
    public void test15() throws Exception {
        SearchResponse searchResponse = transportClient.prepareSearch(index)
            .setTypes(type)
            .addAggregation(AggregationBuilders.terms("name_term").field("name")//指定分組字段
                    .subAggregation(AggregationBuilders.sum("score_sum").field("score")))//指定求sum的字段
            .get();
        
        Terms terms = searchResponse.getAggregations().get("name_term");
        List<Bucket> buckets = terms.getBuckets();
        for (Bucket bucket : buckets) {
            Sum sum = bucket.getAggregations().get("score_sum");
            System.out.println(bucket.getKey()+"----"+sum.getValue());
        }
    }
    /*
     * 打印輸出
        ls----150.0
        zs----150.0
     */
    
    
    /**
     * 指定分片查詢(_shards),指定某個節點(_only_node)和某些節點(自定義的_only_nodes)
     * @throws Exception
     */
    @Test
    public void test16() throws Exception {
        SearchResponse searchResponse = transportClient.prepareSearch(index)
                .setTypes(type)
                .setQuery(QueryBuilders.matchAllQuery())
                //.setPreference("_shards:0,2") //.setPreference("_local")   .setPreference("")
                .setPreference("_only_node:8PoWbRVvQQ6NU283Bfd_7A,BJwexRvDTJ-VRx7Njs8uxA")//8PoWbRVvQQ6NU283Bfd_7A是一個節點的id
                .get();
        
        SearchHits hits = searchResponse.getHits();
        long totalHits = hits.getTotalHits();
        System.out.println("總數："+totalHits);
        SearchHit[] hits2 = hits.getHits();
        for (SearchHit searchHit : hits2) {
            System.out.println(searchHit.getSourceAsString());
        }
    }
    
}

 

 

    @Test
    public void test16() throws Exception {
        SearchResponse searchResponse = transportClient.prepareSearch("crxy*")
                .setTypes(type)
                .setQuery(QueryBuilders.matchAllQuery())
                //查詢指定分片的數據可以使用下面這兩種方案，第一種直接指定分片id，第二種根據routing參數的值計算分片id
                //.setPreference("_shards:0")
                //.setRouting("hehe")
                .get();
        
        SearchHits hits = searchResponse.getHits();
        long totalHits = hits.getTotalHits();
        System.out.println("總數："+totalHits);
        SearchHit[] hits2 = hits.getHits();
        for (SearchHit searchHit : hits2) {
            System.out.println(searchHit.getSourceAsString());
        }
    }
    
    
    @Test
    public void test17() throws Exception {
        HashMap<String, Object> hashMap = new HashMap<String, Object>();
        hashMap.put("name", "hello world");
        hashMap.put("age", 20);
        IndexResponse indexResponse = transportClient.prepareIndex(index, type, "7")
                .setSource(hashMap)
                .setRouting("heha")//指定一個路由參數，參數相同的數據會保存到同一個分片
                .get();
        System.out.println(indexResponse.getVersion());
    }