10.4 Elasticsearch Client Libraries (客户端)

10.4 Elasticsearch Client Libraries (客户端)

10.4 Elasticsearch Client Libraries (客户端) 详解与实践

10.4.1 客户端库的重要性

Elasticsearch 提供了 RESTful API，理论上任何能够发送 HTTP 请求的程序都可以与之交互。然而，直接使用 RESTful API 进行开发存在一些挑战：

• 学习成本高: 需要深入了解 Elasticsearch RESTful API 的各种端点、请求方法、请求体结构以及响应格式。

• 开发效率低: 需要手动构建 HTTP 请求，处理 JSON 序列化和反序列化，以及错误处理等繁琐任务。

• 维护性差: 代码可读性降低，维护和调试变得困难。

客户端库的出现正是为了解决这些问题。它们是对 Elasticsearch RESTful API 的一层封装，提供了更高级、更友好的 API 接口，使得开发者可以使用编程语言的习惯方式来操作 Elasticsearch。使用客户端库的优势包括：

• 简化操作: 抽象了底层的 HTTP 通信细节，开发者只需关注业务逻辑。

• 提高效率: 提供了便捷的 API，减少了重复代码的编写，加速开发进程。

• 增强可读性: 使用面向对象或函数式编程风格的 API，代码更易读、易维护。

• 语言集成: 针对不同的编程语言提供了相应的客户端库，方便在各种应用环境中使用 Elasticsearch。

• 官方支持与社区活跃: Elasticsearch 官方提供了多种语言的客户端库，并有活跃的社区支持，保证了库的质量和持续更新。

10.4.2 Elasticsearch 官方客户端库概览

Elasticsearch 官方维护并推荐使用以下客户端库，覆盖了主流的编程语言：

• Java Client: 作为 Elasticsearch 的原生语言客户端，功能最全面，性能最佳，分为 High Level REST Client 和 Low Level REST Client。High Level REST Client 基于 Low Level REST Client 构建，提供了更高级别的 API 抽象，更易于使用。

• Python Client: 流行的 Python 客户端库 elasticsearch-py，易用性强，广泛应用于数据分析、Web 应用等领域。

• JavaScript Client: 用于 Node.js 和浏览器环境的 elasticsearch-js 客户端，方便构建前端搜索应用和后端服务。

• Go Client: go-elasticsearch 客户端，为 Go 语言开发者提供高效的 Elasticsearch 交互能力。

• .NET Client: Elasticsearch.Net 和 NEST 客户端，分别对应 Low Level 和 High Level 的 .NET 客户端，适用于 .NET 平台的应用。

• Ruby Client: elasticsearch-ruby 客户端，为 Ruby 开发者提供 Elasticsearch 集成方案。

• PHP Client: elasticsearch-php 客户端，方便 PHP 应用与 Elasticsearch 集成。

除了官方客户端库，社区也贡献了许多其他的客户端库，例如 Perl、C# (非官方 NEST) 等。但官方客户端库由于其稳定性和官方支持，通常是首选。

10.4.3 Java High Level REST Client 代码实践详解

1. 添加依赖 (Maven)

首先，需要在 pom.xml 文件中添加 Java High Level REST Client 的依赖：

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>版本号</version> <!- -  请替换为您的 Elasticsearch 版本对应的客户端版本 -->
</dependency>
​

注意: 客户端库的版本需要与 Elasticsearch 服务端的版本兼容。建议使用与服务端版本一致的客户端版本。

2. 创建 REST High Level Client 实例

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
public class ElasticsearchClientExample {
    public static void main(String[] args) throws Exception {
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http") // Elasticsearch 服务地址
                )
        );
        // ... 后续操作 ...
        client.close(); // 关闭客户端
    }
}
​

这段代码创建了一个 RestHighLevelClient 实例，连接到本地 9200 端口的 Elasticsearch 服务。可以配置多个 HttpHost 来连接到集群中的多个节点。

3. 索引文档 (Index API)

