Python 操作 Elasticsearch 超全实战教程

前言

Elasticsearch（简称 ES）是目前最主流的分布式全文搜索引擎，广泛用于日志检索、业务模糊查询、内容搜索、数据分析等场景。Python 操作 ES 分为同步和异步两种方式，在 Web 异步项目（FastAPI/Starlette）、爬虫、高并发业务中，异步 AsyncElasticsearch 是最优选择，不会阻塞事件循环，并发性能更强。

本文基于 elasticsearch 8.x 版本，从零讲解 Python 异步操作 ES 全套核心用法，包含：异步连接配置、索引管理、文档 CRUD、各类常用查询，所有代码可直接运行，解决新手连接报错、连接不释放、查询不会写等问题。

一、环境准备

1.1 安装依赖

安装官方 ES 异步客户端依赖：

pip install elasticsearch

1.2 环境说明

• ES 服务：本地 localhost:9200（无账号密码、无 HTTPS）

• Python 版本：3.8+

• 客户端：AsyncElasticsearch（官方异步客户端）

• 规避问题：自动管理连接，杜绝 Unclosed connector 连接泄露警告

二、ES 异步连接初始化（最佳实践）

核心要点：使用 async with 上下文管理器，自动创建、释放连接，无需手动关闭，彻底解决连接未关闭警告，是生产环境标准写法。

import asyncio
from elasticsearch import AsyncElasticsearch

async def get_es_client() -> AsyncElasticsearch:
    """
    【单函数封装】获取异步ES客户端实例 + 校验连接状态
    整合原有连接创建、连接测试逻辑，极简调用，自动管理连接
    :return: 可用的异步ES客户端对象
    """
    # 初始化ES异步客户端
    client = AsyncElasticsearch(
        hosts=["http://localhost:9200"],
        request_timeout=30,  # 请求超时时间
        retry_on_timeout=True  # 超时自动重试
    )
    # 校验连接有效性
    is_connect = await client.ping()
    if is_connect:
        print("✅ ES 异步连接成功！")
    else:
        print("❌ ES 连接失败！请检查ES服务是否启动、地址是否正确")
    return client

# 调用示例
if __name__ == "__main__":
    # 一行获取可直接使用的ES客户端
    es_client = asyncio.run(get_es_client())

三、索引（Index）核心操作

索引可以理解为 MySQL 的数据表，是存储文档数据的载体，操作包含：查询是否存在、创建索引、删除索引、获取索引映射。

3.1 索引字段类型说明

日常开发最常用两种字段类型：

• text（分词文本类型）：支持分词、全文模糊检索，不支持精确匹配、聚合排序。适用于用户名、简介、文章内容、地址描述等需要模糊搜索的文本场景，检索时会根据词条重合度计算相关性分数。

• keyword（精确文本类型）：不分词、完整字符串匹配，支持精确查询、聚合、排序、过滤。适用于用户ID、手机号、分类标签、城市名称、状态码等不需要模糊匹配的固定文本字段。

• integer（整型数值）：存储整数数字，支持数值区间筛选、大小排序、聚合统计。适用于年龄、浏览量、点赞数、状态数值等整型业务字段。

• long（长整型）：超大整数存储，取值范围远大于integer，适用于订单号、日志ID、超大计数等数值过大的场景。

• float/double（浮点型）：存储小数数值，float精度较低、double高精度，适用于价格、评分、比例、经纬度小数数据。

• boolean（布尔类型）：仅存储 true/false，适用于是否启用、是否删除、是否VIP等二元状态字段。

• date（时间日期类型）：专用时间类型，支持时间范围筛选、时间排序，适配yyyy-MM-dd、yyyy-MM-dd HH:mm:ss、时间戳等格式，比字符串存储更节省空间、查询效率更高。

• array（数组类型）：无单独声明类型，字段写入数组即可自动识别，数组内元素类型统一，适用于用户标签、多分类、多角色等多值场景。

• object（对象类型）：存储嵌套JSON对象，支持多层级字段映射，适用于用户拓展信息、地址详情、设备信息等结构化嵌套数据。

3.2 完整索引操作代码

