import org.elasticsearch.search.aggregations.metrics.tophits.TopHits; //导入依赖的package包/类
@Override
public <T> FacetedPageImpl<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
	Aggregations aggregations = response.getAggregations();
	Terms agg = aggregations.get(term) ;
	long total = agg.getSumOfOtherDocCounts() ;
	List<T> results = new ArrayList<T>();
	if(agg.getBuckets()!=null && agg.getBuckets().size()>0){
		for (Terms.Bucket entry : agg.getBuckets()) {
			if(!StringUtils.isBlank(name) && entry.getAggregations().get(name)!=null){
				TopHits topHits = entry.getAggregations().get(name);
				for (SearchHit hit : topHits.getHits().getHits()) {
					T data = mapEntity(hit.getSourceAsString() , hit , clazz) ;
					if(data instanceof UKAgg){
						((UKAgg) data).setRowcount((int) topHits.getHits().getTotalHits());
					}
					results.add(data) ;
				}
			}
		}
	}
	return new FacetedPageImpl<T>(results, pageable, total);
}
 

import org.elasticsearch.search.aggregations.metrics.tophits.TopHits; //导入依赖的package包/类
public void testEmptyIndex() throws Exception {
    SearchResponse response = client().prepareSearch("empty").setTypes("type")
            .addAggregation(topHits("hits"))
            .get();
    assertSearchResponse(response);

    TopHits hits = response.getAggregations().get("hits");
    assertThat(hits, notNullValue());
    assertThat(hits.getName(), equalTo("hits"));
    assertThat(hits.getHits().getTotalHits(), equalTo(0L));
}
 

import org.elasticsearch.search.aggregations.metrics.tophits.TopHits; //导入依赖的package包/类
@Override
public void onResponse(SearchResponse response) {
    long timeTaken = System.currentTimeMillis() - timeStartESQuery;

    Aggregations aggregs = response.getAggregations();

    Terms agg = aggregs.get("partition");

    int numhits = 0;
    int numBuckets = 0;
    int alreadyprocessed = 0;

    synchronized (buffer) {
        // For each entry
        for (Terms.Bucket entry : agg.getBuckets()) {
            String key = (String) entry.getKey(); // bucket key
            long docCount = entry.getDocCount(); // Doc count

            int hitsForThisBucket = 0;

            // filter results so that we don't include URLs we are already
            // being processed
            TopHits topHits = entry.getAggregations().get("docs");
            for (SearchHit hit : topHits.getHits().getHits()) {
                hitsForThisBucket++;

                Map<String, Object> keyValues = hit.sourceAsMap();
                String url = (String) keyValues.get("url");

                LOG.debug("{} -> id [{}], _source [{}]", logIdprefix,
                        hit.getId(), hit.getSourceAsString());

                // is already being processed - skip it!
                if (beingProcessed.containsKey(url)) {
                    alreadyprocessed++;
                    continue;
                }
                Metadata metadata = fromKeyValues(keyValues);
                buffer.add(new Values(url, metadata));
            }

            if (hitsForThisBucket > 0)
                numBuckets++;

            numhits += hitsForThisBucket;

            LOG.debug("{} key [{}], hits[{}], doc_count [{}]", logIdprefix,
                    key, hitsForThisBucket, docCount, alreadyprocessed);
        }

        // Shuffle the URLs so that we don't get blocks of URLs from the
        // same
        // host or domain
        Collections.shuffle((List) buffer);
    }

    LOG.info(
            "{} ES query returned {} hits from {} buckets in {} msec with {} already being processed",
            logIdprefix, numhits, numBuckets, timeTaken, alreadyprocessed);

    esQueryTimes.addMeasurement(timeTaken);
    eventCounter.scope("already_being_processed").incrBy(alreadyprocessed);
    eventCounter.scope("ES_queries").incrBy(1);
    eventCounter.scope("ES_docs").incrBy(numhits);

    // remove lock
    isInESQuery.set(false);
}
 

