OpenSearch优化查询：找出共享地址的客户（匹配数>1）

现有一个客户索引，其中addressList字段存储的是逗号分隔的地址字符串，示例数据如下：

id:c1, addressList:a1,a2,a3
id:c2, addressList:a4,a5,a6
id:c3, addressList:a1,a7
id:c4, addressList:a2,a5
id:c5, addressList:a8

标准化后的地址字符串格式示例：

1234_water_dr_suite_10_montgomery_ny_23134

需求是找出所有共享同一地址的客户，预期结果：

a1 - c1,c3
a2 - c1,c4
a5 - c2,c4 

当前采用的暴力解法：

1. 先通过以下查询获取所有地址并分词得到地址集合：

{
    "size": 10000,
    "_source": false,
    "fields": ["addressList"],
    "query": {
        "exists": {
            "field": "addressList"
        }
    }
}

这部分效率尚可，无需优化。

2. 遍历每个地址，逐个调用查询：

{
    "size": 10000,
    "query": {
        "match" : {
            "addressList": "<currAddress>"
        }
    } 
}

问题在于这种逐个查询的方式效率低下，希望能传入地址列表，直接返回匹配文档数大于1的地址对应的客户。由于addressList是逗号分隔的文本字段，无法设为term字段，因此无法直接使用带min_doc_count的聚合。

方案1：使用脚本字段+嵌套聚合（无需修改索引映射） #

可以通过脚本聚合一次性拆分地址、统计每个地址对应的客户，并过滤出被多个客户共享的地址，无需多次请求。

查询示例：

{
  "size": 0,
  "aggs": {
    "shared_addresses": {
      "scripted_metric": {
        "init_script": "state.address_map = new HashMap()",
        "map_script": """
          def addresses = doc['addressList'].value.split(',');
          def customer_id = doc['id'].value;
          for (def addr : addresses) {
            def trimmed_addr = addr.trim();
            if (!state.address_map.containsKey(trimmed_addr)) {
              state.address_map.put(trimmed_addr, new ArrayList());
            }
            state.address_map.get(trimmed_addr).add(customer_id);
          }
        """,
        "combine_script": """
          def filtered = new HashMap();
          for (def entry : state.address_map.entrySet()) {
            if (entry.getValue().size() > 1) {
              filtered.put(entry.getKey(), entry.getValue());
            }
          }
          return filtered;
        """,
        "reduce_script": """
          def final_result = new HashMap();
          for (def map : states) {
            for (def entry : map.entrySet()) {
              if (!final_result.containsKey(entry.getKey())) {
                final_result.put(entry.getKey(), new ArrayList());
              }
              final_result.get(entry.getKey()).addAll(entry.getValue());
            }
          }
          // 对客户ID去重，避免分片数据重复
          def deduplicated = new HashMap();
          for (def entry : final_result.entrySet()) {
            deduplicated.put(entry.getKey(), entry.getValue().stream().distinct().collect(Collectors.toList()));
          }
          return deduplicated;
        """
      }
    }
  }
}

查询返回结果格式示例：

{
  "aggregations": {
    "shared_addresses": {
      "value": {
        "a1": ["c1", "c3"],
        "a2": ["c1", "c4"],
        "a5": ["c2", "c4"]
      }
    }
  }
}

方案2：修改索引映射（长期最优解） #

如果允许修改索引映射，建议将addressList改为keyword类型的数组字段，这样可以直接使用标准聚合实现需求，性能更稳定高效。

1. 更新字段映射 #

PUT /customers/_mapping
{
  "properties": {
    "addressList": {
      "type": "keyword",
      "fields": {
        "text": {
          "type": "text"
        }
      }
    }
  }
}

（保留text子字段是为了兼容原有的文本查询需求，若不需要可省略）

2. 重新索引数据 #

将原逗号分隔的字符串转换为数组格式，比如把"a1,a2,a3"转为["a1","a2","a3"]。

3. 执行聚合查询 #

{
  "size": 0,
  "aggs": {
    "address_groups": {
      "terms": {
        "field": "addressList",
        "min_doc_count": 2
      },
      "aggs": {
        "customer_ids": {
          "top_hits": {
            "size": 10000,
            "_source": false,
            "fields": ["id"]
          }
        }
      }
    }
  }
}

查询会直接返回每个共享地址对应的客户ID列表，只需在客户端整理成预期格式即可。

方案3：批量查询减少请求次数 #

如果暂时无法修改映射或使用脚本聚合，可以将地址列表分批，通过bool查询的should子句批量查询，减少HTTP请求次数。

示例（批量查询a1、a2、a5）：

{
  "size": 10000,
  "_source": ["id", "addressList"],
  "query": {
    "bool": {
      "should": [
        {"match": {"addressList": "a1"}},
        {"match": {"addressList": "a2"}},
        {"match": {"addressList": "a5"}}
      ],
      "minimum_should_match": 1
    }
  }
}

拿到结果后，在客户端自行整理每个地址对应的客户ID列表，过滤出客户数大于1的地址即可。

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