Elasticsearch 索引客户端

返回《Elasticsearch 搜索与索引》

适用于 Python 3.13，依赖为 elasticsearch==9.2.0。ElasticsearchIndexClient 是示例定义的类，一个实例绑定一个索引。连接地址、认证、超时和重试由调用方创建的官方 Elasticsearch 客户端管理。

接口	作用	返回值
`create_index(mappings, settings=None)`	按字段定义和设置创建新索引	官方创建响应；索引已存在时报错
`bulk_index(records, chunk_size=500, refresh=False)`	分批新增文档，同 ID 整条替换	全部批次成功时返回写入数
`search(query, size=10, **options)`	执行 Query DSL，传入排序、返回字段或聚合等选项	完整搜索响应
`get(document_id)`	按 ID 读取文档	含 `_source` 的完整响应
`update(document_id, fields, refresh=False)`	合并给定字段，不存在时报错	官方更新响应
`delete(document_id, refresh=False)`	删除指定文档	官方删除响应
`close()`	关闭传入客户端的连接	无返回值

search() 保留原生 Query DSL。query 填查询条件；source、sort、from_、aggs 等选项使用官方 Python 客户端的参数名。返回的完整响应保留总数、聚合和分片执行情况；命中文档位于 ["hits"]["hits"]，聚合结果位于 ["aggregations"]。

批量写入中的 id 是文档 ID，doc 是完整正文。 两个键是这个示例的调用约定，封装将它们转成 Bulk API 的 _id 和 _source。id 不会自动成为正文里的字段。

代码块 Python

1
2
records = [{"id": "1", "doc": {"text": "全文检索依靠倒排索引。", "category": "检索"}}]
count = store.bulk_index(records, refresh="wait_for")

bulk_index() 使用官方 helpers.bulk()，每批最多 chunk_size 条，较大的文档还受 helper 的批次字节数限制。同 ID 的 index 操作会替换整条文档：新 doc 没有的旧字段会消失。只修改几个字段时调用 update()；该方法不会为不存在的 ID 新建文档。

批量写入不是整体事务。 某批包含失败项时，helper 默认抛出 BulkIndexError，错误详情在异常的 errors 属性中；同批和之前批次已成功的文档仍然保留，尚未发送的批次不再执行。请求级异常同样交给调用方，封装不吞掉错误，也不额外重试。底层请求是否重试取决于创建官方客户端时的配置。

refresh=False 不等待写入对搜索可见；refresh="wait_for" 会等待受影响分片刷新。批量 helper 为每一批传入这个参数，多批依次等待会延长导入时间。运行示例需要立即检查搜索结果，因而显式传入 "wait_for"。按 ID 的 get() 默认实时读取，不依赖搜索刷新。

常驻服务在每个工作进程中创建官方客户端，请求间复用。停止接收请求、进行中的操作全部完成后关闭连接。多个索引封装共享同一个官方客户端时，调用其中任意实例的 close() 都会关闭这组共享连接，应由同一处管理关闭时机。

request_timeout 是单次底层请求的等待时限，不是整个批量任务的总时限。认证信息、证书和连接重试参数均传给官方 Elasticsearch 构造函数，封装不修改它们。

将代码作为 elasticsearch_client.py 放在调用脚本同目录或 Python 可导入路径中。文件末尾的独立示例连接正文部署的本地服务，环境变量名是示例约定：

代码块 Shell

1
2
3
4
python -m pip install "elasticsearch==9.2.0"
export ELASTICSEARCH_URL="http://localhost:9200"
export ELASTICSEARCH_INDEX="client_example_es"
python elasticsearch_client.py

直接运行会创建 client_example_es 索引，写入三条文档，修改其中一条并删除另一条。 示例保留索引和剩余文档，重复运行前更换一个未使用的索引名。它设置一个主分片、零副本，与本地单节点服务对应；已有索引的查询方式见正文第 8 节。

接口参数与批量操作参考：官方 Python API、官方 Bulk helpers。