import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.common.xcontent.XContentType;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
// ... 客户端初始化代码 ...
IndexRequest request = new IndexRequest("my_index"); // 索引名称
// request.type("_doc"); // 在 Elasticsearch 及以上版本中，type 已被弃用
// 方式一：使用 Map 构建文档
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("user", "kimchy");
jsonMap.put("post_date", new Date());
jsonMap.put("message", "trying out Elasticsearch");
request.source(jsonMap);
// 方式二：使用 JSON 字符串构建文档
// String jsonString = "{\"user\":\"kimchy\", \"post_date\":\"2023-10-27T10:00:00Z\", \"message\":\"trying out Elasticsearch\"}";
// request.source(jsonString, XContentType.JSON);
// 方式三：使用 XContentBuilder 构建文档 (更复杂，但更灵活)
// XContentBuilder builder = XContentFactory.jsonBuilder();
// builder.startObject();
// {
//     builder.field("user", "kimchy");
//     builder.timeField("post_date", new Date());
//     builder.field("message", "trying out Elasticsearch");
// }
// builder.endObject();
// request.source(builder);
IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
System.out.println("Index: " + indexResponse.getIndex());
System.out.println("Type: " + indexResponse.getType());
System.out.println("Id: " + indexResponse.getId());
System.out.println("Version: " + indexResponse.getVersion());
System.out.println("Result: " + indexResponse.getResult());
System.out.println("SeqNo: " + indexResponse.getSeqNo());
System.out.println("PrimaryTerm: " + indexResponse.getPrimaryTerm());
​

这段代码演示了如何使用 IndexRequest 索引文档到名为 my_index 的索引中。文档内容可以使用 Map, JSON 字符串或 XContentBuilder 构建。 RequestOptions.DEFAULT 表示使用默认的请求选项。IndexResponse 包含了索引操作的结果信息。

4. 获取文档 (Get API)

import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
// ... 客户端初始化代码 ...
GetRequest getRequest = new GetRequest(
        "my_index", // 索引名称
        "_doc",     // 类型 (7.x 以上版本可选)
        "1"         // 文档 ID
);
GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
if (getResponse.isExists()) {
    String index = getResponse.getIndex();
    String type = getResponse.getType();
    String id = getResponse.getId();
    long version = getResponse.getVersion();
    String sourceAsString = getResponse.getSourceAsString(); // 获取文档 JSON 字符串
    Map<String, Object> sourceAsMap = getResponse.getSourceAsMap(); // 获取文档 Map
    byte[] sourceAsBytes = getResponse.getSourceAsBytes(); // 获取文档字节数组
    System.out.println("Index: " + index);
    System.out.println("Type: " + type);
    System.out.println("Id: " + id);
    System.out.println("Version: " + version);
    System.out.println("Source (String): " + sourceAsString);
    System.out.println("Source (Map): " + sourceAsMap);
} else {
    System.out.println("Document not found");
}
​

这段代码使用 GetRequest 根据索引、类型和 ID 获取文档。GetResponse 包含了文档是否存在以及文档内容。

5. 搜索文档 (Search API)

import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
// ... 客户端初始化代码 ...
SearchRequest searchRequest = new SearchRequest("my_index"); // 索引名称
// searchRequest.types("_doc"); // 类型 (7.x 以上版本可选)
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.query(QueryBuilders.matchAllQuery()); // 查询所有文档
// sourceBuilder.query(QueryBuilders.matchQuery("message", "Elasticsearch")); // 关键词查询
searchRequest.source(sourceBuilder);
SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
SearchHits hits = searchResponse.getHits();
SearchHit[] searchHits = hits.getHits();
for (SearchHit hit : searchHits) {
    String index = hit.getIndex();
    String type = hit.getType();
    String id = hit.getId();
    float score = hit.getScore();
    String sourceAsString = hit.getSourceAsString();
    Map<String, Object> sourceAsMap = hit.getSourceAsMap();
    System.out.println("Index: " + index);
    System.out.println("Type: " + type);
    System.out.println("Id: " + id);
    System.out.println("Score: " + score);
    System.out.println("Source (String): " + sourceAsString);
    System.out.println("Source (Map): " + sourceAsMap);
}
​

