1.项目地址

https://github.com/GuardFTC/elasticsearch-test.git

2.创建实体类

  1. import lombok.Data;
  3. import java.util.Date;
  5. /**
  6.  * @author: 冯铁城 [17615007230@163.com]
  7.  * @date: 2022-08-24 19:42:52
  8.  * @describe: 汽车订单
  9.  */
  10. @Data
  11. public class CarOrder {
  13.     /**
  14.      * 订单汽车售价
  15.      */
  16.     private double price;
  18.     /**
  19.      * 订单汽车颜色
  20.      */
  21.     private String color;
  23.     /**
  24.      * 订单汽车厂商
  25.      */
  26.     private String make;
  28.     /**
  29.      * 订单售卖时间
  30.      */
  31.     private Date soldDate;
  32. }
  1. import lombok.Data;
  3. import java.util.Date;
  5. /**
  6.  * @author: 冯铁城 [17615007230@163.com]
  7.  * @date: 2022-08-28 18:05:27
  8.  * @describe: 网站请求数据
  9.  */
  10. @Data
  11. public class WebsiteRequest {
  13.     /**
  14.      * 延迟 单位ms
  15.      */
  16.     private long latency;
  18.     /**
  19.      * 所在地区
  20.      */
  21.     private String zone;
  23.     /**
  24.      * 时间戳
  25.      */
  26.     private Date timestamp;
  27. }

3.聚合API命令整合

terms

在获取聚合结果时，提供了多种terms聚合，包括sterms，dterms等很多

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void terms() {
  5.     //1.根据颜色进行聚合
  6.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .aggregations("colorCars", a -> a.terms(t -> t
  9.                     .field("color"))
  10.             ), CarOrder.class
  11.     );
  13.     //2.校验
  14.     Aggregate colorCars = search.aggregations().get("colorCars");
  15.     Buckets<StringTermsBucket> buckets = colorCars.sterms().buckets();
  16.     Assert.isTrue(4 == buckets.array().size());
  17.     buckets.array().forEach(b -> {
  18.         Console.log(b.key());
  19.         Console.log(b.docCount());
  20.     });
  21. }
  23. @Test
  24. @SneakyThrows(IOException.class)
  25. void nested() {
  27.     //1.同级嵌套，统计每种颜色，每种厂商的汽车销量
  28.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  29.             .index(INDEX_NAME)
  30.             .aggregations("colorOrders", a -> a
  31.                     .terms(t -> t.field("color")
  32.                     ))
  33.             .aggregations("makeOrders", a -> a
  34.                     .terms(t -> t.field("make"))
  35.             ), CarOrder.class
  36.     );
  38.     //2.校验
  39.     Aggregate colorOrders = search.aggregations().get("colorOrders");
  40.     List<StringTermsBucket> colorBuckets = colorOrders.sterms().buckets().array();
  41.     Assert.isTrue(4 == colorBuckets.size());
  42.     Aggregate makeOrders = search.aggregations().get("makeOrders");
  43.     List<StringTermsBucket> makeBuckets = makeOrders.sterms().buckets().array();
  44.     Assert.isTrue(4 == makeBuckets.size());
  46.     //3.递归嵌套，统计取不同颜色汽车的销量，同时统计每种颜色汽车中每个制造商的销量以及每个制造商的总销售额
  47.     search = primaryClient.search(s -> s
  48.             .index(INDEX_NAME)
  49.             .aggregations("colorOrders", a -> a
  50.                     .terms(t -> t.field("color"))
  51.                     .aggregations("makeOrders", colorA -> colorA.terms(
  52.                                     t -> t.field("make"))
  53.                             .aggregations("makeTotalPrice", makeA -> makeA.sum(
  54.                                     sum -> sum.field("price")
  55.                             ))
  56.                     )
  57.             ), CarOrder.class
  58.     );
  60.     //4.校验
  61.     colorOrders = search.aggregations().get("colorOrders");
  62.     colorBuckets = colorOrders.sterms().buckets().array();
  63.     Assert.isTrue(4 == colorBuckets.size());
  64.     Console.log("按照颜色聚合-----");
  65.     for (StringTermsBucket colorBucket : colorBuckets) {
  66.         Console.log("颜色:" + colorBucket.key() + " 数量:" + colorBucket.docCount());
  67.         Console.log("按照厂商聚合-----");
  68.         makeOrders = colorBucket.aggregations().get("makeOrders");
  69.         makeBuckets = makeOrders.sterms().buckets().array();
  70.         for (StringTermsBucket makeBucket : makeBuckets) {
  71.             Console.log("厂商:" + makeBucket.key() + " 数量:" + makeBucket.docCount());
  72.             Aggregate makeTotalPrice = makeBucket.aggregations().get("makeTotalPrice");
  73.             SumAggregate sum = makeTotalPrice.sum();
  74.             Console.log("厂商销售总额:" + sum.value());
  75.         }
  76.     }
  77. }

histogram

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void histogram() {
  5.     //1.按照销售额步长为20000统计不同颜色汽车的销量
  6.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .aggregations("priceCarOrders", a -> a
  9.                     .histogram(h -> h
  10.                             .field("price")
  11.                             .interval((double) 20000)
  12.                     )
  13.                     .aggregations("colorOrders", ha -> ha
  14.                             .terms(t -> t.field("color"))
  15.                     )
  16.             ), CarOrder.class
  17.     );
  19.     //2.校验
  20.     Aggregate priceCarOrders = search.aggregations().get("priceCarOrders");
  21.     List<HistogramBucket> priceCarOrdersBuckets = priceCarOrders.histogram().buckets().array();
  22.     Assert.isTrue(7 == priceCarOrdersBuckets.size());
  23.     Console.log("按照20000销售额价格聚合-----");
  24.     for (HistogramBucket priceCarOrdersBucket : priceCarOrdersBuckets) {
  25.         Console.log("[" + priceCarOrdersBucket.key() + "-" + (priceCarOrdersBucket.key() + (double) 20000)
  26.                 + "]区间内销售量:" + priceCarOrdersBucket.docCount()
  27.         );
  28.         Aggregate colorOrders = priceCarOrdersBucket.aggregations().get("colorOrders");
  29.         List<StringTermsBucket> colorBuckets = colorOrders.sterms().buckets().array();
  30.         for (StringTermsBucket colorBucket : colorBuckets) {
  31.             Console.log("颜色:" + colorBucket.key() + " 数量:" + colorBucket.docCount());
  32.         }
  33.     }
  34. }