async def index_operate():
    idx_name = "user_test"  # 定义索引名称

    async with get_es_client() as client:
        # 1. 判断索引是否存在
        index_exists = await client.indices.exists(index=idx_name)
        print(f"索引是否存在: {index_exists}")

        # 2. 不存在则创建索引 + 自定义映射
        if not index_exists:
            await client.indices.create(
                index=idx_name,
                mappings={
                    "properties": {
                        "user_id": {"type": "keyword"},    # 精确匹配
                        "username": {"type": "text"},      # 模糊检索
                        "age": {"type": "integer"},        # 数值类型
                        "city": {"type": "keyword"}        # 城市精确筛选
                    }
                }
            )
            print(f"✅ 索引【{idx_name}】创建成功")

        # 3. 获取索引映射结构
        mapping = await client.indices.get_mapping(index=idx_name)
        print("📋 索引结构：", mapping["user_test"]["mappings"]["properties"])

        # 4. 删除索引（谨慎使用！）
        # await client.indices.delete(index=idx_name)
        # print(f"❌ 索引【{idx_name}】删除成功")

if __name__ == "__main__":
    asyncio.run(index_operate())

四、文档（Document）CRUD 操作

ES 文档对应 MySQL 的一行数据，是 JSON 格式数据，核心操作：新增、查询、修改、删除、批量插入。

4.1 单条文档新增/查询/修改/删除

async def doc_crud():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 1. 新增文档（指定文档id，不指定则ES自动生成）
        doc_data = {
            "user_id": "u001",
            "username": "王五",
            "age": 18,
            "city": "北京"
        }
        # 创建/覆盖文档
        await client.index(index=idx_name, id="1", document=doc_data)
        print("✅ 文档新增/更新成功")

        # 2. 根据ID查询单条文档
        res = await client.get(index=idx_name, id="1")
        print("📄 单条文档数据：", res["_source"])

        # 3. 更新文档（局部更新，只修改指定字段）
        update_data = {"doc": {"age": 19, "city": "上海"}}
        await client.update(index=idx_name, id="1", doc=update_data["doc"])
        print("✅ 文档局部更新成功")

        # 4. 删除单条文档
        # await client.delete(index=idx_name, id="1")
        # print("❌ 文档删除成功")

if __name__ == "__main__":
    asyncio.run(doc_crud())

4.2 批量插入文档（高效批量写入）

批量插入适合初始化数据、批量同步业务数据，比单条插入性能提升数十倍：

from elasticsearch.helpers import async_bulk

async def batch_insert_doc():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 构造批量数据
        bulk_data = [
            {"_index": idx_name, "_id": 2, "_source": {"user_id": "u002", "username": "张三", "age": 20, "city": "北京"}},
            {"_index": idx_name, "_id": 3, "_source": {"user_id": "u003", "username": "李四", "age": 22, "city": "广州"}},
            {"_index": idx_name, "_id": 4, "_source": {"user_id": "u004", "username": "王五", "age": 25, "city": "深圳"}}
        ]
        # 异步批量插入
        success, fail = await async_bulk(client=client, actions=bulk_data)
        print(f"✅ 批量插入成功：{success} 条，失败：{fail} 条")

if __name__ == "__main__":
    asyncio.run(batch_insert_doc())

五、ES 核心查询操作（最全业务场景）

本章覆盖企业开发100%常用ES查询语法，包含基础查询、精准查询、模糊查询、组合条件、高亮、排序、分页、去重、聚合统计等所有核心能力，适配搜索、筛选、统计、后台列表等各类业务场景，所有异步代码开箱即用。

这是开发最常用的部分，包含：全量查询、模糊检索、精确匹配、条件筛选、否定查询、分页查询，同时解答你之前的「多字少字能否查询」问题。

5.1 全量查询（查询所有数据）

async def search_all():
    idx_name = "user_test"
    async with get_es_client() as client:
        res = await client.search(
            index=idx_name,
            query={"match_all": {}}  # 匹配所有文档
        )
        # 解析结果
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("📋 所有数据：", result_list)

if __name__ == "__main__":
    asyncio.run(search_all())

5.2 全文模糊查询（match 分词查询）

核心特性：文本字段分词匹配，搜索词多字、少字、部分匹配都能查到，适合搜索内容模糊匹配。

例：文档内容「王五」，搜索「王五测试」也能匹配到（部分词条重合）