elasticsearch_client.py Python

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
"""绑定一个索引的 Elasticsearch 客户端，适用于 Python 3.13、elasticsearch 9.2.0。

安装依赖：python -m pip install "elasticsearch==9.2.0"
直接运行需设置 ELASTICSEARCH_URL 和 ELASTICSEARCH_INDEX，后者必须是尚未使用的索引名。
演示会创建索引并写入三条文档，完成搜索、更新和删除后保留索引，不自动删除整个索引。
"""

import os

from elasticsearch import Elasticsearch, helpers


class ElasticsearchIndexClient:
    """固定索引名；连接、认证、超时和重试由传入的官方客户端管理。"""

    def __init__(self, client: Elasticsearch, index_name: str):
        self.client = client
        self.index_name = index_name

    def create_index(self, mappings: dict, settings=None):
        """按传入的字段定义和索引设置创建新索引；已有索引会报错。"""
        return self.client.indices.create(index=self.index_name, mappings=mappings, settings=settings)

    def bulk_index(self, records, chunk_size=500, refresh=False) -> int:
        """每条记录包含 id 和 doc；同 ID 整条替换，不存在则新增，返回成功写入数。"""
        actions = (
            {"_op_type": "index", "_index": self.index_name, "_id": record["id"], "_source": record["doc"]}
            for record in records
        )
        count, _ = helpers.bulk(self.client, actions, chunk_size=chunk_size, refresh=refresh)
        return count

    def search(self, query: dict, size=10, **options):
        """接收原生 Query DSL；source、sort、from_、aggs 等参数交给官方 search。"""
        return self.client.search(index=self.index_name, query=query, size=size, **options)

    def get(self, document_id: str):
        """按 ID 读取文档，保留 _source、版本信息等官方响应字段。"""
        return self.client.get(index=self.index_name, id=document_id)

    def update(self, document_id: str, fields: dict, refresh=False):
        """合并给定字段；文档不存在时报错，不自动插入。"""
        return self.client.update(index=self.index_name, id=document_id, doc=fields, refresh=refresh)

    def delete(self, document_id: str, refresh=False):
        """删除指定文档；文档不存在时原样抛出官方客户端异常。"""
        return self.client.delete(index=self.index_name, id=document_id, refresh=refresh)

    def close(self):
        """关闭传入客户端的连接；共享这个客户端的使用者须全部完成操作。"""
        self.client.close()


def main():
    client = Elasticsearch(os.environ["ELASTICSEARCH_URL"], request_timeout=30, max_retries=0)
    store = ElasticsearchIndexClient(client, os.environ["ELASTICSEARCH_INDEX"])
    try:
        mappings = {
            "properties": {
                "text": {"type": "text", "analyzer": "cjk"},
                "category": {"type": "keyword"},
            }
        }
        store.create_index(mappings, settings={"number_of_shards": 1, "number_of_replicas": 0})
        records = [
            {"id": "1", "doc": {"text": "全文检索依靠倒排索引。", "category": "检索"}},
            {"id": "2", "doc": {"text": "字段过滤限定检索范围。", "category": "检索"}},
            {"id": "3", "doc": {"text": "日志记录服务运行状态。", "category": "运维"}},
        ]
        # 等待本批写入可被搜索，便于紧接着检查；日常批量导入不必每批等待。
        print("写入数：", store.bulk_index(records, refresh="wait_for"))
        query = {"bool": {"must": [{"match": {"text": "检索"}}], "filter": [{"term": {"category": "检索"}}]}}
        result = store.search(query, source=["text", "category"])
        for hit in result["hits"]["hits"]:
            print(hit["_id"], hit["_score"], hit["_source"])

        print("更新结果：", store.update("1", {"category": "示例"}, refresh="wait_for")["result"])
        print("按 ID 读取：", store.get("1")["_source"])
        print("删除结果：", store.delete("3", refresh="wait_for")["result"])
        print("删除后命中数：", store.search({"ids": {"values": ["3"]}})["hits"]["total"]["value"])
    finally:
        store.close()


if __name__ == "__main__":
    main()