1 // 批量索引数据
 2     public void indexHotSpotDataList(List<Hotspotdata> dataList) {
 3         if (dataList != null) {
 4             int size = dataList.size();
 5             if (size > 0) {
 6                 BulkRequestBuilder bulkRequest = client.prepareBulk();
 7                 for (int i = 0; i < size; ++i) {
 8                     Hotspotdata data = dataList.get(i);
 9                     String jsonSource = getIndexDataFromHotspotData(data);
10                     if (jsonSource != null) {
11                         bulkRequest.add(client
12                                 .prepareIndex(IndexName, TypeName,
13                                         data.getId().toString())
14                                 .setRefresh(true).setSource(jsonSource));
15                     }
16                 }
17 
18                 BulkResponse bulkResponse = bulkRequest.execute().actionGet();
19                 if (bulkResponse.hasFailures()) {
20                     Iterator<BulkItemResponse> iter = bulkResponse.iterator();
21                     while (iter.hasNext()) {
22                         BulkItemResponse itemResponse = iter.next();
23                         if (itemResponse.isFailed()) {
24                             log.error(itemResponse.getFailureMessage());
25                         }
26                     }
27                 }
28             }
29         }
30     }
31 
32     // 索引数据
33     public boolean indexHotspotData(Hotspotdata data) {
34         String jsonSource = getIndexDataFromHotspotData(data);
35         if (jsonSource != null) {
36             IndexRequestBuilder requestBuilder = client.prepareIndex(IndexName,
37                     TypeName).setRefresh(true);
38             requestBuilder.setSource(jsonSource)
39                     .execute().actionGet();
40             return true;
41         }
42 
43         return false;
44     }
45 
46     // 得到索引字符串
47     public String getIndexDataFromHotspotData(Hotspotdata data) {
48         String jsonString = null;
49         if (data != null) {
50             try {
51                 XContentBuilder jsonBuilder = XContentFactory.jsonBuilder();
52                 jsonBuilder.startObject().field(IDFieldName, data.getId())
53                         .field(SeqNumFieldName, data.getSeqNum())
54                         .field(IMSIFieldName, data.getImsi())
55                         .field(IMEIFieldName, data.getImei())
56                         .field(DeviceIDFieldName, data.getDeviceID())
57                         .field(OwnAreaFieldName, data.getOwnArea())
58                         .field(TeleOperFieldName, data.getTeleOper())
59                         .field(TimeFieldName, data.getCollectTime())
60                         .endObject();
61                 jsonString = jsonBuilder.string();
62             } catch (IOException e) {
63                 log.equals(e);
64             }
65         }
66 
67         return jsonString;
68     }

 1 // 获取少量数据100个
 2     private List<Integer> getSearchData(QueryBuilder queryBuilder) {
 3         List<Integer> ids = new ArrayList<>();
 4         SearchResponse searchResponse = client.prepareSearch(IndexName)
 5                 .setTypes(TypeName).setQuery(queryBuilder).setSize(100)
 6                 .execute().actionGet();
 7         SearchHits searchHits = searchResponse.getHits();
 8         for (SearchHit searchHit : searchHits) {
 9             Integer id = (Integer) searchHit.getSource().get("id");
10             ids.add(id);
11         }
12         return ids;
13     }
14 
15     // 获取大量数据
16     private List<Integer> getSearchDataByScrolls(QueryBuilder queryBuilder) {
17         List<Integer> ids = new ArrayList<>();
18         // 一次获取100000数据
19         SearchResponse scrollResp = client.prepareSearch(IndexName)
20                 .setSearchType(SearchType.SCAN).setScroll(new TimeValue(60000))
21                 .setQuery(queryBuilder).setSize(100000).execute().actionGet();
22         while (true) {
23             for (SearchHit searchHit : scrollResp.getHits().getHits()) {
24                 Integer id = (Integer) searchHit.getSource().get(IDFieldName);
25                 ids.add(id);
26             }
27             scrollResp = client.prepareSearchScroll(scrollResp.getScrollId())
28                     .setScroll(new TimeValue(600000)).execute().actionGet();
29             if (scrollResp.getHits().getHits().length == 0) {
30                 break;
31             }
32         }
33 
34         return ids;
35     }

 1 // 得到某段时间内设备列表上每个设备的数据分布情况<设备ID，数量>
 2     public Map<String, String> getDeviceDistributedInfo(String startTime,
 3             String endTime, List<String> deviceList) {
 4 
 5         Map<String, String> resultsMap = new HashMap<>();
 6 
 7         QueryBuilder deviceQueryBuilder = getDeviceQueryBuilder(deviceList);
 8         QueryBuilder rangeBuilder = getDateRangeQueryBuilder(startTime, endTime);
 9         QueryBuilder queryBuilder = QueryBuilders.boolQuery()
10                 .must(deviceQueryBuilder).must(rangeBuilder);
11 
12         TermsBuilder termsBuilder = AggregationBuilders.terms("DeviceIDAgg").size(Integer.MAX_VALUE)
13                 .field(DeviceIDFieldName);
14         SearchResponse searchResponse = client.prepareSearch(IndexName)
15                 .setQuery(queryBuilder).addAggregation(termsBuilder)
16                 .execute().actionGet();
17         Terms terms = searchResponse.getAggregations().get("DeviceIDAgg");
18         if (terms != null) {
19             for (Terms.Bucket entry : terms.getBuckets()) {
20                 resultsMap.put(entry.getKey(),
21                         String.valueOf(entry.getDocCount()));
22             }
23         }
24         return resultsMap;
25     }