【翻译】Elasticsearch Java API Client 8.13.2 （第四章-使用 Java API 客户端）

以下部分提供了有关 Elasticsearch 最常用和一些不太明显的功能的教程。

有关完整参考，请参阅 Elasticsearch 文档，特别是 REST API 部分。 Java API 客户端使用 Java API 约定，严格遵循此处描述的 JSON 结构。

索引单个文档

如果您是 Elasticsearch 的新手，请务必阅读 Elasticsearch 的快速入门，其中提供了很好的介绍。

Java API 客户端提供了多种索引数据的方法：您可以提供将自动映射到 JSON 的应用程序对象，也可以提供原始 JSON 数据。使用应用程序对象更适合具有明确定义的域模型的应用程序，而原始 JSON 更适合使用半结构化数据记录用例。

在下面的示例中，我们使用具有 sku 、 name 和 price 属性的 Product 域对象。

使用 Fluent DSL

构建请求最直接的方法是使用 Fluent DSL。在下面的示例中，我们使用产品的 SKU 作为索引中的文档标识符，在 products 索引中对产品描述进行索引。 product 对象将使用 Elasticsearch 客户端上配置的对象映射器映射到 JSON。

Product product = new Product("bk-1", "City bike", 123.0);
 
IndexResponse response = esClient.index(i -> i
    .index("products")
    .id(product.getSku())
    .document(product)
);
 
logger.info("Indexed with version " + response.version());

您还可以将使用 DSL 创建的对象分配给变量。 Java API 客户端类为此有一个静态 of() 方法，该方法使用 DSL 语法创建对象。（指基于JSON数据格式的查询）

Product product = new Product("bk-1", "City bike", 123.0);
 
IndexRequest<Product> request = IndexRequest.of(i -> i
    .index("products")
    .id(product.getSku())
    .document(product)
);
 
IndexResponse response = esClient.index(request);
 
logger.info("Indexed with version " + response.version());

使用经典构建器

如果您更习惯经典的构建器模式，它也可用。构建器对象通过流畅的 DSL 语法在底层使用。

Product product = new Product("bk-1", "City bike", 123.0);
 
IndexRequest.Builder<Product> indexReqBuilder = new IndexRequest.Builder<>();
indexReqBuilder.index("product");
indexReqBuilder.id(product.getSku());
indexReqBuilder.document(product);
 
IndexResponse response = esClient.index(indexReqBuilder.build());
 
logger.info("Indexed with version " + response.version());

使用异步客户端

上面的示例使用同步 Elasticsearch 客户端。所有 Elasticsearch API 也可在异步客户端中使用，使用相同的请求和响应类型。另请参阅阻塞和异步客户端以了解更多详细信息。

ElasticsearchAsyncClient esAsyncClient = new ElasticsearchAsyncClient(transport);
 
Product product = new Product("bk-1", "City bike", 123.0);
 
esAsyncClient.index(i -> i
    .index("products")
    .id(product.getSku())
    .document(product)
).whenComplete((response, exception) -> {
    if (exception != null) {
        logger.error("Failed to index", exception);
    } else {
        logger.info("Indexed with version " + response.version());
    }
});

使用原始 JSON 数据

当您想要索引的数据来自外部源时，必须创建域对象可能很麻烦，或者对于半结构化数据来说完全不可能。

您可以使用 withJson() 对任意来源的数据进行索引。使用此方法将读取源并将其用于索引请求的 document 属性。有关更多详细信息，请参阅从 JSON 数据创建 API 对象。

Reader input = new StringReader(
    "{'@timestamp': '2022-04-08T13:55:32Z', 'level': 'warn', 'message': 'Some log message'}"
    .replace('\'', '"'));
 
IndexRequest<JsonData> request = IndexRequest.of(i -> i
    .index("logs")
    .withJson(input)
);
 
IndexResponse response = esClient.index(request);
 
logger.info("Indexed with version " + response.version());

批量：索引多个文档

批量请求允许在一个请求中向 Elasticsearch 发送多个与文档相关的操作。当您有多个文档要摄取时，这比通过单独的请求发送每个文档更有效。

批量请求可以包含多种操作：

索引应用程序对象

BulkRequest 包含操作的集合，每个操作都是具有多个变体的类型。要创建此请求，可以方便地为主请求使用构建器对象，并为每个操作使用流畅的 DSL。

下面的示例显示如何索引列表或应用程序对象。

List<Product> products = fetchProducts();
 
BulkRequest.Builder br = new BulkRequest.Builder();
 
for (Product product : products) {
    br.operations(op -> op   //添加一个操作（请记住，列表属性是可加的）。 op 是 BulkOperation 的构建器，它是一种变体类型。此类型有 index 、 create 、 update 和 delete 变体。        
        .index(idx -> idx    //选择 index 操作变体， idx 是 IndexOperation 的构建器。        
            .index("products")   //设置索引操作的属性，类似于单文档索引：索引名称、标识符和文档。    
            .id(product.getSku())
            .document(product)
        )
    );
}
 
BulkResponse result = esClient.bulk(br.build());
 