dateHistogram

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void DateHistogram() {
  5.     //1.每个季度，不同汽车品牌的销售总额
  6.     SearchResponse<CarOrder> search = secondaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .aggregations("quarterCarOrders", a -> a
  9.                     .dateHistogram(d -> d
  10.                             .field("soldDate")
  11.                             .calendarInterval(CalendarInterval.Quarter)
  12.                             .format(DatePattern.NORM_DATE_PATTERN)
  13.                     )
  14.                     .aggregations("makeOrders", da -> da
  15.                             .terms(t -> t.field("make"))
  16.                             .aggregations("totalPrice", ta -> ta
  17.                                     .sum(sum -> sum.field("price"))
  18.                             )
  19.                     )
  20.             ), CarOrder.class
  21.     );
  23.     //2.校验
  24.     Aggregate quarterCarOrders = search.aggregations().get("quarterCarOrders");
  25.     List<DateHistogramBucket> dateHistogramBuckets = quarterCarOrders.dateHistogram().buckets().array();
  26.     Assert.isTrue(5 == dateHistogramBuckets.size());
  27.     Console.log("按照季度聚合-----");
  28.     for (DateHistogramBucket dateHistogramBucket : dateHistogramBuckets) {
  29.         Console.log(dateHistogramBucket.keyAsString() + "季度售卖量" + dateHistogramBucket.docCount());
  30.         Aggregate makeOrders = dateHistogramBucket.aggregations().get("makeOrders");
  31.         List<StringTermsBucket> makeBuckets = makeOrders.sterms().buckets().array();
  32.         for (StringTermsBucket makeBucket : makeBuckets) {
  33.             Console.log("厂商" + makeBucket.key() + "售卖" + makeBucket.docCount() + "台");
  34.             Aggregate totalPrice = makeBucket.aggregations().get("totalPrice");
  35.             Console.log("共盈利" + totalPrice.sum().value());
  36.         }
  37.     }
  38. }

查询结果和聚合结果都过滤

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void searchAndAggregationsFilter() {
  5.     //1.获取福特汽车均价以及所有汽车均价，看看福特汽车的价格是否合理
  6.     SearchResponse<CarOrder> search = secondaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .query(q -> q.constantScore(c -> c.filter(f -> f.term(t -> t
  9.                     .field("make")
  10.                     .value(FieldValue.of("ford"))
  11.             ))))
  12.             .aggregations("fortAvePrice", a -> a
  13.                     .avg(avg -> avg.field("price"))
  14.             )
  15.             .aggregations("totalMake", a -> a
  16.                     .global(g -> g)
  17.                     .aggregations("totalAvgPrice", ta -> ta.
  18.                             avg(avg -> avg.field("price"))
  19.                     )
  20.             ), CarOrder.class
  21.     );
  23.     //2.校验
  24.     Aggregate fortAvePrice = search.aggregations().get("fortAvePrice");
  25.     Aggregate totalMake = search.aggregations().get("totalMake");
  26.     Assert.isTrue(4 == search.hits().hits().size());
  27.     Console.log("福特汽车均价:" + fortAvePrice.avg().value());
  28.     Console.log("所有汽车均价:" + totalMake.global().aggregations().get("totalAvgPrice").avg().value());
  29. }

查询结果不过滤，聚合结果过滤

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void searchNotFilterButAggregationsFilter() {
  5.     //1.查询所有福特汽车订单，统计福特汽车中黄色汽车的销售总额
  6.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .query(q -> q.constantScore(c -> c.filter(f -> f.term(t -> t
  9.                     .field("make")
  10.                     .value(FieldValue.of("ford"))
  11.             ))))
  12.             .aggregations("yellowColorCarOrders", a -> a
  13.                     .filter(f -> f.constantScore(c -> c.filter(cf -> cf.term(t -> t
  14.                             .field("color")
  15.                             .value(FieldValue.of("yellow"))
  16.                     ))))
  17.                     .aggregations("totalPrice", ta -> ta.sum(sum -> sum
  18.                             .field("price")
  19.                     ))
  20.             ), CarOrder.class
  21.     );
  23.     //2.校验
  24.     Assert.isTrue(4 == search.hits().hits().size());
  25.     Aggregate yellowColorCarOrders = search.aggregations().get("yellowColorCarOrders");
  26.     Assert.isTrue(2 == yellowColorCarOrders.filter().docCount());
  27.     Aggregate totalPrice = yellowColorCarOrders.filter().aggregations().get("totalPrice");
  28.     Console.log("福特黄色汽车的销售总额:" + totalPrice.sum().value());
  29. }

查询结果过滤，局和结果不过滤

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void searchFilterButAggregationsNotFilter() {
  5.     //1.查询展示蓝色福特汽车，同时统计不同颜色汽车的销售均价
  6.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .aggregations("colorCarOrders", a -> a.terms(t -> t
  9.                             .field("color"))
  10.                     .aggregations("avgPrice", fa -> fa.avg(avg -> avg
  11.                             .field("price")
  12.                     ))
  13.             )
  14.             .postFilter(pf -> pf.bool(b -> b
  15.                     .must(m -> m.term(t -> t
  16.                             .field("color")
  17.                             .value(FieldValue.of("blue"))
  18.                     ))
  19.                     .must(m -> m.term(t -> t
  20.                             .field("make")
  21.                             .value(FieldValue.of("ford"))
  22.                     )))
  23.             ), CarOrder.class
  24.     );
  26.     //2.校验
  27.     Assert.isTrue(1 == search.hits().hits().size());
  28.     Aggregate colorCarOrders = search.aggregations().get("colorCarOrders");
  29.     List<StringTermsBucket> colorBuckets = colorCarOrders.sterms().buckets().array();
  30.     Assert.isTrue(4 == colorBuckets.size());
  32.     for (StringTermsBucket colorBucket : colorBuckets) {
  33.         Aggregate avePrice = colorBucket.aggregations().get("avgPrice");
  34.         Console.log(colorBucket.key() + "汽车销售均价" + avePrice.avg().value());
  35.     }
  36. }

