有一个es商品索引，分别有商品id productId,商家id userId,createTime,商品分类category字段,每个商家可以有多个字段，请帮我用java写段代码，使用聚合查询查出每个商家最新的3个不重复的商品分类

在Elasticsearch中，我们可以使用Java的elasticsearch客户端库来执行聚合查询。首先，你需要添加依赖到你的项目中。这里是一个简单的示例代码片段：

import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchAssistant;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.sort.SortAggregation;

public class ElasticsearchQueryExample {

    private static final String INDEX_NAME = "your_index_name";
    private static final Client client = // 初始化你的Elasticsearch客户端

    public void getCategoryTopThreePerUser(String userId) throws Exception {
        SearchRequest searchRequest = new SearchRequest(INDEX_NAME);
        
        // 创建聚合，按商家ID分组并获取最新创建时间的商品
        SortAggregation sortAgg = AggregationBuilders
                .terms("user_categories")
                .field("userId.keyword") // 使用keyword类型处理字符串防止分词影响
                .subAggregation(AggregationBuilders
                        .topHits("latest_products")
                        .sort(SortBuilders.fieldSort("createTime").desc())
                        .size(3)) // 获取每个商家最近的三个商品

        // 添加聚合到搜索请求
        searchRequest.aggregations(sortAgg);

        // 查询条件：商家ID等于指定用户ID
        QueryBuilders boolQuery = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("userId", userId));

        searchRequest.query(boolQuery);

        // 执行搜索请求
        SearchAssistant response = client.prepareSearch(searchRequest).get();

        if (response.isSucceeded()) {
            System.out.println("Top three categories for user " + userId + ":");
            TopHits hits = response.getAggregations().get("user_categories");
            for (Terms.Bucket bucket : hits.getBuckets()) {
                String categoryName = bucket.getKeyAsString(); // 商家ID对应的类别名称
                int count = bucket.getDocCount(); // 最新的商品数量
                List<SearchHit> latestProducts = bucket.getAggregations().get("latest_products").getHits().getHits();
                for (SearchHit latestProduct : latestProducts) {
                    String productId = latestProduct.getId(); // 最新的商品ID
                    // 输出结果
                    System.out.println("Category: " + categoryName + ", Product ID: " + productId);
                }
            }
        } else {
            System.err.println("Failed to execute query: " + response.toString());
        }
    }
}