这段代码使用 SearchRequest 和 SearchSourceBuilder 构建搜索请求。QueryBuilders.matchAllQuery() 表示匹配所有文档。 SearchResponse 包含了搜索结果，可以通过 getHits() 方法获取 SearchHits 对象，再遍历 SearchHit 数组获取每个匹配的文档信息。

6. 更新文档 (Update API)

import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.common.xcontent.XContentType;
// ... 客户端初始化代码 ...
UpdateRequest request = new UpdateRequest(
        "my_index", // 索引名称
        "_doc",     // 类型 (7.x 以上版本可选)
        "1"         // 文档 ID
);
// 方式一：使用 Map 更新文档
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("updated_field", "updated value");
jsonMap.put("new_field", "new value");
request.doc(jsonMap);
// 方式二：使用 JSON 字符串更新文档
// String jsonString = "{\"updated_field\":\"updated value\", \"new_field\":\"new value\"}";
// request.doc(jsonString, XContentType.JSON);
// 方式三：使用 Script 更新文档 (更复杂，但更强大，可以执行更复杂的更新逻辑)
// Script inlineScript = new Script(ScriptType.INLINE, "painless", "ctx._source.counter += params.count", Collections.singletonMap("count", 1));
// request.script(inlineScript);
UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
System.out.println("Index: " + updateResponse.getIndex());
System.out.println("Type: " + updateResponse.getType());
System.out.println("Id: " + updateResponse.getId());
System.out.println("Version: " + updateResponse.getVersion());
System.out.println("Result: " + updateResponse.getResult());
​

这段代码演示了如何使用 UpdateRequest 更新文档。 可以使用 Map 或 JSON 字符串更新文档字段，也可以使用 Script 执行更复杂的更新逻辑。

7. 删除文档 (Delete API)

import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
// ... 客户端初始化代码 ...
DeleteRequest request = new DeleteRequest(
        "my_index", // 索引名称
        "_doc",     // 类型 (7.x 以上版本可选)
        "1"         // 文档 ID
);
DeleteResponse deleteResponse = client.delete(request, RequestOptions.DEFAULT);
System.out.println("Index: " + deleteResponse.getIndex());
System.out.println("Type: " + deleteResponse.getType());
System.out.println("Id: " + deleteResponse.getId());
System.out.println("Version: " + deleteResponse.getVersion());
System.out.println("Result: " + deleteResponse.getResult());
​

这段代码使用 DeleteRequest 根据索引、类型和 ID 删除文档。

8. Bulk API (批量操作)

Bulk API 可以将多个索引、更新或删除操作批量发送到 Elasticsearch，提高效率。

import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
// ... 客户端初始化代码 ...
BulkRequest bulkRequest = new BulkRequest();
// 添加 Index 操作
bulkRequest.add(new IndexRequest("my_index").id("2").source(XContentType.JSON, "field", "value2"));
bulkRequest.add(new IndexRequest("my_index").id("3").source(XContentType.JSON, "field", "value3"));
// 添加 Update 操作
bulkRequest.add(new UpdateRequest("my_index", "_doc", "1").doc(XContentType.JSON, "updated_field", "bulk updated"));
// 添加 Delete 操作
bulkRequest.add(new DeleteRequest("my_index", "_doc", "4"));
BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
if (bulkResponse.hasFailures()) {
    System.out.println("Bulk operation failed: " + bulkResponse.buildFailureMessage());
} else {
    System.out.println("Bulk operation successful");
}
​

这段代码演示了如何使用 BulkRequest 批量执行索引、更新和删除操作。

10.4.4 Python Elasticsearch Client 代码实践详解

Python 客户端 elasticsearch-py 也非常流行且易用。