sort

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void sort() {
  5.     //1.同级排序-统计所有厂商的销售总额，并明确了解谁的销售总额最高
  6.     HashMap<String, SortOrder> sortMaps = MapUtil.newHashMap(1);
  7.     sortMaps.put("totalPrice", SortOrder.Desc);
  8.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  9.             .index(INDEX_NAME)
  10.             .aggregations("makeCarOrders", a -> a.terms(t -> t
  11.                                     .field("make")
  12.                                     .order(sortMaps)
  13.                             )
  14.                             .aggregations("totalPrice", fa -> fa.sum(sum -> sum
  15.                                     .field("price")
  16.                             ))
  17.             ), CarOrder.class
  18.     );
  20.     //2.校验
  21.     Aggregate makeCarOrders = search.aggregations().get("makeCarOrders");
  22.     List<StringTermsBucket> makeBuckets = makeCarOrders.sterms().buckets().array();
  23.     for (StringTermsBucket makeBucket : makeBuckets) {
  24.         Aggregate totalPrice = makeBucket.aggregations().get("totalPrice");
  25.         Console.log(makeBucket.key() + "汽车销售总额" + totalPrice.sum().value());
  26.     }
  27. }
  1.     @Test
  2.     @SneakyThrows(IOException.class)
  3.     void sortDeep() {
  5.         //1.统计每个季度，每个厂商黄色汽车销售总额，并且直观看到黄色汽车销售总额高的厂商
  6.         HashMap<String, SortOrder> sortMaps = MapUtil.newHashMap(1);
  7.         sortMaps.put("totalPrice>yellowTotalPrice", SortOrder.Desc);
  8.         SearchResponse<CarOrder> search = secondaryClient.search(s -> s
  9.                 .index(INDEX_NAME)
  10.                 .aggregations("quarterOrders", a -> a
  11.                         .dateHistogram(d -> d
  12.                                 .field("soldDate")
  13.                                 .calendarInterval(CalendarInterval.Quarter)
  14.                                 .format(DatePattern.NORM_DATE_PATTERN)
  15.                         )
  16.                         .aggregations("makeOrders", da -> da
  17.                                 .terms(t -> t
  18.                                         .field("make")
  19.                                         .order(sortMaps)
  20.                                 )
  21.                                 .aggregations("totalPrice", daa -> daa
  22.                                         .filter(f -> f.term(t -> t
  23.                                                 .field("color")
  24.                                                 .value(FieldValue.of("yellow"))
  25.                                         ))
  26.                                         .aggregations("yellowTotalPrice", daaa -> daaa.sum(sum -> sum
  27.                                                 .field("price")
  28.                                         ))
  29.                                 )
  30.                         )
  31.                 ), CarOrder.class
  32.         );
  34.         //2.校验
  35.         Aggregate quarterOrders = search.aggregations().get("quarterOrders");
  36.         List<DateHistogramBucket> histogramBuckets = quarterOrders.dateHistogram().buckets().array();
  37.         Assert.isTrue(5 == histogramBuckets.size());
  38.         for (DateHistogramBucket histogramBucket : histogramBuckets) {
  39.             Console.log(histogramBucket.keyAsString() + "-季度共卖出" + histogramBucket.docCount() + "台汽车");
  40.             Console.log("其中黄色汽车订单销售额如下:");
  41.             Aggregate makeOrders = histogramBucket.aggregations().get("makeOrders");
  42.             List<StringTermsBucket> makeBuckets = makeOrders.sterms().buckets().array();
  43.             for (StringTermsBucket makeBucket : makeBuckets) {
  44.                 Aggregate totalPrice = makeBucket.aggregations().get("totalPrice");
  45.                 Aggregate yellowTotalPrice = totalPrice.filter().aggregations().get("yellowTotalPrice");
  46.                 Console.log(makeBucket.key() + "共卖出黄色汽车总价:" + yellowTotalPrice.sum().value());
  47.             }
  48.         }
  49.     }

cardinality

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void cardinality() {
  5.     //1.统计共有多少个厂商的汽车订单
  6.     SearchResponse<CarOrder> search = primaryClient.search(s -> s
  7.             .index(INDEX_NAME)
  8.             .aggregations("makeCount", a -> a.cardinality(c -> c
  9.                     .field("make")
  10.                     .precisionThreshold(100)
  11.             )), CarOrder.class
  12.     );
  14.     //2.校验
  15.     Aggregate makeCount = search.aggregations().get("makeCount");
  16.     Assert.isTrue(4 == makeCount.cardinality().value());
  17. }

percentiles

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void percentiles() {
  5.     //1.存储网站请求数据
  6.     saveWebSite();
  8.     //2.统计不同区域网站请求的平均值，通过百分数进行异常值分析，查看哪个区域网络延迟较高
  9.     SearchResponse<WebsiteRequest> search = primaryClient.search(s -> s
  10.             .index(WEBSITE_INDEX)
  11.             .aggregations("zoneWebRequest", a -> a
  12.                     .terms(t -> t.field("zone"))
  13.                     .aggregations("avgLatency", ta -> ta.avg(avg -> avg
  14.                             .field("latency"))
  15.                     )
  16.                     .aggregations("percentilesLatency", ta -> ta.percentiles(p -> p
  17.                             .field("latency")
  18.                             .percents((double) 5, (double) 25, (double) 50, (double) 75, (double) 100)
  19.                     ))
  20.             ), WebsiteRequest.class
  21.     );
  23.     //3.校验
  24.     Aggregate zoneWebRequest = search.aggregations().get("zoneWebRequest");
  25.     List<StringTermsBucket> zoneBuckets = zoneWebRequest.sterms().buckets().array();
  26.     for (StringTermsBucket zoneBucket : zoneBuckets) {
  27.         String key = zoneBucket.key();
  28.         double avgLatency = zoneBucket.aggregations().get("avgLatency").avg().value();
  29.         Console.log(key + "地区 网站平均响应延迟:" + avgLatency);
  31.         TDigestPercentilesAggregate percentilesLatency = zoneBucket.aggregations().get("percentilesLatency").tdigestPercentiles();
  32.         Percentiles values = percentilesLatency.values();
  33.         for (String percentilesItem : values.keyed().keySet()) {
  34.             Console.log(key + "地区" + percentilesItem + "%的请求 响应延迟:" + values.keyed().get(percentilesItem));
  35.         }
  36.     }
  37. }