async def search_match():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 模糊匹配username字段
        res = await client.search(
            index=idx_name,
            query={
                "match": {
                    "username": "王五测试"
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 模糊查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_match())

5.3 精确匹配查询（term 关键字查询）

核心特性：针对 keyword 字段，严格完整匹配字符串，多字少字都查不到，适合ID、城市、状态筛选。

async def search_term():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 精确匹配city=北京
        res = await client.search(
            index=idx_name,
            query={
                "term": {
                    "city": "北京"
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 精确查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_term())

5.4 条件筛选 + 否定查询（must_not）

解决你之前的「不在某地、不包含某内容」查询需求，ES 不识别自然语言「不」，需手动用 must_not 实现否定筛选。

async def search_must_not():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 查询：城市 不是北京 的所有用户
        res = await client.search(
            index=idx_name,
            query={
                "bool": {
                    "must_not": [
                        {"term": {"city": "北京"}}
                    ]
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 非北京用户：", result_list)

if __name__ == "__main__":
    asyncio.run(search_must_not())

5.5 分页查询 + 数值范围筛选

async def search_page():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 分页参数：第1页，每页2条
        page = 1
        size = 2
        res = await client.search(
            index=idx_name,
            from_=(page - 1) * size,
            size=size,
            query={
                "bool": {
                    "filter": [  # filter无打分、不影响相关性，查询性能更高
                        {"range": {"age": {"gte": 20, "lte": 30}}}  # 年龄20-30区间
                    ]
                }
            }
        )
        total = res["hits"]["total"]["value"]
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print(f"✅ 总条数：{total}，分页数据：", result_list)

if __name__ == "__main__":
    asyncio.run(search_page())

5.6 短语精确匹配查询（match_phrase）

核心特性：要求搜索词条连续、顺序一致，无错乱、无遗漏，严格匹配完整短句，解决你之前「多字少字查不到」的精准场景。

async def search_match_phrase():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 精准匹配连续短语，顺序必须一致
        res = await client.search(
            index=idx_name,
            query={
                "match_phrase": {
                    "username": "王五"
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 短语精确查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_match_phrase())

5.7 多字段联合查询（multi_match）

核心特性：同时对多个text字段模糊检索，只需匹配任意一个字段即可命中，适合全站搜索场景。

async def search_multi_match():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 同时匹配 username、city 两个字段
        res = await client.search(
            index=idx_name,
            query={
                "multi_match": {
                    "query": "北京 王五",
                    "fields": ["username", "city"]  # 指定查询字段
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 多字段联合查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_multi_match())

5.8 模糊纠错查询（fuzzy）

核心特性：支持错别字、少字、多字纠错匹配，适合用户输入错误的搜索场景（例：输入“王武”匹配“王五”）。

async def search_fuzzy():
    idx_name = "user_test"
    async with get_es_client() as client:
        res = await client.search(
            index=idx_name,
            query={
                "fuzzy": {
                    "username": {
                        "value": "王武",  # 错误输入
                        "fuzziness": 1  # 允许1个字符误差
                    }
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 模糊纠错查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_fuzzy())

5.9 通配符查询（wildcard）

核心特性：支持占位符匹配，*匹配任意多个字符，?匹配单个字符，仅适用于keyword/text字段。

async def search_wildcard():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 匹配所有 王开头 的用户名
        res = await client.search(
            index=idx_name,
            query={
                "wildcard": {
                    "username": "王*"
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 通配符查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_wildcard())

5.10 多条件组合查询（must + must_not + should + filter）

Bool查询四剑客，企业最常用组合查询：

• must：必须匹配，参与相关性打分

• must_not：必须不匹配，不打分

• should：可选匹配，匹配到加分，不匹配不影响

• filter：必须匹配，无打分、性能最优

async def search_bool_all():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 需求：年龄20-30、不是广州、用户名包含王、优先北京用户
        res = await client.search(
            index=idx_name,
            query={
                "bool": {
                    "must": [{"match": {"username": "王"}}],
                    "must_not": [{"term": {"city": "广州"}}],
                    "should": [{"term": {"city": "北京"}}],
                    "filter": [{"range": {"age": {"gte": 20, "lte": 30}}}]
                }
            }
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 多条件组合查询结果：", result_list)

if __name__ == "__main__":
    asyncio.run(search_bool_all())

5.11 搜索结果高亮显示（highlight）

核心特性：自动给匹配关键词添加标签高亮，前端直接渲染，文章搜索必备功能。

async def search_highlight():
    idx_name = "user_test"
    async with get_es_client() as client:
        res = await client.search(
            index=idx_name,
            query={"match": {"username": "王五"}},
            highlight={
                "fields": ["username"],  # 高亮字段
                "pre_tags": ["<span style='color:red'>"],
                "post_tags": ["</span>"]
            }
        )
        # 解析高亮结果
        for hit in res["hits"]["hits"]:
            print("原数据：", hit["_source"])
            print("高亮关键词：", hit.get("highlight", {}))

if __name__ == "__main__":
    asyncio.run(search_highlight())

5.12 结果排序查询（sort）

支持字段升序、降序排序，支持多字段优先级排序。

async def search_sort():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 先按年龄降序，年龄相同按id升序
        res = await client.search(
            index=idx_name,
            sort=[
                {"age": "desc"},
                {"user_id": "asc"}
            ],
            query={"match_all": {}}
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 排序后数据：", result_list)

if __name__ == "__main__":
    asyncio.run(search_sort())

5.13 去重查询（collapse）

核心特性：根据指定字段去重，保留每条唯一数据，适配重复数据过滤场景。

async def search_collapse():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 根据city字段去重，每个城市只保留一条数据
        res = await client.search(
            index=idx_name,
            collapse={"field": "city"},
            query={"match_all": {}}
        )
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print("🔍 去重后数据：", result_list)

if __name__ == "__main__":
    asyncio.run(search_collapse())

5.14 简单聚合查询（分组统计）

实现分组计数、平均值、最值统计，替代数据库group by，适配大数据统计场景。

async def search_agg():
    idx_name = "user_test"
    async with get_es_client() as client:
        res = await client.search(
            index=idx_name,
            size=0,  # 不需要原始数据，只看统计结果
            aggs={
                # 按城市分组统计数量
                "group_by_city": {
                    "terms": {"field": "city"}
                },
                # 年龄平均值、最大最小值
                "age_stat": {
                    "stats": {"field": "age"}
                }
            }
        )
        print("📊 城市分组统计：", res["aggregations"]["group_by_city"]["buckets"])
        print("📊 年龄统计：", res["aggregations"]["age_stat"])

if __name__ == "__main__":
    asyncio.run(search_agg())

async def search_page():
    idx_name = "user_test"
    async with get_es_client() as client:
        # 分页参数：第1页，每页2条
        page = 1
        size = 2
        res = await client.search(
            index=idx_name,
            from_=(page - 1) * size,
            size=size,
            query={
                "bool": {
                    "filter": [  # filter无打分，查询性能更高
                        {"range": {"age": {"gte": 20}}}  # 年龄>=20
                    ]
                }
            }
        )
        total = res["hits"]["total"]["value"]
        result_list = [hit["_source"] for hit in res["hits"]["hits"]]
        print(f"✅ 总条数：{total}，分页数据：", result_list)

if __name__ == "__main__":
    asyncio.run(search_page())

5.15 搜索匹配核心规则总结（解答多字/少字匹配问题）

• text字段 + match查询：部分词条重合即可命中，多字、少字、语序错乱均可查到，分数随匹配度变化

• text字段 + match_phrase短语查询：必须词条连续、顺序完全一致，轻微差异即查不到，用于精准短句匹配

• keyword字段 + term查询：严格全量字符串匹配，多字少字、符号差异均无法命中，用于精确筛选

• fuzzy模糊查询：支持错别字、字符误差匹配，适配用户输入错误场景

• wildcard通配符查询：支持前后模糊匹配，适配前缀、后缀检索

六、全文总结

1. 异步 ES 核心优势：高并发不阻塞，适配异步 Python 项目，生产环境首选AsyncElasticsearch + async with；

2. 索引操作核心是字段映射区分 text/keyword，模糊查 text、精确查 keyword；

3. 文档操作优先使用批量写入，提升性能，局部更新使用 update 避免覆盖全量数据；

4. 查询核心：match 模糊、term 精确、must_not 否定、filter 条件筛选，覆盖 90% 业务场景。

（注：部分内容可能由 AI 生成）