基于Elasticsearch 6.2.2高级REST客户端获取嵌套字段去重键值对

嘿，刚好我之前处理过类似的需求，用Elasticsearch 6.2.2的高级REST客户端来实现你要的功能，主要有两种靠谱的方式，我给你详细说说：

方法一：利用Elasticsearch聚合（推荐） #

这是最符合ES设计理念的方式，性能也更好，适合数据量较大的场景。核心思路是通过两层聚合：第一层提取FunFacts里的所有键，第二层针对每个键聚合去重后的值。

1. 对应的DSL查询 #

{
  "size": 0,  // 不需要返回原始文档，只取聚合结果
  "aggs": {
    "fun_facts_keys": {
      "terms": {
        "script": "ctx._source.FunFacts.keySet()",  // 提取FunFacts的所有键
        "size": 1000  // 调整这个值覆盖所有可能的键数量
      },
      "aggs": {
        "distinct_values": {
          "terms": {
            "script": "ctx._source.FunFacts[terms.key]",  // 根据当前键提取对应值
            "size": 1000  // 调整这个值覆盖对应键的所有可能值数量
          }
        }
      }
    }
  }
}

2. 高级REST客户端的Java实现代码 #

import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.http.HttpHost;

public class EsAggregationExample {
    public static void main(String[] args) {
        // 初始化客户端
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(new HttpHost("localhost", 9200, "http")));

        try {
            SearchRequest searchRequest = new SearchRequest("DOCUMENTS");
            searchRequest.types("DOCUMENTS");

            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            sourceBuilder.size(0);  // 不返回原始文档

            // 构建第一层聚合：提取FunFacts的所有键
            TermsAggregationBuilder keyAgg = AggregationBuilders.terms("fun_facts_keys")
                    .script("ctx._source.FunFacts.keySet()")
                    .size(1000);

            // 构建第二层聚合：对每个键的值去重
            TermsAggregationBuilder valueAgg = AggregationBuilders.terms("distinct_values")
                    .script("ctx._source.FunFacts[terms.key]")
                    .size(1000);

            keyAgg.subAggregation(valueAgg);
            sourceBuilder.aggregation(keyAgg);

            searchRequest.source(sourceBuilder);

            // 执行查询
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);

            // 解析聚合结果，转换成你需要的格式
            Terms keyTerms = response.getAggregations().get("fun_facts_keys");
            Map<String, List<Object>> resultMap = new HashMap<>();

            for (Terms.Bucket keyBucket : keyTerms.getBuckets()) {
                String key = keyBucket.getKeyAsString();
                Terms valueTerms = keyBucket.getAggregations().get("distinct_values");
                List<Object> values = new ArrayList<>();
                for (Terms.Bucket valueBucket : valueTerms.getBuckets()) {
                    values.add(valueBucket.getKey());
                }
                resultMap.put(key, values);
            }

            // 输出结果，格式就是你要的{"FavColor": ["Green"], "Age": [32, 33], "FavFood": ["Pizza"]}
            System.out.println(new ObjectMapper().writeValueAsString(resultMap));

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

注意事项 #

• 确保ES允许使用脚本：在elasticsearch.yml中添加配置script.inline: true和script.stored: true（开发环境可以这么设置，生产环境要注意脚本安全）。
• 根据你的实际数据量调整size参数，避免遗漏键或值。

方法二：客户端遍历处理（仅适合小数据量） #

如果你的文档数量很少，可以先查询所有文档，然后在客户端手动收集FunFacts的键和去重值。这种方法简单但性能差，数据量大时不推荐。

Java代码示例 #

import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.http.HttpHost;
import org.elasticsearch.search.SearchHit;

public class EsClientProcessingExample {
    public static void main(String[] args) {
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(new HttpHost("localhost", 9200, "http")));

        try {
            SearchRequest searchRequest = new SearchRequest("DOCUMENTS");
            searchRequest.types("DOCUMENTS");

            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            sourceBuilder.size(Integer.MAX_VALUE);  // 注意：数据量大时不要这么用，会内存溢出
            searchRequest.source(sourceBuilder);

            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);

            // 临时用Set存值实现去重
            Map<String, Set<Object>> tempMap = new HashMap<>();
            for (SearchHit hit : response.getHits().getHits()) {
                Map<String, Object> funFacts = (Map<String, Object>) hit.getSourceAsMap().get("FunFacts");
                if (funFacts != null) {
                    for (Map.Entry<String, Object> entry : funFacts.entrySet()) {
                        tempMap.computeIfAbsent(entry.getKey(), k -> new HashSet<>()).add(entry.getValue());
                    }
                }
            }

            // 把Set转成List，得到目标格式
            Map<String, List<Object>> resultMap = new HashMap<>();
            tempMap.forEach((key, values) -> resultMap.put(key, new ArrayList<>(values)));

            System.out.println(new ObjectMapper().writeValueAsString(resultMap));

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者mrbarret