percentileRanks

  1. @Test
  2. @SneakyThrows(IOException.class)
  3. void percentileRanks() {
  5.     //1.存储网站请求数据
  6.     saveWebSite();
  8.     //2.获取美国地区，网络延迟达到30,80,100的用户百分比
  9.     SearchResponse<WebsiteRequest> search = secondaryClient.search(s -> s
  10.             .index(WEBSITE_INDEX)
  11.             .aggregations("usWebRequests", a -> a
  12.                     .filter(f -> f.term(t -> t
  13.                             .field("zone")
  14.                             .value(FieldValue.of("US"))
  15.                     ))
  16.                     .aggregations("usPercentileRanks", fa -> fa.percentileRanks(p -> p
  17.                             .field("latency")
  18.                             .values((double) 30, (double) 80, (double) 100)
  19.                     ))
  20.             ), WebsiteRequest.class
  21.     );
  23.     //3.校验
  24.     Aggregate usWebRequests = search.aggregations().get("usWebRequests");
  25.     Aggregate usPercentileRanks = usWebRequests.filter().aggregations().get("usPercentileRanks");
  26.     TDigestPercentileRanksAggregate tDigestPercentileRanks = usPercentileRanks.tdigestPercentileRanks();
  27.     Map<String, String> keyed = tDigestPercentileRanks.values().keyed();
  28.     Assert.isTrue(3 == keyed.size());
  29.     for (String key : keyed.keySet()) {
  30.         Console.log(keyed.get(key) + "%的用户 网络延迟请求达到了" + key + "毫秒");
  31.     }
  32. }