1. 安装客户端库

pip install elasticsearch
​

2. 连接 Elasticsearch

from elasticsearch import Elasticsearch
es = Elasticsearch(
    ['http://localhost:9200'] # Elasticsearch 服务地址列表
)
if es.ping():
    print("Connected to Elasticsearch")
else:
    print("Failed to connect to Elasticsearch")
​

3. 索引文档

index_name = "my_index_python"
document = {
    "user": "python_user",
    "post_date": "2023-10-27T12:00:00Z",
    "message": "Hello from Python Elasticsearch client!"
}
response = es.index(index=index_name, document=document)
print(response)
​

4. 获取文档

index_name = "my_index_python"
doc_id = "1"
response = es.get(index=index_name, id=doc_id)
if response['found']:
    print(response['_source'])
else:
    print("Document not found")
​

5. 搜索文档

index_name = "my_index_python"
query = {
    "match_all": {} # 查询所有文档
    # "match": { "message": "Python" } # 关键词查询
}
response = es.search(index=index_name, query=query)
hits = response['hits']['hits']
for hit in hits:
    print(hit['_source'])
​

6. 更新文档

index_name = "my_index_python"
doc_id = "1"
document_update = {
    "doc": {
        "updated_field": "updated value by python",
        "new_field": "new field by python"
    }
}
response = es.update(index=index_name, id=doc_id, body=document_update)
print(response)
​

7. 删除文档

index_name = "my_index_python"
doc_id = "1"
response = es.delete(index=index_name, id=doc_id)
print(response)
​

8. Bulk API

index_name = "my_index_python"
bulk_data = [
    {"index": {"_index": index_name, "_id": 2}},
    {"field": "value2 from python bulk"},
    {"index": {"_index": index_name, "_id": 3}},
    {"field": "value3 from python bulk"},
    {"update": {"_index": index_name, "_type": "_doc", "_id": 1}},
    {"doc": {"updated_field": "bulk updated by python"}},
    {"delete": {"_index": index_name, "_type": "_doc", "_id": 4}}
]
response = es.bulk(operations=bulk_data)
print(response)
​

10.4.5 JavaScript Elasticsearch Client 代码实践详解 (Node.js)

JavaScript 客户端 elasticsearch-js 用于 Node.js 环境。

1. 安装客户端库

npm install @elastic/elasticsearch
​

2. 连接 Elasticsearch

const { Client } = require('@elastic/elasticsearch')
const client = new Client({ node: 'http://localhost:9200' }) // Elasticsearch 服务地址
async function checkConnection() {
  try {
    const response = await client.ping()
    console.log("Connected to Elasticsearch");
  } catch (error) {
    console.error("Failed to connect to Elasticsearch", error);
  }
}
checkConnection();
​

3. 索引文档

async function indexDocument() {
  try {
    const response = await client.index({
      index: 'my_index_js',
      document: {
        user: 'js_user',
        post_date: new Date(),
        message: 'Hello from JavaScript Elasticsearch client!'
      }
    })
    console.log(response);
  } catch (error) {
    console.error(error);
  }
}
indexDocument();
​

4. 获取文档

async function getDocument() {
  try {
    const response = await client.get({
      index: 'my_index_js',
      id: '1'
    })
    if (response.found) {
      console.log(response._source);
    } else {
      console.log("Document not found");
    }
  } catch (error) {
    console.error(error);
  }
}
getDocument();
​

5. 搜索文档

async function searchDocument() {
  try {
    const response = await client.search({
      index: 'my_index_js',
      query: {
        match_all: {} // 查询所有文档
        // match: { message: 'JavaScript' } // 关键词查询
      }
    })
    response.hits.hits.forEach(hit => {
      console.log(hit._source);
    });
  } catch (error) {
    console.error(error);
  }
}
searchDocument();
​

6. 更新文档