// Log errors, if any
if (result.errors()) {
    logger.error("Bulk had errors");
    for (BulkResponseItem item: result.items()) {
        if (item.error() != null) {
            logger.error(item.error().reason());
        }
    }
}

索引原始 JSON 数据

批量索引请求的 document 属性可以是任何可以使用 Elasticsearch 客户端的 JSON 映射器序列化为 JSON 的对象。然而，批量摄取的数据通常以 JSON 文本形式提供（例如磁盘上的文件），而解析此 JSON 只是为了重新序列化它以发送批量请求会浪费资源。因此，批量操作中的文档也可以是 BinaryData 类型，逐字发送（无需解析）到 Elasticsearch 服务器。

在下面的示例中，我们将使用 Java API 客户端的 BinaryData 从日志目录中读取 json 文件并在批量请求中发送它们。

// List json log files in the log directory
File[] logFiles = logDir.listFiles(
    file -> file.getName().matches("log-.*\\.json")
);
 
BulkRequest.Builder br = new BulkRequest.Builder();
 
for (File file: logFiles) {
    FileInputStream input = new FileInputStream(file);
    BinaryData data = BinaryData.of(IOUtils.toByteArray(input), ContentType.APPLICATION_JSON);
 
    br.operations(op -> op
        .index(idx -> idx
            .index("logs")
            .document(data)
        )
    );
}

使用 Bulk Ingester 进行流式摄取

BulkIngester 通过提供一个实用程序类来简化批量 API 的使用，该实用程序类允许将索引/更新/删除操作透明地分组到批量请求中。您只需 add() 对摄取器进行批量操作，它将根据其配置进行分组和批量发送。

The ingester will send a bulk request when one of the following criteria is met:
当满足以下条件之一时，摄取器将发送批量请求：

此外，您还可以定义等待 Elasticsearch 执行的并发请求的最大数量（默认为 1）。当达到该最大值并且已收集最大操作数时，向索引器添加新操作将被阻止。这可以避免通过对客户端应用程序施加背压而导致 Elasticsearch 服务器过载。

BulkIngester<Void> ingester = BulkIngester.of(b -> b
    .client(esClient)    //设置用于发送批量请求的 Elasticsearch 客户端。
    .maxOperations(100)  //设置发送批量请求之前要收集的最大操作数。
    .flushInterval(1, TimeUnit.SECONDS) //设置冲洗间隔。
);
 
for (File file: logFiles) {
    FileInputStream input = new FileInputStream(file);
    BinaryData data = BinaryData.of(IOUtils.toByteArray(input), ContentType.APPLICATION_JSON);
 
    ingester.add(op -> op //向摄取器添加批量操作。
        .index(idx -> idx
            .index("logs")
            .document(data)
        )
    );
}
 
ingester.close(); //关闭摄取器以刷新挂起的操作并释放资源。

此外，批量摄取器接受侦听器，以便您的应用程序可以收到发送的批量请求及其结果的通知。为了允许将批量操作与应用程序上下文相关联， add() 方法可以选择接受 context 参数。此上下文参数的类型用作 BulkIngester 对象的通用参数。您可能已经注意到上面 BulkIngester<Void> 中的 Void 类型：这是因为我们没有注册侦听器，因此不关心上下文值。

以下示例展示了如何使用上下文值来实现批量摄取侦听器：与之前一样，它批量发送 JSON 日志文件，但跟踪批量请求错误和失败的操作。当操作失败时，根据错误类型，您可能需要将其重新添加到摄取器中。

BulkListener<String> listener = new BulkListener<String>() { //创建一个侦听器，其中上下文值是所摄取文件名的字符串。
    @Override
    public void beforeBulk(long executionId, BulkRequest request, List<String> contexts) {
    }
 
    @Override
    public void afterBulk(long executionId, BulkRequest request, List<String> contexts, BulkResponse response) {
        // The request was accepted, but may contain failed items.
        // The "context" list gives the file name for each bulk item.
        logger.debug("Bulk request " + executionId + " completed");
        for (int i = 0; i < contexts.size(); i++) {
            BulkResponseItem item = response.items().get(i);
            if (item.error() != null) {
                // Inspect the failure cause
                logger.error("Failed to index file " + contexts.get(i) + " - " + item.error().reason());
            }
        }
    }
 
    @Override
    public void afterBulk(long executionId, BulkRequest request, List<String> contexts, Throwable failure) {
        // The request could not be sent
        logger.debug("Bulk request " + executionId + " failed", failure);
    }
};
 
BulkIngester<String> ingester = BulkIngester.of(b -> b
    .client(esClient)
    .maxOperations(100)
    .flushInterval(1, TimeUnit.SECONDS)
    .listener(listener) //在批量接收器上注册侦听器。
);
 
for (File file: logFiles) {
    FileInputStream input = new FileInputStream(file);
    BinaryData data = BinaryData.of(IOUtils.toByteArray(input), ContentType.APPLICATION_JSON);
 
    ingester.add(op -> op
        .index(idx -> idx
            .index("logs")
            .document(data)
        ),
        file.getName() //将文件名设置为批量操作的上下文值。
    );
}
 
ingester.close();