4.完整版单元测试

  1. import cn.hutool.core.date.DatePattern;
  2. import cn.hutool.core.lang.Assert;
  3. import cn.hutool.core.lang.Console;
  4. import cn.hutool.core.map.MapUtil;
  5. import cn.hutool.core.util.ObjectUtil;
  6. import cn.hutool.json.JSONUtil;
  7. import co.elastic.clients.elasticsearch.ElasticsearchClient;
  8. import co.elastic.clients.elasticsearch._types.FieldValue;
  9. import co.elastic.clients.elasticsearch._types.Refresh;
  10. import co.elastic.clients.elasticsearch._types.SortOrder;
  11. import co.elastic.clients.elasticsearch._types.aggregations.*;
  12. import co.elastic.clients.elasticsearch.core.BulkRequest;
  13. import co.elastic.clients.elasticsearch.core.BulkResponse;
  14. import co.elastic.clients.elasticsearch.core.DeleteByQueryResponse;
  15. import co.elastic.clients.elasticsearch.core.SearchResponse;
  16. import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
  17. import co.elastic.clients.transport.endpoints.BooleanResponse;
  18. import com.ftc.elasticsearchtest.entity.CarOrder;
  19. import com.ftc.elasticsearchtest.entity.WebsiteRequest;
  20. import lombok.SneakyThrows;
  21. import org.junit.jupiter.api.BeforeEach;
  22. import org.junit.jupiter.api.Test;
  23. import org.springframework.beans.factory.annotation.Qualifier;
  24. import org.springframework.boot.test.context.SpringBootTest;
  26. import javax.annotation.Resource;
  27. import java.io.IOException;
  28. import java.util.HashMap;
  29. import java.util.List;
  30. import java.util.Map;
  32. /**
  33.  * @author: 冯铁城 [17615007230@163.com]
  34.  * @date: 2022-08-24 19:35:36
  35.  * @describe: ElasticSearch聚合操作Api
  36.  */
  37. @SpringBootTest
  38. public class ElasticSearchAggregationTest {
  40.     @Resource
  41.     @Qualifier("primaryElasticsearchClient")
  42.     private ElasticsearchClient primaryClient;
  44.     @Resource
  45.     @Qualifier("secondaryElasticsearchClient")
  46.     private ElasticsearchClient secondaryClient;
  48.     /**
  49.      * 测试Mock索引名称常量
  50.      */
  51.     private static final String INDEX_NAME = "test_car_order";
  53.     /**
  54.      * 网站索引名称
  55.      */
  56.     private static final String WEBSITE_INDEX = "website";
  58.     /**
  59.      * 测试数据
  60.      */
  61.     private static final List<CarOrder> CAR_ORDER_LIST;
  63.     /*
  64.       加载测试数据
  65.      */
  66.     static {
  67.         String testData = "[{\"price\":10000,\"color\":\"red\",\"make\":\"honda\",\"soldDate\":\"2014-10-28\"},{\"price\":20000,\"color\":\"red\",\"make\":\"honda\",\"soldDate\":\"2014-11-05\"},{\"price\":30000,\"color\":\"green\",\"make\":\"ford\",\"soldDate\":\"2014-05-18\"},{\"price\":25000,\"color\":\"blue\",\"make\":\"ford\",\"soldDate\":\"2014-02-12\"},{\"price\":120000,\"color\":\"yellow\",\"make\":\"ford\",\"soldDate\":\"2014-06-11\"},{\"price\":90000,\"color\":\"yellow\",\"make\":\"ford\",\"soldDate\":\"2014-03-03\"},{\"price\":15000,\"color\":\"blue\",\"make\":\"toyota\",\"soldDate\":\"2014-07-02\"},{\"price\":12000,\"color\":\"green\",\"make\":\"toyota\",\"soldDate\":\"2014-08-19\"},{\"price\":80000,\"color\":\"red\",\"make\":\"bmw\",\"soldDate\":\"2014-01-01\"}]";
  68.         CAR_ORDER_LIST = JSONUtil.toList(JSONUtil.parseArray(testData), CarOrder.class);
  69.     }
  71.     @BeforeEach
  72.     void saveTestData() throws IOException {
  74.         //1.判定索引是否存在
  75.         boolean exist = primaryClient.indices().exists(e -> e.index(INDEX_NAME)).value();
  77.         //2.不存在创建索引
  78.         if (!exist) {
  80.             //3.创建索引
  81.             CreateIndexResponse createIndexResponse = primaryClient.indices().create(c -> c
  82.                     .index(INDEX_NAME + "_1")
  83.                     .aliases(INDEX_NAME, a -> a)
  84.                     .mappings(m -> m
  85.                             .properties("price", p -> p.double_(d -> d))
  86.                             .properties("color", p -> p.keyword(k -> k))
  87.                             .properties("make", p -> p.keyword(k -> k))
  88.                             .properties("soldDate", p -> p.date(d -> d))
  89.                     )
  90.                     .settings(s -> s
  91.                             .refreshInterval(r -> r.time("1s"))
  92.                             .numberOfShards("3")
  93.                             .numberOfReplicas("1")
  94.                     )
  95.             );
  96.             Assert.isTrue(Boolean.TRUE.equals(createIndexResponse.acknowledged()));
  97.         }
  99.         //4.删除数据
  100.         DeleteByQueryResponse deleteByQuery = primaryClient.deleteByQuery(d -> d
  101.                 .index(INDEX_NAME)
  102.                 .query(q -> q.matchAll(m -> m))
  103.                 .refresh(true)
  104.         );
  105.         Assert.isTrue(ObjectUtil.isNotNull(deleteByQuery.deleted()));
  107.         //5.批量存入数据
  108.         BulkRequest.Builder builder = new BulkRequest.Builder();
  109.         builder.index(INDEX_NAME);
  110.         CAR_ORDER_LIST.forEach(car -> builder.operations(o -> o.create(c -> c.document(car))));
  111.         builder.refresh(Refresh.True);
  112.         BulkRequest build = builder.build();
  114.         //6.存储数据
  115.         BulkResponse bulk = primaryClient.bulk(build);
  116.         Assert.isFalse(bulk.errors());
  117.     }
  119.     @Test
  120.     @SneakyThrows(IOException.class)
  121.     void terms() {
  123.         //1.根据颜色进行聚合
  124.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  125.                 .index(INDEX_NAME)
  126.                 .aggregations("colorCars", a -> a.terms(t -> t
  127.                         .field("color"))
  128.                 ), CarOrder.class
  129.         );
  131.         //2.校验
  132.         Aggregate colorCars = search.aggregations().get("colorCars");
  133.         Buckets<StringTermsBucket> buckets = colorCars.sterms().buckets();
  134.         Assert.isTrue(4 == buckets.array().size());
  135.         buckets.array().forEach(b -> {
  136.             Console.log(b.key());
  137.             Console.log(b.docCount());
  138.         });
  139.     }
  141.     @Test
  142.     @SneakyThrows(IOException.class)
  143.     void nested() {
  145.         //1.同级嵌套，统计每种颜色，每种厂商的汽车销量
  146.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  147.                 .index(INDEX_NAME)
  148.                 .aggregations("colorOrders", a -> a
  149.                         .terms(t -> t.field("color")
  150.                         ))
  151.                 .aggregations("makeOrders", a -> a
  152.                         .terms(t -> t.field("make"))
  153.                 ), CarOrder.class
  154.         );
  156.         //2.校验
  157.         Aggregate colorOrders = search.aggregations().get("colorOrders");
  158.         List<StringTermsBucket> colorBuckets = colorOrders.sterms().buckets().array();
  159.         Assert.isTrue(4 == colorBuckets.size());
  160.         Aggregate makeOrders = search.aggregations().get("makeOrders");
  161.         List<StringTermsBucket> makeBuckets = makeOrders.sterms().buckets().array();
  162.         Assert.isTrue(4 == makeBuckets.size());
  164.         //3.递归嵌套，统计取不同颜色汽车的销量，同时统计每种颜色汽车中每个制造商的销量以及每个制造商的总销售额
  165.         search = primaryClient.search(s -> s
  166.                 .index(INDEX_NAME)
  167.                 .aggregations("colorOrders", a -> a
  168.                         .terms(t -> t.field("color"))
  169.                         .aggregations("makeOrders", colorA -> colorA.terms(
  170.                                         t -> t.field("make"))
  171.                                 .aggregations("makeTotalPrice", makeA -> makeA.sum(
  172.                                         sum -> sum.field("price")
  173.                                 ))
  174.                         )
  175.                 ), CarOrder.class
  176.         );
  178.         //4.校验
  179.         colorOrders = search.aggregations().get("colorOrders");
  180.         colorBuckets = colorOrders.sterms().buckets().array();
  181.         Assert.isTrue(4 == colorBuckets.size());
  182.         Console.log("按照颜色聚合-----");
  183.         for (StringTermsBucket colorBucket : colorBuckets) {
  184.             Console.log("颜色:" + colorBucket.key() + " 数量:" + colorBucket.docCount());
  185.             Console.log("按照厂商聚合-----");
  186.             makeOrders = colorBucket.aggregations().get("makeOrders");
  187.             makeBuckets = makeOrders.sterms().buckets().array();
  188.             for (StringTermsBucket makeBucket : makeBuckets) {
  189.                 Console.log("厂商:" + makeBucket.key() + " 数量:" + makeBucket.docCount());
  190.                 Aggregate makeTotalPrice = makeBucket.aggregations().get("makeTotalPrice");
  191.                 SumAggregate sum = makeTotalPrice.sum();
  192.                 Console.log("厂商销售总额:" + sum.value());
  193.             }
  194.         }
  195.     }
  197.     @Test
  198.     @SneakyThrows(IOException.class)
  199.     void histogram() {
  201.         //1.按照销售额步长为20000统计不同颜色汽车的销量
  202.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  203.                 .index(INDEX_NAME)
  204.                 .aggregations("priceCarOrders", a -> a
  205.                         .histogram(h -> h
  206.                                 .field("price")
  207.                                 .interval((double) 20000)
  208.                         )
  209.                         .aggregations("colorOrders", ha -> ha
  210.                                 .terms(t -> t.field("color"))
  211.                         )
  212.                 ), CarOrder.class
  213.         );
  215.         //2.校验
  216.         Aggregate priceCarOrders = search.aggregations().get("priceCarOrders");
  217.         List<HistogramBucket> priceCarOrdersBuckets = priceCarOrders.histogram().buckets().array();
  218.         Assert.isTrue(7 == priceCarOrdersBuckets.size());
  219.         Console.log("按照20000销售额价格聚合-----");
  220.         for (HistogramBucket priceCarOrdersBucket : priceCarOrdersBuckets) {
  221.             Console.log("[" + priceCarOrdersBucket.key() + "-" + (priceCarOrdersBucket.key() + (double) 20000)
  222.                     + "]区间内销售量:" + priceCarOrdersBucket.docCount()
  223.             );
  224.             Aggregate colorOrders = priceCarOrdersBucket.aggregations().get("colorOrders");
  225.             List<StringTermsBucket> colorBuckets = colorOrders.sterms().buckets().array();
  226.             for (StringTermsBucket colorBucket : colorBuckets) {
  227.                 Console.log("颜色:" + colorBucket.key() + " 数量:" + colorBucket.docCount());
  228.             }
  229.         }
  230.     }
  232.     @Test
  233.     @SneakyThrows(IOException.class)
  234.     void DateHistogram() {
  236.         //1.每个季度，不同汽车品牌的销售总额
  237.         SearchResponse<CarOrder> search = secondaryClient.search(s -> s
  238.                 .index(INDEX_NAME)
  239.                 .aggregations("quarterCarOrders", a -> a
  240.                         .dateHistogram(d -> d
  241.                                 .field("soldDate")
  242.                                 .calendarInterval(CalendarInterval.Quarter)
  243.                                 .format(DatePattern.NORM_DATE_PATTERN)
  244.                         )
  245.                         .aggregations("makeOrders", da -> da
  246.                                 .terms(t -> t.field("make"))
  247.                                 .aggregations("totalPrice", ta -> ta
  248.                                         .sum(sum -> sum.field("price"))
  249.                                 )
  250.                         )
  251.                 ), CarOrder.class
  252.         );
  254.         //2.校验
  255.         Aggregate quarterCarOrders = search.aggregations().get("quarterCarOrders");
  256.         List<DateHistogramBucket> dateHistogramBuckets = quarterCarOrders.dateHistogram().buckets().array();
  257.         Assert.isTrue(5 == dateHistogramBuckets.size());
  258.         Console.log("按照季度聚合-----");
  259.         for (DateHistogramBucket dateHistogramBucket : dateHistogramBuckets) {
  260.             Console.log(dateHistogramBucket.keyAsString() + "季度售卖量" + dateHistogramBucket.docCount());
  261.             Aggregate makeOrders = dateHistogramBucket.aggregations().get("makeOrders");
  262.             List<StringTermsBucket> makeBuckets = makeOrders.sterms().buckets().array();
  263.             for (StringTermsBucket makeBucket : makeBuckets) {
  264.                 Console.log("厂商" + makeBucket.key() + "售卖" + makeBucket.docCount() + "台");
  265.                 Aggregate totalPrice = makeBucket.aggregations().get("totalPrice");
  266.                 Console.log("共盈利" + totalPrice.sum().value());
  267.             }
  268.         }
  269.     }
  271.     @Test
  272.     @SneakyThrows(IOException.class)
  273.     void searchAndAggregationsFilter() {
  275.         //1.获取福特汽车均价以及所有汽车均价，看看福特汽车的价格是否合理
  276.         SearchResponse<CarOrder> search = secondaryClient.search(s -> s
  277.                 .index(INDEX_NAME)
  278.                 .query(q -> q.constantScore(c -> c.filter(f -> f.term(t -> t
  279.                         .field("make")
  280.                         .value(FieldValue.of("ford"))
  281.                 ))))
  282.                 .aggregations("fortAvePrice", a -> a
  283.                         .avg(avg -> avg.field("price"))
  284.                 )
  285.                 .aggregations("totalMake", a -> a
  286.                         .global(g -> g)
  287.                         .aggregations("totalAvgPrice", ta -> ta.
  288.                                 avg(avg -> avg.field("price"))
  289.                         )
  290.                 ), CarOrder.class
  291.         );
  293.         //2.校验
  294.         Aggregate fortAvePrice = search.aggregations().get("fortAvePrice");
  295.         Aggregate totalMake = search.aggregations().get("totalMake");
  296.         Assert.isTrue(4 == search.hits().hits().size());
  297.         Console.log("福特汽车均价:" + fortAvePrice.avg().value());
  298.         Console.log("所有汽车均价:" + totalMake.global().aggregations().get("totalAvgPrice").avg().value());
  299.     }
  301.     @Test
  302.     @SneakyThrows(IOException.class)
  303.     void searchNotFilterButAggregationsFilter() {
  305.         //1.查询所有福特汽车订单，统计福特汽车中黄色汽车的销售总额
  306.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  307.                 .index(INDEX_NAME)
  308.                 .query(q -> q.constantScore(c -> c.filter(f -> f.term(t -> t
  309.                         .field("make")
  310.                         .value(FieldValue.of("ford"))
  311.                 ))))
  312.                 .aggregations("yellowColorCarOrders", a -> a
  313.                         .filter(f -> f.constantScore(c -> c.filter(cf -> cf.term(t -> t
  314.                                 .field("color")
  315.                                 .value(FieldValue.of("yellow"))
  316.                         ))))
  317.                         .aggregations("totalPrice", ta -> ta.sum(sum -> sum
  318.                                 .field("price")
  319.                         ))
  320.                 ), CarOrder.class
  321.         );
  323.         //2.校验
  324.         Assert.isTrue(4 == search.hits().hits().size());
  325.         Aggregate yellowColorCarOrders = search.aggregations().get("yellowColorCarOrders");
  326.         Assert.isTrue(2 == yellowColorCarOrders.filter().docCount());
  327.         Aggregate totalPrice = yellowColorCarOrders.filter().aggregations().get("totalPrice");
  328.         Console.log("福特黄色汽车的销售总额:" + totalPrice.sum().value());
  329.     }
  331.     @Test
  332.     @SneakyThrows(IOException.class)
  333.     void searchFilterButAggregationsNotFilter() {
  335.         //1.查询展示蓝色福特汽车，同时统计不同颜色汽车的销售均价
  336.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  337.                 .index(INDEX_NAME)
  338.                 .aggregations("colorCarOrders", a -> a.terms(t -> t
  339.                                 .field("color"))
  340.                         .aggregations("avgPrice", fa -> fa.avg(avg -> avg
  341.                                 .field("price")
  342.                         ))
  343.                 )
  344.                 .postFilter(pf -> pf.bool(b -> b
  345.                         .must(m -> m.term(t -> t
  346.                                 .field("color")
  347.                                 .value(FieldValue.of("blue"))
  348.                         ))
  349.                         .must(m -> m.term(t -> t
  350.                                 .field("make")
  351.                                 .value(FieldValue.of("ford"))
  352.                         )))
  353.                 ), CarOrder.class
  354.         );
  356.         //2.校验
  357.         Assert.isTrue(1 == search.hits().hits().size());
  358.         Aggregate colorCarOrders = search.aggregations().get("colorCarOrders");
  359.         List<StringTermsBucket> colorBuckets = colorCarOrders.sterms().buckets().array();
  360.         Assert.isTrue(4 == colorBuckets.size());
  362.         for (StringTermsBucket colorBucket : colorBuckets) {
  363.             Aggregate avePrice = colorBucket.aggregations().get("avgPrice");
  364.             Console.log(colorBucket.key() + "汽车销售均价" + avePrice.avg().value());
  365.         }
  366.     }
  368.     @Test
  369.     @SneakyThrows(IOException.class)
  370.     void sort() {
  372.         //1.同级排序-统计所有厂商的销售总额，并明确了解谁的销售总额最高
  373.         HashMap<String, SortOrder> sortMaps = MapUtil.newHashMap(1);
  374.         sortMaps.put("totalPrice", SortOrder.Desc);
  375.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  376.                 .index(INDEX_NAME)
  377.                 .aggregations("makeCarOrders", a -> a.terms(t -> t
  378.                                         .field("make")
  379.                                         .order(sortMaps)
  380.                                 )
  381.                                 .aggregations("totalPrice", fa -> fa.sum(sum -> sum
  382.                                         .field("price")
  383.                                 ))
  384.                 ), CarOrder.class
  385.         );
  387.         //2.校验
  388.         Aggregate makeCarOrders = search.aggregations().get("makeCarOrders");
  389.         List<StringTermsBucket> makeBuckets = makeCarOrders.sterms().buckets().array();
  390.         for (StringTermsBucket makeBucket : makeBuckets) {
  391.             Aggregate totalPrice = makeBucket.aggregations().get("totalPrice");
  392.             Console.log(makeBucket.key() + "汽车销售总额" + totalPrice.sum().value());
  393.         }
  394.     }
  396.     @Test
  397.     @SneakyThrows(IOException.class)
  398.     void sortDeep() {
  400.         //1.统计每个季度，每个厂商黄色汽车销售总额，并且直观看到黄色汽车销售总额高的厂商
  401.         HashMap<String, SortOrder> sortMaps = MapUtil.newHashMap(1);
  402.         sortMaps.put("totalPrice>yellowTotalPrice", SortOrder.Desc);
  403.         SearchResponse<CarOrder> search = secondaryClient.search(s -> s
  404.                 .index(INDEX_NAME)
  405.                 .aggregations("quarterOrders", a -> a
  406.                         .dateHistogram(d -> d
  407.                                 .field("soldDate")
  408.                                 .calendarInterval(CalendarInterval.Quarter)
  409.                                 .format(DatePattern.NORM_DATE_PATTERN)
  410.                         )
  411.                         .aggregations("makeOrders", da -> da
  412.                                 .terms(t -> t
  413.                                         .field("make")
  414.                                         .order(sortMaps)
  415.                                 )
  416.                                 .aggregations("totalPrice", daa -> daa
  417.                                         .filter(f -> f.term(t -> t
  418.                                                 .field("color")
  419.                                                 .value(FieldValue.of("yellow"))
  420.                                         ))
  421.                                         .aggregations("yellowTotalPrice", daaa -> daaa.sum(sum -> sum
  422.                                                 .field("price")
  423.                                         ))
  424.                                 )
  425.                         )
  426.                 ), CarOrder.class
  427.         );
  429.         //2.校验
  430.         Aggregate quarterOrders = search.aggregations().get("quarterOrders");
  431.         List<DateHistogramBucket> histogramBuckets = quarterOrders.dateHistogram().buckets().array();
  432.         Assert.isTrue(5 == histogramBuckets.size());
  433.         for (DateHistogramBucket histogramBucket : histogramBuckets) {
  434.             Console.log(histogramBucket.keyAsString() + "-季度共卖出" + histogramBucket.docCount() + "台汽车");
  435.             Console.log("其中黄色汽车订单销售额如下:");
  436.             Aggregate makeOrders = histogramBucket.aggregations().get("makeOrders");
  437.             List<StringTermsBucket> makeBuckets = makeOrders.sterms().buckets().array();
  438.             for (StringTermsBucket makeBucket : makeBuckets) {
  439.                 Aggregate totalPrice = makeBucket.aggregations().get("totalPrice");
  440.                 Aggregate yellowTotalPrice = totalPrice.filter().aggregations().get("yellowTotalPrice");
  441.                 Console.log(makeBucket.key() + "共卖出黄色汽车总价:" + yellowTotalPrice.sum().value());
  442.             }
  443.         }
  444.     }
  446.     @Test
  447.     @SneakyThrows(IOException.class)
  448.     void cardinality() {
  450.         //1.统计共有多少个厂商的汽车订单
  451.         SearchResponse<CarOrder> search = primaryClient.search(s -> s
  452.                 .index(INDEX_NAME)
  453.                 .aggregations("makeCount", a -> a.cardinality(c -> c
  454.                         .field("make")
  455.                         .precisionThreshold(100)
  456.                 )), CarOrder.class
  457.         );
  459.         //2.校验
  460.         Aggregate makeCount = search.aggregations().get("makeCount");
  461.         Assert.isTrue(4 == makeCount.cardinality().value());
  462.     }
  464.     @Test
  465.     @SneakyThrows(IOException.class)
  466.     void percentiles() {
  468.         //1.存储网站请求数据
  469.         saveWebSite();
  471.         //2.统计不同区域网站请求的平均值，通过百分数进行异常值分析，查看哪个区域网络延迟较高
  472.         SearchResponse<WebsiteRequest> search = primaryClient.search(s -> s
  473.                 .index(WEBSITE_INDEX)
  474.                 .aggregations("zoneWebRequest", a -> a
  475.                         .terms(t -> t.field("zone"))
  476.                         .aggregations("avgLatency", ta -> ta.avg(avg -> avg
  477.                                 .field("latency"))
  478.                         )
  479.                         .aggregations("percentilesLatency", ta -> ta.percentiles(p -> p
  480.                                 .field("latency")
  481.                                 .percents((double) 5, (double) 25, (double) 50, (double) 75, (double) 100)
  482.                         ))
  483.                 ), WebsiteRequest.class
  484.         );
  486.         //3.校验
  487.         Aggregate zoneWebRequest = search.aggregations().get("zoneWebRequest");
  488.         List<StringTermsBucket> zoneBuckets = zoneWebRequest.sterms().buckets().array();
  489.         for (StringTermsBucket zoneBucket : zoneBuckets) {
  490.             String key = zoneBucket.key();
  491.             double avgLatency = zoneBucket.aggregations().get("avgLatency").avg().value();
  492.             Console.log(key + "地区 网站平均响应延迟:" + avgLatency);
  494.             TDigestPercentilesAggregate percentilesLatency = zoneBucket.aggregations().get("percentilesLatency").tdigestPercentiles();
  495.             Percentiles values = percentilesLatency.values();
  496.             for (String percentilesItem : values.keyed().keySet()) {
  497.                 Console.log(key + "地区" + percentilesItem + "%的请求 响应延迟:" + values.keyed().get(percentilesItem));
  498.             }
  499.         }
  500.     }
  502.     @Test
  503.     @SneakyThrows(IOException.class)
  504.     void percentileRanks() {
  506.         //1.存储网站请求数据
  507.         saveWebSite();
  509.         //2.获取美国地区，网络延迟达到30,80,100的用户百分比
  510.         SearchResponse<WebsiteRequest> search = secondaryClient.search(s -> s
  511.                 .index(WEBSITE_INDEX)
  512.                 .aggregations("usWebRequests", a -> a
  513.                         .filter(f -> f.term(t -> t
  514.                                 .field("zone")
  515.                                 .value(FieldValue.of("US"))
  516.                         ))
  517.                         .aggregations("usPercentileRanks", fa -> fa.percentileRanks(p -> p
  518.                                 .field("latency")
  519.                                 .values((double) 30, (double) 80, (double) 100)
  520.                         ))
  521.                 ), WebsiteRequest.class
  522.         );
  524.         //3.校验
  525.         Aggregate usWebRequests = search.aggregations().get("usWebRequests");
  526.         Aggregate usPercentileRanks = usWebRequests.filter().aggregations().get("usPercentileRanks");
  527.         TDigestPercentileRanksAggregate tDigestPercentileRanks = usPercentileRanks.tdigestPercentileRanks();
  528.         Map<String, String> keyed = tDigestPercentileRanks.values().keyed();
  529.         Assert.isTrue(3 == keyed.size());
  530.         for (String key : keyed.keySet()) {
  531.             Console.log(keyed.get(key) + "%的用户 网络延迟请求达到了" + key + "毫秒");
  532.         }
  533.     }
  535.     @SneakyThrows(IOException.class)
  536.     void saveWebSite() {
  538.         //1.判定索引是否存在
  539.         BooleanResponse exists = primaryClient.indices().exists(e -> e.index(WEBSITE_INDEX));
  541.         //2.不存在创建索引
  542.         if (!exists.value()) {
  544.             //3.创建索引
  545.             CreateIndexResponse createIndexResponse = primaryClient.indices()
  546.                     .create(c -> c
  547.                             .index(WEBSITE_INDEX)
  548.                             .mappings(m -> m
  549.                                     .properties("latency", p -> p.long_(l -> l))
  550.                                     .properties("timestamp", p -> p.date(d -> d))
  551.                                     .properties("zone", p -> p.keyword(k -> k))
  552.                             )
  553.                     );
  554.             Assert.isTrue(Boolean.TRUE.equals(createIndexResponse.acknowledged()));
  555.         }
  557.         //4.清空数据
  558.         DeleteByQueryResponse deleteByQueryResponse = primaryClient.deleteByQuery(d -> d
  559.                 .index(WEBSITE_INDEX)
  560.                 .query(q -> q.matchAll(m -> m))
  561.                 .refresh(true)
  562.         );
  563.         Assert.isTrue(ObjectUtil.isNotNull(deleteByQueryResponse.deleted()));
  565.         //5.解析数据
  566.         String dataStr = "[{\"latency\":100,\"zone\":\"US\",\"timestamp\":\"2014-10-28\"},{\"latency\":80,\"zone\":\"US\",\"timestamp\":\"2014-10-29\"},{\"latency\":99,\"zone\":\"US\",\"timestamp\":\"2014-10-29\"},{\"latency\":102,\"zone\":\"US\",\"timestamp\":\"2014-10-28\"},{\"latency\":75,\"zone\":\"US\",\"timestamp\":\"2014-10-28\"},{\"latency\":82,\"zone\":\"US\",\"timestamp\":\"2014-10-29\"},{\"latency\":100,\"zone\":\"EU\",\"timestamp\":\"2014-10-28\"},{\"latency\":280,\"zone\":\"EU\",\"timestamp\":\"2014-10-29\"},{\"latency\":155,\"zone\":\"EU\",\"timestamp\":\"2014-10-29\"},{\"latency\":623,\"zone\":\"EU\",\"timestamp\":\"2014-10-28\"},{\"latency\":380,\"zone\":\"EU\",\"timestamp\":\"2014-10-28\"},{\"latency\":319,\"zone\":\"EU\",\"timestamp\":\"2014-10-29\"}]";
  567.         List<WebsiteRequest> websiteRequests = JSONUtil.toList(dataStr, WebsiteRequest.class);
  569.         //6.封装Request
  570.         BulkRequest.Builder builder = new BulkRequest.Builder();
  571.         builder.index(WEBSITE_INDEX);
  572.         for (WebsiteRequest websiteRequest : websiteRequests) {
  573.             builder.operations(o -> o.create(c -> c.document(websiteRequest)));
  574.         }
  575.         builder.refresh(Refresh.True);
  577.         //7.存储数据
  578.         BulkResponse bulk = primaryClient.bulk(builder.build());
  579.         Assert.isFalse(bulk.errors());
  580.     }
  581. }