async function updateDocument() {
  try {
    const response = await client.update({
      index: 'my_index_js',
      id: '1',
      doc: {
        updated_field: 'updated value by js',
        new_field: 'new field by js'
      }
    })
    console.log(response);
  } catch (error) {
    console.error(error);
  }
}
updateDocument();
​

7. 删除文档

async function deleteDocument() {
  try {
    const response = await client.delete({
      index: 'my_index_js',
      id: '1'
    })
    console.log(response);
  } catch (error) {
    console.error(error);
  }
}
deleteDocument();
​

8. Bulk API

async function bulkOperation() {
  try {
    const response = await client.bulk({
      operations: [
        { index: { _index: 'my_index_js', _id: 2 } },
        { field: 'value2 from js bulk' },
        { index: { _index: 'my_index_js', _id: 3 } },
        { field: 'value3 from js bulk' },
        { update: { _index: 'my_index_js', _type: '_doc', _id: 1 } },
        { doc: { updated_field: 'bulk updated by js' } },
        { delete: { _index: 'my_index_js', _type: '_doc', _id: 4 } }
      ]
    })
    console.log(response);
  } catch (error) {
    console.error(error);
  }
}
bulkOperation();
​

10.4.6 客户端库架构示意图 (Mermaid Graph TD)

图示说明:

• 应用程序 (Java/Python/JS等): 代表使用不同编程语言开发的应用程序。

• Elasticsearch Client Library: 客户端库，例如 Java High Level REST Client, Python elasticsearch-py, JavaScript elasticsearch-js 等。

• RESTful API: Elasticsearch 提供的 RESTful API 接口。

• Elasticsearch 集群: Elasticsearch 服务端集群。

应用程序通过客户端库与 Elasticsearch 集群交互。客户端库封装了 RESTful API 的细节，使得应用程序开发者可以使用更友好的 API 进行操作。

10.4.7 选择合适的客户端库

选择客户端库时，需要考虑以下因素：

• 编程语言: 根据应用程序使用的编程语言选择对应的客户端库。

• 官方 vs. 社区: 优先选择官方客户端库，它们通常更稳定、功能更完善、且有官方支持。

• 功能需求: 根据项目需求选择客户端库，例如是否需要高级功能 (如聚合、Scroll API 等)。

• 性能需求: 对于性能敏感的应用，可以考虑使用性能更优的客户端库，例如 Java High Level REST Client。

• 易用性: 对于快速开发或原型验证，可以选择易用性更强的客户端库，例如 Python elasticsearch-py。

• 社区活跃度: 活跃的社区意味着更好的支持和更快的 bug 修复。

10.4.8 最佳实践与注意事项

• 版本兼容性: 确保客户端库版本与 Elasticsearch 服务端版本兼容，避免版本不兼容导致的问题。

• 连接管理: 合理管理客户端连接，避免频繁创建和关闭连接，可以使用连接池来提高性能。

• 错误处理: 完善的错误处理机制，捕获客户端库抛出的异常，并进行适当的处理。

• 性能优化: 使用 Bulk API 进行批量操作，减少网络请求次数，提高性能。

• 安全配置: 配置客户端连接的认证信息，确保安全访问 Elasticsearch 集群。

• 异步操作: 对于耗时操作，可以考虑使用客户端库提供的异步 API，提高应用程序的响应速度。

• 文档阅读: 详细阅读客户端库的官方文档，了解其 API 和功能，以便更好地使用。

10.4.9 总结

Elasticsearch 客户端库是构建 Elasticsearch 应用的重要组成部分。它们极大地简化了与 Elasticsearch 的交互，提高了开发效率。本文详细介绍了 Java, Python 和 JavaScript 官方客户端库的代码实践，并提供了架构示意图和选择客户端库的建议。 掌握客户端库的使用，能够帮助开发者更好地利用 Elasticsearch 的强大功能，构建高效、可靠的搜索和分析应用。 选择合适的客户端库，并遵循最佳实践，将能充分发挥 Elasticsearch 的优势，为业务带来价值。