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());
    }