elasticsearch parent-child

package com.prime.demo.util;


import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
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.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;

import com.prime.demo.beans.ESDocument;
import com.prime.demo.multithread.MessageService;
import com.prime.mcms.base.beans.ArticleBean;
import com.prime.mcms.base.utils.DateUtil;
import com.prime.mcms.base.utils.SecurityUtil;
import com.prime.smm.base.beans.TweetDocument;
import com.prime.smm.base.utils.PropConfig;
import com.prime.smm.base.utils.json.JSONArray;
import com.prime.smm.base.utils.json.JSONException;

public class ElasticSearchUtils {
    private static String clusterName = PropConfig.getFileConfig("elasticsearch", "es.clusterName");
    private static String indexname= PropConfig.getFileConfig("elasticsearch", "es.indexname");
    private static String indextype= PropConfig.getFileConfig("elasticsearch", "es.indextype");
    private static String message = MessageService.getInstance(4) ;
    static Map<String, String> m = new HashMap<String, String>();
    // 设置client.transport.sniff为true来使客户端去嗅探整个集群的状态，把集群中其它机器的ip地址加到客户端中，
    static Settings settings = ImmutableSettings.settingsBuilder().put(m).put("cluster.name",clusterName).put("client.transport.sniff", true).build();
    public ElasticSearchUtils() {}
    
    private static TransportClient client;
 
