es 如何查询数据并去重

Question

假如我有一个电商系统，我使用的时候mysql+es 进行数据存储，mysql 存储商品的简单信息

字段名	字段描述	备注
productId	商品Id
productName	商品名称

然后在es 里面，我存储了相关的词条，进行商品的检索，一个商品下可能存在多个词条

[{
  productId:'1',
  termItem:'霸王洗发水防脱发'
},
{
  productId:'1',
  termItem:'霸王洗发水防白发'
}
]

我现在想通过词条把商品查出来，然后将商品分页展示在前端，但是我遇到一个问题，因为词条查商品，是需要去重的，但是es 似乎要高版本才有collapse，且该字段必须是keyword，不能是text

Answer 1

阿里嘎多学长整理AIGC生成，因移动端显示问题导致当前答案未能完全显示，请使用PC端查看更加详细的解答过程

解决方案

你想使用 Elasticsearch 查询数据并去重，具体来说是从 MySQL 中获取商品信息，然后将其存储到 Elasticsearch 中，并使用 Elasticsearch 的去重功能来删除重复的数据。

首先，你需要使用 MySQL 连接到你的数据库，然后使用 SQL 语句来获取商品信息：

SELECT * FROM products;

然后，你可以使用 Elasticsearch 的 bulk API 将数据批量导入到 Elasticsearch 中：

// 使用 Java 语言
String[] columns = {"productId", "productName"};
List<Map<String, Object>> data = new ArrayList<>();
// 从 MySQL 中获取数据
try (Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/mydb", "username", "password");
     Statement stmt = conn.createStatement()) {
    ResultSet rs = stmt.executeQuery("SELECT * FROM products");
    while (rs.next()) {
        Map<String, Object> row = new HashMap<>();
        row.put("productId", rs.getLong("productId"));
        row.put("productName", rs.getString("productName"));
        data.add(row);
    }
}

// 使用 Elasticsearch 的 bulk API 将数据批量导入
XContentBuilder builder = new XContentBuilder();
builder.startObject();
for (Map<String, Object> row : data) {
    builder.startObject();
    builder.field("productId", row.get("productId"));
    builder.field("productName", row.get("productName"));
    builder.endObject();
}
builder.endObject();

String bulkIndexRequest = builder.toString();
RestClient client = RestClient.builder(new HttpHost("localhost", 9200, "http")).build();
Response response = client.performRequest("POST", "/_bulk", bulkIndexRequest);

在上面的代码中，我们使用 Java 语言连接到 MySQL 数据库，然后使用 SQL 语句获取商品信息。然后，我们使用 Elasticsearch 的 bulk API 将数据批量导入到 Elasticsearch 中。

如果你想使用 Elasticsearch 的去重功能，可以使用 terms aggregation：

// 使用 Java 语言
SearchRequest request = new SearchRequest("products");
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.query(QueryBuilders.matchAllQuery());
sourceBuilder.aggregation(AggregationBuilders.terms("uniqueProducts").field("productId"));
request.source(sourceBuilder);

SearchResponse response = client.search(request);

在上面的代码中，我们使用 Elasticsearch 的 terms aggregation 来将数据分组，并使用 uniqueProducts 作为聚合名称。然后，我们可以使用 getAggregations() 方法来获取聚合结果，并使用 getBuckets() 方法来获取每个 bucket 中的数据。

这样，你就可以使用 Elasticsearch 查询数据并去重了。