如何在ElasticSearch或Java API中聚合拼接文档中的字符串？

可以通过ElasticSearch的脚本聚合（Scripted Metric Aggregation） 实现字符串拼接，Java API也能对应实现相同逻辑。以下是具体实现方式：

一、ES DSL实现（直接通过请求完成聚合拼接） #

使用scripted_metric聚合自定义拼接逻辑，它支持四个阶段的脚本定义来完成数据收集与合并：

POST /your_index/_search
{
  "size": 0,
  "aggs": {
    "merged_address": {
      "scripted_metric": {
        "init_script": "state.addresses = []",
        "map_script": "state.addresses.add(doc['address'].value)",
        "combine_script": "return state.addresses.join(', ')",
        "reduce_script": "return states.join(', ')"
      }
    }
  }
}

逻辑说明： #

• init_script：初始化一个空数组用于存储所有address字段值
• map_script：遍历每条文档，将address字段值加入数组
• combine_script：在每个分片上把数组元素用, 拼接成字符串
• reduce_script：把所有分片的拼接结果再次合并成最终字符串

如果需要按customerId分组拼接（比如示例中可能存在的分组需求），可以在外层套terms聚合：

POST /your_index/_search
{
  "size": 0,
  "aggs": {
    "group_by_customer": {
      "terms": {
        "field": "customerId"
      },
      "aggs": {
        "merged_address": {
          "scripted_metric": {
            "init_script": "state.addresses = []",
            "map_script": "state.addresses.add(doc['address'].value)",
            "combine_script": "return state.addresses.join(', ')",
            "reduce_script": "return states.join(', ')"
          }
        }
      }
    }
  }
}

二、Java API实现 #

对应上述DSL，用ElasticSearch Java High Level Client构建聚合请求：

import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.scripted.ScriptedMetricAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;

import java.io.IOException;

public class EsStringAggregation {
    public static void main(String[] args) throws IOException {
        try (RestHighLevelClient client = new RestHighLevelClient(...)) { // 初始化你的客户端实例
            SearchRequest searchRequest = new SearchRequest("your_index");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            sourceBuilder.size(0);

            // 构建scripted_metric聚合
            ScriptedMetricAggregationBuilder mergedAddressAgg = AggregationBuilders.scriptedMetric("merged_address")
                    .initScript(new Script(ScriptType.INLINE, "painless", "state.addresses = []", null))
                    .mapScript(new Script(ScriptType.INLINE, "painless", "state.addresses.add(doc['address'].value)", null))
                    .combineScript(new Script(ScriptType.INLINE, "painless", "return state.addresses.join(', ')", null))
                    .reduceScript(new Script(ScriptType.INLINE, "painless", "return states.join(', ')", null));

            sourceBuilder.aggregation(mergedAddressAgg);
            searchRequest.source(sourceBuilder);

            // 执行请求并解析结果
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
            String mergedAddress = (String) response.getAggregations().get("merged_address").getProperty("value");
            System.out.println("合并后的地址：" + mergedAddress);
        }
    }
}

注意事项： #

1. 确保address字段是text或keyword类型，能被脚本直接读取字段值
2. 脚本使用Painless语言，这是ES默认支持的安全脚本语言
3. 如果数据量极大，拼接后的字符串可能过长，需注意ES的字段长度限制与内存占用

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