    static {
        try {
            Class<?> clazz = Class.forName(TransportClient.class.getName());
            Constructor<?> constructor = clazz.getDeclaredConstructor(Settings.class);
            constructor.setAccessible(true);
            client = (TransportClient) constructor.newInstance(settings);
            client.addTransportAddress(new InetSocketTransportAddress("192.168.9.103", 9300));
            client.addTransportAddress(new InetSocketTransportAddress("192.168.9.115", 9300));
            client.addTransportAddress(new InetSocketTransportAddress("192.168.9.116", 9300));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
    // 取得实例
    public static synchronized TransportClient getTransportClient() {
        return client;
    }
    
    public static void prepareIndex(){
        IndexResponse response = client.prepareIndex("tweet", "tweet").setSource(message).execute().actionGet();
    }
    
    
    /**
     * 插入数据
     */
    public boolean bulkBuilder(List<ArticleBean> objects,String indexName,String indexType){
        System.out.println(objects.size()) ;
        if(objects!=null && objects.size()>0){
            BulkRequestBuilder bulkRequest = client.prepareBulk();
            for(ArticleBean article : objects){
                System.out.println("--<>"+article.toString()) ;
                IndexRequestBuilder indexRequest = client.prepareIndex(indexName, indexType).setSource(article.toString());
                bulkRequest.add(indexRequest);
            }
            System.out.println(bulkRequest.toString()) ;
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
                System.out.println(bulkResponse.buildFailureMessage());
                  return false;
            }
            return true ;
        }
        return false ;
    }
    
    public static void bulkBuilder(){
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        IndexRequestBuilder indexRequest = client.prepareIndex("tweet", "tweet").setSource(message);
        bulkRequest.add(indexRequest);
        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
        if (bulkResponse.hasFailures()) {
            System.out.println(bulkResponse.buildFailureMessage());
        }
    }
    
    
    public static void bulkBuilder(List<TweetDocument> docs){
        if(docs!=null && docs.size()>0){
            BulkRequestBuilder bulkRequest = client.prepareBulk();
            for(TweetDocument tweet:docs){
                IndexRequestBuilder indexRequest = client.prepareIndex("tweet", "tweet").setSource(tweet.toString()).setId(tweet.getFolderDocId());
                bulkRequest.add(indexRequest);
            }
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
                System.out.println(bulkResponse.buildFailureMessage());
            }
        }
    }
    
    
    public static void bulkBuilderLists(List<ESDocument> docs){
        long t1 = System.currentTimeMillis();
        if(docs!=null && docs.size()>0){
            BulkRequestBuilder bulkRequest = client.prepareBulk();
            for(ESDocument tweet:docs){
                String tweeString = tweet.toString() ;
                String pMessage = ESDocument.parseParentDocument(tweeString).toString() ;
//                System.out.println("pMessage::"+pMessage) ;
                String cMessage = ESDocument.parseChildDocument(tweeString).toString() ;
//                System.out.println("cMessage::"+cMessage) ;
                IndexRequestBuilder indexRequest = client.prepareIndex("alert", "tweet").setSource(pMessage);
                bulkRequest.add(indexRequest);
                indexRequest = client.prepareIndex("alert", "index").setSource(cMessage).setParent(tweet.getFolderDocId());
                bulkRequest.add(indexRequest);
            }
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
                System.out.println(bulkResponse.buildFailureMessage());
            }
        }
        long t2 = System.currentTimeMillis();
        System.out.println("ES index doc size:"+docs.size()+" ,cost:"+(t2-t1)/1000+"s");
    }
    
    
    public static void bulkProcess() {
        BulkProcessor bulkProcessor = BulkProcessor.builder(client,
                new BulkProcessor.Listener() {
                    long a = 0 ;
                    long b = 0 ;
                    public void beforeBulk(long executionId,  BulkRequest request) {
                        a = System.currentTimeMillis() ;
                         System.out.println("beforeBulk numberOfActions = " + request.numberOfActions()+"|request:"+request.requests().size());
                    }
                    public void afterBulk(long executionId, BulkRequest request,  BulkResponse response) {
                        b = System.currentTimeMillis() ;
                        System.out.println("afterBulk"+(b-a)+" ms");
                    }
                    //璁剧疆ConcurrentRequest 涓�锛孴hrowable涓嶆姏閿�
                    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
//                        System.out.println("happen fail = " + failure.getMessage() + " cause = " +failure.getCause());
                    }
                })
                .setBulkActions(10000)
                .setBulkSize(new ByteSizeValue(20, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(10)
                .build();
                for(int i=0;i<1000;i++){
                    bulkProcessor.add(new IndexRequest("tweet", "tweet").source(message));
                }
        bulkProcessor.close();
        client.close();
    }
    
    public static void updateIndexRequest(){
        try {
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.index("alert");
            updateRequest.type("index");
            updateRequest.parent("tweet");        //parent-child更新必须加上父类
            updateRequest.id("3804792499471673");
            
//            更新ptIds
//            String ptIds = "[{\"catId\":27607,\"catTitle\":\"英菲尼迪\",\"contentMention\":1,\"docId\":\"2015112500000413192\",\"level\":1,\"titleMention\":0,\"treeType\":1}]" ;
//            updateRequest.doc(XContentFactory.jsonBuilder().startObject().field("ptIds", net.sf.json.JSONArray.fromObject(ptIds)).endObject());
            
            updateRequest.doc(XContentFactory.jsonBuilder().startObject().field("activeCount", 123).endObject());
            client.update(updateRequest).get();
            
            
            
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    
    /**
     * 执行搜索
     * @param queryBuilder
     * @param indexname
     * @param type
     * @return
     */
    public static  void searcher(QueryBuilder queryBuilder, String indexname, String type){
        SearchResponse searchResponse = client.prepareSearch(indexname).setTypes(type)
        .setQuery(queryBuilder)
        .execute()
        .actionGet();
        SearchHits hits = searchResponse.getHits() ;
        System.out.println("查询到记录数=" + hits.getTotalHits());
        SearchHit[] searchHists = hits.getHits();
        if(searchHists.length>0){
            for(SearchHit hit:searchHists){
                System.out.println("source:"+hit.getSource().toString()) ;
                String folderDocId = hit.getSource().get("folderDocId")+"";
                System.out.println("folderDocId:"+folderDocId) ;
                String ptIds = hit.getSource().get("ptIds")+"";
                System.out.println("ptIds:"+ptIds) ;
                String dtIds = hit.getSource().get("dtIds")+"";
                System.out.println("dtIds:"+dtIds) ;
                String terms = hit.getSource().get("terms")+"";
                System.out.println("terms:"+terms) ;
            }
        }
    }
    
    
    /**
     * 执行搜索
     * @param queryBuilder
     * @param indexname
     * @param type
     * @return
     */
    public static  void hasChildSearcher(QueryBuilder queryBuilder, String indexname, String type){
        SearchResponse searchResponse = client.prepareSearch(indexname)
        .setQuery(queryBuilder)
        .execute()
        .actionGet();
        SearchHits hits = searchResponse.getHits() ;
        System.out.println("查询到记录数=" + hits.getTotalHits());
        SearchHit[] searchHists = hits.getHits();
        if(searchHists.length>0){
            for(SearchHit hit:searchHists){
                System.out.println("source:"+hit.getSource().toString()) ;
                String folderDocId = hit.getSource().get("folderDocId")+"";
                System.out.println("folderDocId:"+folderDocId) ;
            }
        }
    }
    
    
    public static void main(String[] args) throws IOException {
        String link = "http://club.autohome.com.cn/bbs/brand-58-c-433-100.htm" ;
        ArticleBean articleBean = new ArticleBean();
        articleBean.setContent("测试我的文章内容123");
        articleBean.setTitle("这是我的文章标题");
        articleBean.setAuthorName("付金山");
        articleBean.setLink(link) ;
        articleBean.setPublishDate(DateUtil.date2String(new Date())) ;
        articleBean.setInputTime(DateUtil.date2String(new Date())) ;
        String id = SecurityUtil.SHA1(link);
        articleBean.setSysId(id);
        String projectIds = "[{\"projectId\":27607},{\"projectId\":27605},{\"projectId\":27606}]" ;
        net.sf.json.JSONArray projectJsonArray = new net.sf.json.JSONArray() ;
        if(projectIds!=null && !projectIds.equals("")){
            projectJsonArray = net.sf.json.JSONArray.fromObject(projectIds) ;
        }
        articleBean.setProjectIds(projectJsonArray) ;
        List<ArticleBean> objects= new ArrayList<ArticleBean>() ;
        objects.add(articleBean) ;
        new ElasticSearchUtils().bulkBuilder(objects, "article", "article") ;
        
        
//        ElasticSearchUtils.prepareIndex() ;
//        ElasticSearchUtils.bulkProcess() ;
//        ElasticSearchUtils.bulkBuilder() ;
//        ElasticSearchUtils.updateIndexRequest(); //更新索引
        
        
        /*
         * https://www.elastic.co/guide/en/elasticsearch/guide/current/parent-child-mapping.html
         * http://sense.qbox.io/gist/73f6d2f347a08bfe0c254a977a4a05a68d2f3a8d
         * */
        /**
        XContentBuilder cdoc=XContentFactory.jsonBuilder().startObject().field("b_num_prop",1).field("b_str_prop","部门1").endObject();
        IndexResponse cresponse = client.prepareIndex("company", "branch").setSource(cdoc).setId("1").execute().actionGet();
        System.out.println("branch:"+cresponse.getId()) ;
        XContentBuilder edoc=XContentFactory.jsonBuilder().startObject().field("_parent",1).field("e_child_num",1).field("e_child_str","工号1").endObject();
        IndexResponse eresponse = client.prepareIndex("company", "employee").setSource(edoc).setId("1").execute().actionGet();
        System.out.println("employee:"+eresponse.getId()) ;
        **/
        
        /** 建立父子关联的索引
        List<ESDocument> tweets = new ArrayList<ESDocument>() ;
        tweets.add(ESDocument.arrayTweetDocument(message)) ;
        if(tweets!=null && tweets.size()>0){
            BulkRequestBuilder bulkRequest = client.prepareBulk();
            for(ESDocument tweet:tweets){
                String tweeString = tweet.toString() ;
                String pMessage = ESDocument.parseParentDocument(tweeString).toString() ;
                System.out.println("pMessage:"+pMessage) ;
                String cMessage = ESDocument.parseChildDocument(tweeString).toString() ;
                System.out.println("cMessage:"+cMessage) ;
                IndexRequestBuilder indexRequest = client.prepareIndex("blog", "text").setSource(pMessage);
                bulkRequest.add(indexRequest);
                indexRequest = client.prepareIndex("blog", "index").setSource(cMessage).setParent(tweet.getFolderDocId());
                bulkRequest.add(indexRequest);
            }
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
                System.out.println(bulkResponse.buildFailureMessage());
            }
        }
        **/
        /**普通查询
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.termQuery("blogId", "3822866355643510"));
//        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.termQuery("title", "胆小鬼"));
//        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.queryStringQuery("ptIds.level:2"));  //数组查询
        System.out.println(queryBuilder.toString()) ;
        ElasticSearchUtils.searcher(queryBuilder, "alert", "index") ;
        **/
        
        /** 查询根据子类查询，返回父类结果
        QueryBuilder queryBuilder = QueryBuilders.hasChildQuery("indicators", QueryBuilders.termQuery("blogId","1ee526c4af7372ff04bd4e1d6e4cf872ed6b11a7")) ;
        System.out.println(queryBuilder.toString()) ;
        SearchResponse searchResponse = client.prepareSearch("blog").setTypes("blog").setQuery(queryBuilder).execute().actionGet() ;
        SearchHits hits = searchResponse.getHits() ;
        System.out.println("查询到记录数=" + hits.getTotalHits());
        SearchHit[] searchHists = hits.getHits();
        if(searchHists.length>0){
            for(SearchHit hit:searchHists){
                System.out.println("source:"+hit.getSource().toString()) ;
                String folderDocId = hit.getSource().get("folderDocId")+"";
                String campaignId = hit.getSource().get("campaignId")+"";
                String polar = hit.getSource().get("polar")+"";
                String view = hit.getSource().get("view")+"";
                
                System.out.println("folderDocId:"+folderDocId) ;
                System.out.println("campaignId:"+campaignId) ;
                System.out.println("polar:"+polar) ;
                System.out.println("view:"+view) ;
            }
        }
        **/
        
        /** 查询根据父类查询，返回子类结果
        QueryBuilder queryBuilder = QueryBuilders.hasParentQuery("text", QueryBuilders.boolQuery().must(QueryBuilders.queryStringQuery("terms.attr:67436545"))) ;
        System.out.println(queryBuilder.toString()) ;
        SearchResponse searchResponse = client.prepareSearch("blog").setTypes("index").setQuery(queryBuilder).execute().actionGet() ;
        SearchHits hits = searchResponse.getHits() ;
        System.out.println("查询到记录数=" + hits.getTotalHits());
        SearchHit[] searchHists = hits.getHits();
        if(searchHists.length>0){
            for(SearchHit hit:searchHists){
                System.out.println("source:"+hit.getSource().toString()) ;
                String folderDocId = hit.getSource().get("folderDocId")+"";
                String campaignId = hit.getSource().get("campaignId")+"";
                String polar = hit.getSource().get("polar")+"";
                String view = hit.getSource().get("view")+"";
                
                System.out.println("folderDocId:"+folderDocId) ;
                System.out.println("campaignId:"+campaignId) ;
                System.out.println("polar:"+polar) ;
                System.out.println("view:"+view) ;
            }
        }
        **/
        
        /** 查询根据子类查询，返回父和子类结果 方法一: 获取总数为2
        QueryBuilder queryBuilder = QueryBuilders.indicesQuery(QueryBuilders.termQuery("blogId","3903007903499940"), "indicators") ;
        System.out.println(queryBuilder.toString()) ;
        SearchResponse searchResponse = client.prepareSearch("blog").setQuery(queryBuilder).execute().actionGet() ;
        SearchHits hits = searchResponse.getHits() ;
        System.out.println("查询到记录数=" + hits.getTotalHits());
        SearchHit[] searchHists = hits.getHits();
        if(searchHists.length>0){
            for(SearchHit hit:searchHists){
                System.out.println("source:"+hit.getSource().toString()) ;
                String folderDocId = hit.getSource().get("folderDocId")+"";
                String campaignId = hit.getSource().get("campaignId")+"";
                String polar = hit.getSource().get("polar")+"";
                String view = hit.getSource().get("view")+"";
                
                System.out.println("folderDocId:"+folderDocId) ;
                System.out.println("campaignId:"+campaignId) ;
                System.out.println("polar:"+polar) ;
                System.out.println("view:"+view) ;
            }
        }
       */
        
        /** 查询根据子类查询，返回父和子类结果 方法二： 获取总数为1
        QueryBuilder queryBuilder = QueryBuilders.hasChildQuery("indicators", QueryBuilders.termQuery("blogId","1ee526c4af7372ff04bd4e1d6e4cf872ed6b11a7")).innerHit(new QueryInnerHitBuilder()) ; ;
        System.out.println(queryBuilder.toString()) ;
        SearchResponse searchResponse = client.prepareSearch("blog").setTypes("blog").setQuery(queryBuilder).execute().actionGet() ;
        SearchHits hits = searchResponse.getHits() ;
        System.out.println("blog indicators 查询到记录数=" + hits.getTotalHits());
        SearchHit[] searchHists = hits.getHits();
        if(searchHists.length>0){
            for(SearchHit hit:searchHists){
                System.out.println("source:"+hit.getSource().toString()) ;
                System.out.println("getInnerHits:"+hit.getInnerHits().toString()) ;
                String folderDocId = hit.getSource().get("folderDocId")+"";
                String campaignId = hit.getSource().get("campaignId")+"";
                System.out.println("folderDocId:"+folderDocId) ;
                System.out.println("campaignId:"+campaignId) ;
                
                Map<String, SearchHits> innerHits = hit.getInnerHits();
                SearchHits innerHit  = innerHits.get("indicators");
                System.out.println("indicators 查询到记录数=" + innerHit.getTotalHits());
                SearchHit[] searchinnerHits = innerHit.getHits();
                if(searchinnerHits.length>0){
                     for(SearchHit inner:searchinnerHits){
                         String polar = inner.getSource().get("polar")+"";
                         String view = inner.getSource().get("view")+"";
                         System.out.println("polar:"+polar) ;
                         System.out.println("view:"+view) ;
                     }
                }
            }
        }
        **/
        
      /**组合分组查询
      SearchResponse sr = client.prepareSearch("blog").setTypes("blog").setQuery(QueryBuilders.matchAllQuery())
                   .addAggregation(AggregationBuilders.terms("1").field("mediaType"))
                   .execute().actionGet();
      org.elasticsearch.search.aggregations.bucket.terms.Terms a = sr.getAggregations().get("1");
      for(org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket bk:a.getBuckets()){
        System.out.println("类型: "+bk.getKey()+"  分组统计数量 "+bk.getDocCount()+"  ");
      }
      System.out.println("聚合数量："+a.getBuckets().size());
      **/
    /**组合分组查询 http://www.nosqldb.cn/1393917706416.html
    SearchResponse response = client.prepareSearch("blog").setTypes("blog")
            .addAggregation(AggregationBuilders.terms("terms")
                    .field("mediaType"))
            .execute().actionGet();
    Terms terms = response.getAggregations().get("terms");
    for(org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket b:terms.getBuckets()){
        System.out.println("filedname:"+b.getKey()+"docCount:"+b.getDocCount());
    }
     **/
        
    /**    facet查询 http://www.chepoo.com/elasticsearch-facet-search-example.html
     String eSql = "*:*" ;
     QueryBuilder queryBuilder = QueryBuilders.queryStringQuery(eSql) ;
     SearchResponse searchResponse = client.prepareSearch("blog")
                   .setTypes("blog")
                   .addFacet(FacetBuilders.termsFacet("termFacet").fields("mediaType"))
                   .setQuery(queryBuilder)
                   .execute().actionGet() ;
     TermsFacet f = (TermsFacet) searchResponse.getFacets().facetsAsMap().get("termFacet");
     f.getTotalCount();      // Total terms doc count
     f.getOtherCount();      // Not shown terms doc count
     f.getMissingCount();    // Without term doc count   
     
     for (TermsFacet.Entry entry : f) {
         System.out.println(entry.getTerm() +"    count:"+entry.getCount());
         //entry.getTerm()为catid，分类id
         //entry.getCount()为该分类下的数量
     }
     **/
    }
}