Elasticsearch的Java API的Cookbook

一、查詢索引的數據量

CountResponse response = client.prepareCount(indexname)
		        .setQuery(QueryBuilders.matchAllQuery()).setTypes(indextype)
		        .execute()
		        .actionGet();

二、判斷一個索引是否存在

client.admin().indices().prepareExists(indexName).execute().actionGet().isExists();

三、判斷索引中同時具備兩個字段A和B的文檔的個數

MetricsAggregationBuilder aggregation =
		        AggregationBuilders
		                .cardinality("agg").script("doc['A'].value + '#' + doc['B'].value");

四、設置布爾值查詢條件，設置查詢A、B、C三個條件，同時，設置過濾器，將時間過濾在dTFrom和dtTo的範圍內

SearchRequestBuilder srb = client.prepareSearch(actual_index_arr).setTypes(indexType);
		
		
		//添加查詢條件
		BoolQueryBuilder boolQuery = new BoolQueryBuilder();
		if(A!= null)
			boolQuery.must(QueryBuilders.matchQuery("A", A));
		if(B!= null)
			boolQuery.must(QueryBuilders.matchQuery("B", B));
		if(C!= null)
			boolQuery.must(QueryBuilders.matchQuery("C", C));		
		srb.setQuery(QueryBuilders.filteredQuery(boolQuery,
                FilterBuilders.rangeFilter("WRITETIME").from(dtFrom).to(dtTo))).addField("_id");//only return id

五、建立索引並設置文檔集合的mapping

if(isIndexExist(client, indexName) == false)
			client.admin().indices().prepareCreate(indexName).execute().actionGet();
		
		//read mapping
		BufferedReader br = new BufferedReader(new FileReader(file));
		String tmp = "";
		StringBuilder sb = new StringBuilder();
		while((tmp = br.readLine()) != null)
		{
			sb.append(tmp);
		}
		br.close();
		String ct = sb.toString();
		
		//make mapping
		client.admin().indices().preparePutMapping(indexName).setType(indexType).setSource(ct).execute().actionGet();

六、嵌套匯集查詢求和，相似 select k1,k2,sum(v) from t group by k1,k2

//設置group by k1, k2和sum(v)的彙集
TermsBuilder tb = AggregationBuilders.terms("k1BUILDER").field("k1").size(0);//
tb.subAggregation(AggregationBuilders.terms("k2BUILDER").field("k2").size(0).subAggregation(AggregationBuilders.sum("sum_by_value").field("v")));
srb.addAggregation(tb);		

//得到結果
SearchResponse sr = srb.get();		

//處理結果		
Terms agg = sr.getAggregations().get("k1BUILDER");		
//返回的彙集數據
for(Terms.Bucket entry : agg.getBuckets())
{			
	Terms sub_agg = entry.getAggregations().get("k2BUILDER");
	String eid_key = entry.getKey();   
	for(Terms.Bucket sub_entry : sub_agg.getBuckets())
	{	
		String k2_key = sub_entry.getKey();// Key		    
		Aggregations agg_its = sub_entry.getAggregations();
		Sum sm = agg_its.get("sum_by_value");
				
	}			
}

七、批量寫入數據

BulkProcessor bulkProcessor = BulkProcessor.builder(
	client,  
	new BulkProcessor.Listener() {		            
		@Override
		public void beforeBulk(long executionId, BulkRequest request) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void afterBulk(long executionId,
				BulkRequest request, BulkResponse response) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void afterBulk(long executionId,
				BulkRequest request, Throwable failure) {
			// TODO Auto-generated method stub
			
		} 
	})
	.setBulkActions(bulkCacheSize) 	
	.setFlushInterval(TimeValue.timeValueSeconds(5)) 
	.setConcurrentRequests(1) 		       
	.build();
	
for(int i=0;i<json_lst.size();i++)
{						
	//使用記錄裏的ID做爲主鍵，以保持數據的惟一性
	bulkProcessor.add(new IndexRequest(indexname, typename,  id_lst.get(i)).source(json_lst.get(i)));			
}
bulkProcessor.flush();
bulkProcessor.close();