批量摄取还公开统计信息，允许监控摄取过程并调整其配置：

GetResponse<Product> response = esClient.get(g -> g
    .index("products") //get 请求，带有索引名称和标识符。
    .id("bk-1"),
    Product.class   //目标类，此处为 Product 。   
);
 
if (response.found()) {
    Product product = response.source();
    logger.info("Product name " + product.getName());
} else {
    logger.info ("Product not found");
}

GetResponse<ObjectNode> response = esClient.get(g -> g
    .index("products")
    .id("bk-1"),
    ObjectNode.class    //目标类是原始 JSON 对象。 
);
 
if (response.found()) {
    ObjectNode json = response.source();
    String name = json.get("name").asText();
    logger.info("Product name " + name);
} else {
    logger.info("Product not found");
}

String searchText = "bike";
 
SearchResponse<Product> response = esClient.search(s -> s
    .index("products") //我们要搜索的索引的名称。
    .query(q -> q      //搜索请求的查询部分（搜索请求还可以具有其他组件，例如聚合）。
        .match(t -> t   //从众多可用查询变体中选择一个。我们在这里选择匹配查询（全文搜索）。
            .field("name")  //配置匹配查询：我们在 name 字段中搜索术语。
            .query(searchText)
        )
    ),
    Product.class      //匹配文档的目标类。我们在这里使用 Product ，就像在 get 请求示例中一样。
);
 
TotalHits total = response.hits().total();
boolean isExactResult = total.relation() == TotalHitsRelation.Eq;
 
if (isExactResult) {
    logger.info("There are " + total.value() + " results");
} else {
    logger.info("There are more than " + total.value() + " results");
}
 
List<Hit<Product>> hits = response.hits().hits();
for (Hit<Product> hit: hits) {
    Product product = hit.source();
    logger.info("Found product " + product.getSku() + ", score " + hit.score());
}

String searchText = "bike";
double maxPrice = 200.0;
 
// Search by product name
Query byName = MatchQuery.of(m -> m  //我们正在单独创建针对各个条件的查询。
    .field("name")
    .query(searchText)
)._toQuery(); //MatchQuery 是一个查询变体，我们必须将其转换为 Query 联合类型。有关更多详细信息，请参阅变体类型。
 
// Search by max price
Query byMaxPrice = RangeQuery.of(r -> r
    .field("price")
    .gte(JsonData.of(maxPrice)) //Elasticsearch 范围查询接受大范围的值类型。我们在这里创建最高价格的 JSON 表示形式。（指的是查询product的最大价格）
)._toQuery();
 
// Combine name and price queries to search the product index
SearchResponse<Product> response = esClient.search(s -> s
    .index("products")
    .query(q -> q
        .bool(b -> b  //搜索查询是一个布尔查询，结合了文本搜索和最高价格查询。
            .must(byName)   //两个查询都添加为 must ，因为我们希望结果匹配所有条件。
            .must(byMaxPrice)
        )
    ),
    Product.class
);
 
List<Hit<Product>> hits = response.hits().hits();
for (Hit<Product> hit: hits) {
    Product product = hit.source();
    logger.info("Found product " + product.getSku() + ", score " + hit.score());
}

// Create a script
esClient.putScript(r -> r
    .id("query-script")   //要创建的模板脚本的标识符。
    .script(s -> s
        .lang("mustache")
        .source("{\"query\":{\"match\":{\"{{field}}\":\"{{value}}\"}}}")
    ));

SearchTemplateResponse<Product> response = esClient.searchTemplate(r -> r
        .index("some-index")
        .id("query-script") //要使用的模板脚本的标识符。
        .params("field", JsonData.of("some-field")) //模板参数值。
        .params("value", JsonData.of("some-data")),
    Product.class
);
 
List<Hit<Product>> hits = response.hits().hits();
for (Hit<Product> hit: hits) {
    Product product = hit.source();
    logger.info("Found product " + product.getSku() + ", score " + hit.score());
}

String searchText = "bike";
 
Query query = MatchQuery.of(m -> m
    .field("name")
    .query(searchText)
)._toQuery();
 
SearchResponse<Void> response = esClient.search(b -> b
    .index("products")
    .size(0)   //将匹配文档的数量设置为零，因为我们只使用价格直方图。
    .query(query)   //设置填充过滤要运行聚合的产品的查询
    .aggregations("price-histogram", a -> a //创建一个名为“price-histogram”的聚合。您可以根据需要添加任意数量的命名聚合。
        .histogram(h -> h    //选择 histogram 聚合变体。
            .field("price")
            .interval(50.0)
        )
    ),
    Void.class  //我们不关心匹配（ size 设置为零），使用 Void 将忽略响应中的任何文档。
);

List<HistogramBucket> buckets = response.aggregations()
    .get("price-histogram") //获取“价格直方图”聚合的结果。
    .histogram()  //将其转换为 histogram 变体结果。这必须与聚合定义一致。
    .buckets().array();  //桶可以表示为数组或映射。这会向下转换为数组变体（默认）。
 
for (HistogramBucket bucket: buckets) {
    logger.info("There are " + bucket.docCount() +
        " bikes under " + bucket.key());
}