import org.elasticsearch.search.aggregations.metrics.tophits.TopHits; //导入依赖的package包/类
private  void handleAggregations(Aggregations aggregations, List<String> headers, List<List<String>> lines) throws CsvExtractorException {
    if(allNumericAggregations(aggregations)){
        lines.get(this.currentLineIndex).addAll(fillHeaderAndCreateLineForNumericAggregations(aggregations, headers));
        return;
    }
    //aggregations with size one only supported when not metrics.
    List<Aggregation> aggregationList = aggregations.asList();
    if(aggregationList.size() > 1){
        throw new CsvExtractorException("currently support only one aggregation at same level (Except for numeric metrics)");
    }
    Aggregation aggregation = aggregationList.get(0);
    //we want to skip singleBucketAggregations (nested,reverse_nested,filters)
    if(aggregation instanceof SingleBucketAggregation){
        Aggregations singleBucketAggs = ((SingleBucketAggregation) aggregation).getAggregations();
        handleAggregations(singleBucketAggs, headers, lines);
        return;
    }
    if(aggregation instanceof NumericMetricsAggregation){
        handleNumericMetricAggregation(headers, lines.get(currentLineIndex), aggregation);
        return;
    }
    if(aggregation instanceof GeoBounds){
        handleGeoBoundsAggregation(headers, lines, (GeoBounds) aggregation);
        return;
    }
    if(aggregation instanceof TopHits){
        //todo: handle this . it returns hits... maby back to normal?
        //todo: read about this usages
        // TopHits topHitsAggregation = (TopHits) aggregation;
    }
    if(aggregation instanceof MultiBucketsAggregation){
        MultiBucketsAggregation bucketsAggregation = (MultiBucketsAggregation) aggregation;
        String name = bucketsAggregation.getName();
        //checking because it can comes from sub aggregation again
        if(!headers.contains(name)){
            headers.add(name);
        }
        Collection<? extends MultiBucketsAggregation.Bucket> buckets = bucketsAggregation.getBuckets();

        //clone current line.
        List<String> currentLine = lines.get(this.currentLineIndex);
        List<String> clonedLine = new ArrayList<>(currentLine);

        //call handle_Agg with current_line++
        boolean firstLine = true;
        for (MultiBucketsAggregation.Bucket bucket : buckets) {
            //each bucket need to add new line with current line copied => except for first line
            String key = bucket.getKeyAsString();
            if(firstLine){
                firstLine = false;
            }
            else {
                currentLineIndex++;
                currentLine = new ArrayList<String>(clonedLine);
                lines.add(currentLine);
            }
            currentLine.add(key);
            handleAggregations(bucket.getAggregations(),headers,lines);

        }
    }

}
 

import org.elasticsearch.search.aggregations.metrics.tophits.TopHits; //导入依赖的package包/类
private void handleAggregations(Aggregations aggregations, List<String> headers, List<List<Object>> lines) throws ObjectResultsExtractException {
    if (allNumericAggregations(aggregations)) {
        lines.get(this.currentLineIndex).addAll(fillHeaderAndCreateLineForNumericAggregations(aggregations, headers));
        return;
    }
    //aggregations with size one only supported when not metrics.
    List<Aggregation> aggregationList = aggregations.asList();
    if (aggregationList.size() > 1) {
        throw new ObjectResultsExtractException("currently support only one aggregation at same level (Except for numeric metrics)");
    }
    Aggregation aggregation = aggregationList.get(0);
    //we want to skip singleBucketAggregations (nested,reverse_nested,filters)
    if (aggregation instanceof SingleBucketAggregation) {
        Aggregations singleBucketAggs = ((SingleBucketAggregation) aggregation).getAggregations();
        handleAggregations(singleBucketAggs, headers, lines);
        return;
    }
    if (aggregation instanceof NumericMetricsAggregation) {
        handleNumericMetricAggregation(headers, lines.get(currentLineIndex), aggregation);
        return;
    }
    if (aggregation instanceof GeoBounds) {
        handleGeoBoundsAggregation(headers, lines, (GeoBounds) aggregation);
        return;
    }
    if (aggregation instanceof TopHits) {
        //todo: handle this . it returns hits... maby back to normal?
        //todo: read about this usages
        // TopHits topHitsAggregation = (TopHits) aggregation;
    }
    if (aggregation instanceof MultiBucketsAggregation) {
        MultiBucketsAggregation bucketsAggregation = (MultiBucketsAggregation) aggregation;
        String name = bucketsAggregation.getName();
        //checking because it can comes from sub aggregation again
        if (!headers.contains(name)) {
            headers.add(name);
        }
        Collection<? extends MultiBucketsAggregation.Bucket> buckets = bucketsAggregation.getBuckets();

        //clone current line.
        List<Object> currentLine = lines.get(this.currentLineIndex);
        List<Object> clonedLine = new ArrayList<>(currentLine);

        //call handle_Agg with current_line++
        boolean firstLine = true;
        for (MultiBucketsAggregation.Bucket bucket : buckets) {
            //each bucket need to add new line with current line copied => except for first line
            String key = bucket.getKeyAsString();
            if (firstLine) {
                firstLine = false;
            } else {
                currentLineIndex++;
                currentLine = new ArrayList<Object>(clonedLine);
                lines.add(currentLine);
            }
            currentLine.add(key);
            handleAggregations(bucket.getAggregations(), headers, lines);

        }
    }

}