Python Elasticsearch全文检索实战

引言

在信息爆炸的时代，如何让用户快速找到所需内容是每个应用面临的核心挑战。电商平台的商品搜索、文档站点的知识检索、日志系统的全文分析——这些场景都需要强大的全文检索引擎支撑。Elasticsearch作为基于Lucene构建的分布式搜索和分析引擎，以其近实时的搜索性能、丰富的查询DSL、强大的聚合分析能力和水平扩展能力，成为全文检索领域的事实标准。

在Python生态中，elasticsearch-py是官方提供的客户端库，它完整封装了Elasticsearch的REST API，提供了Pythonic的接口。无论是索引管理、文档CRUD、复杂搜索DSL还是聚合分析，elasticsearch-py都能以简洁的Python字典语法表达。配合elasticsearch-dsl库，还可以使用更高级的声明式查询构建器，进一步提升开发体验。

本文将从elasticsearch-py客户端的基本使用出发，系统讲解索引管理、文档CRUD操作、搜索DSL查询、分词器配置、搜索结果高亮、聚合分析以及批量操作，通过完整的代码示例帮助读者构建生产级的全文检索系统。

核心原理

倒排索引

Elasticsearch的核心是倒排索引（Inverted Index），这是全文检索的基础数据结构。传统数据库使用B+树索引，通过行找到列值；倒排索引则反过来，通过词项（Term）找到包含该词项的文档列表。当文档被索引时，Elasticsearch会对文本进行分词（Analysis），将文本拆分为词项，然后建立词项到文档ID的映射表。搜索时，同样对查询文本进行分词，然后在倒排索引中查找匹配的文档。

倒排索引由两部分组成：词项字典（Term Dictionary）和倒排表（Postings List）。词项字典存储所有出现过的词项及其在倒排表中的位置，通常使用FST（Finite State Transducer）压缩存储，内存占用极小。倒排表记录每个词项出现在哪些文档中、出现的位置和频率等信息。这种结构使得包含某个词的文档查找复杂度为O(1)，是全文检索高效的根本原因。

分析器与分词

分析器（Analyzer）是Elasticsearch文本处理的核心组件，负责将原始文本转换为可索引的词项。分析器由三部分组成：字符过滤器（Character Filter）、分词器（Tokenizer）和词项过滤器（Token Filter）。字符过滤器在分词前处理文本，如去除HTML标签；分词器将文本拆分为词项，如按空格拆分；词项过滤器对分词结果进行后处理，如转小写、去停用词、词干提取等。

Elasticsearch内置了多种分析器。Standard Analyzer是默认分析器，使用Unicode文本分割算法，支持中英文混合分词（但中文分词效果有限，通常需要安装IK分词器）。Simple Analyzer按非字母字符分词并转小写。Whitespace Analyzer仅按空格分词。Keyword Analyzer将整个文本作为单个词项，不做分词。对于中文，IK分词器提供了细粒度（ik_smart）和智能（ik_max_word）两种分词模式，是中文全文检索的标配。

分布式架构

Elasticsearch是分布式系统，数据分布在多个节点上。索引（Index）被分为多个分片（Shard），每个分片是一个独立的Lucene索引。分片分为主分片（Primary Shard）和副本分片（Replica Shard），主分片负责读写，副本分片负责读和容灾。当主分片所在节点故障时，副本分片会被提升为新的主分片，保证服务可用性。

文档路由是分片的核心机制。Elasticsearch使用公式shard = hash(routing) % number_of_primary_shards确定文档存储在哪个分片。默认routing值为文档ID，也可以自定义routing值。这意味着主分片数量在索引创建后不能修改——因为修改后路由公式结果会变化，导致文档找不到。因此，创建索引时需要根据数据量预估合理的分片数量。

代码实战

安装与连接

# 安装命令（终端执行）
# pip install elasticsearch>=8.0.0
# pip install elasticsearch-dsl  # 高级DSL查询库

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
from datetime import datetime
import json

# 创建客户端连接
es = Elasticsearch(
    ["localhost:9200"],
    # Elasticsearch 8.x默认开启安全认证，本地开发可配置
    # basic_auth=("elastic", "your_password"),
    # verify_certs=False,  # 自签名证书时关闭验证
    request_timeout=30,
    max_retries=3,
    retry_on_timeout=True,
    # 连接池配置
    # maxsize=20,  # 最大连接数
)

# 测试连接
if es.ping():
    info = es.info()
    print(f"Elasticsearch版本: {info['version']['number']}")
    print(f"集群名称: {info['cluster_name']}")
else:
    print("连接失败")

索引管理

# 索引定义 - 包含映射和分析器配置
index_settings = {
    "settings": {
        "number_of_shards": 3,  # 主分片数
        "number_of_replicas": 1,  # 副本数
        "refresh_interval": "1s",  # 刷新间隔
        "analysis": {
            "analyzer": {
                "my_analyzer": {
                    "type": "custom",
                    "tokenizer": "ik_max_word",  # 中文分词器（需安装IK插件）
                    "filter": ["lowercase", "my_stop"],
                },
                "pinyin_analyzer": {
                    "type": "custom",
                    "tokenizer": "pinyin_tokenizer",  # 拼音分词（需安装pinyin插件）
                },
            },
            "filter": {
                "my_stop": {
                    "type": "stop",
                    "stopwords": ["的", "了", "是", "在", "我", "有"],
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ik_max_word",
                "search_analyzer": "ik_smart",  # 搜索时使用更细粒度的分词
                "fields": {
                    "keyword": {  # 子字段，用于精确匹配和排序
                        "type": "keyword",
                        "ignore_above": 256,
                    },
                    "pinyin": {
                        "type": "text",
                        "analyzer": "pinyin_analyzer",
                    }
                }
            },
            "content": {
                "type": "text",
                "analyzer": "ik_max_word",
            },
            "category": {
                "type": "keyword",  # keyword类型不分词，用于精确匹配和聚合
            },
            "tags": {
                "type": "keyword",
            },
            "author": {
                "type": "keyword",
            },
            "view_count": {
                "type": "integer",
            },
            "like_count": {
                "type": "integer",
            },
            "price": {
                "type": "float",
            },
            "publish_date": {
                "type": "date",
                "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis",
            },
            "is_published": {
                "type": "boolean",
            },
            "location": {
                "type": "geo_point",  # 地理坐标类型
            },
            "suggest": {
                "type": "completion",  # 自动补全类型
                "analyzer": "ik_max_word",
            }
        }
    }
}

# 创建索引
INDEX_NAME = "articles"
if not es.indices.exists(index=INDEX_NAME):
    es.indices.create(index=INDEX_NAME, body=index_settings)
    print(f"索引 {INDEX_NAME} 创建成功")
else:
    print(f"索引 {INDEX_NAME} 已存在")

# 查看索引信息
index_info = es.indices.get(index=INDEX_NAME)
print(f"索引设置: {json.dumps(index_info[INDEX_NAME]['settings']['index'], indent=2, ensure_ascii=False)}")

# 查看映射
mapping = es.indices.get_mapping(index=INDEX_NAME)
print(f"索引映射: {json.dumps(mapping[INDEX_NAME]['mappings'], indent=2, ensure_ascii=False)}")

# 更新映射（只能添加新字段，不能修改已有字段）
es.indices.put_mapping(
    index=INDEX_NAME,
    body={
        "properties": {
            "summary": {
                "type": "text",
                "analyzer": "ik_max_word",
            }
        }
    }
)

# 索引统计
stats = es.indices.stats(index=INDEX_NAME)
print(f"文档数: {stats['indices'][INDEX_NAME]['primaries']['docs']['count']}")
print(f"存储大小: {stats['indices'][INDEX_NAME]['primaries']['store']['size_in_bytes'] / 1024:.1f}KB")

# 删除索引
# es.indices.delete(index=INDEX_NAME)

文档CRUD

# 索引文档（插入/更新）
doc1 = {
    "title": "Python异步编程完全指南",
    "content": "asyncio是Python标准库中的异步IO框架，提供了事件循环、协程、任务和Future等核心组件。通过async/await语法，开发者可以编写高效的异步代码，处理大量并发IO操作。",
    "category": "编程技术",
    "tags": ["Python", "异步", "asyncio", "并发"],
    "author": "dev_coder",
    "view_count": 1500,
    "like_count": 320,
    "price": 0.0,
    "publish_date": "2024-06-15 10:30:00",
    "is_published": True,
    "location": {"lat": 39.9042, "lon": 116.4074},
    "suggest": {"input": ["Python异步编程", "asyncio指南", "异步编程"]},
}
result = es.index(index=INDEX_NAME, id="article_001", document=doc1)
print(f"文档索引成功: {result['result']}, ID: {result['_id']}")

doc2 = {
    "title": "Elasticsearch全文检索实战教程",
    "content": "Elasticsearch是基于Lucene的分布式搜索引擎，支持全文检索、结构化搜索、地理位置搜索和聚合分析。倒排索引是其核心数据结构，通过分词器将文本拆分为词项建立索引。",
    "category": "数据库技术",
    "tags": ["Elasticsearch", "搜索", "全文检索", "Lucene"],
    "author": "search_expert",
    "view_count": 2800,
    "like_count": 580,
    "price": 29.9,
    "publish_date": "2024-07-20 14:00:00",
    "is_published": True,
    "location": {"lat": 31.2304, "lon": 121.4737},
    "suggest": {"input": ["Elasticsearch教程", "全文检索", "搜索引擎"]},
}
es.index(index=INDEX_NAME, id="article_002", document=doc2)

doc3 = {
    "title": "Redis缓存设计与最佳实践",
    "content": "Redis是高性能内存数据库，支持字符串、哈希、列表、集合、有序集合等数据结构。缓存穿透、缓存雪崩、缓存击穿是缓存系统的三大经典问题，需要通过空值缓存、随机过期、互斥锁等策略解决。",
    "category": "数据库技术",
    "tags": ["Redis", "缓存", "NoSQL", "分布式"],
    "author": "dev_coder",
    "view_count": 3200,
    "like_count": 750,
    "price": 19.9,
    "publish_date": "2024-08-01 09:00:00",
    "is_published": True,
    "location": {"lat": 39.9042, "lon": 116.4074},
    "suggest": {"input": ["Redis缓存", "缓存设计", "NoSQL"]},
}
es.index(index=INDEX_NAME, id="article_003", document=doc3)

doc4 = {
    "title": "Docker容器化部署指南",
    "content": "Docker通过容器技术实现应用的轻量级隔离和部署。Dockerfile定义镜像构建过程，docker-compose编排多容器应用。容器化部署提升了应用的可移植性和一致性。",
    "category": "运维技术",
    "tags": ["Docker", "容器", "部署", "DevOps"],
    "author": "ops_master",
    "view_count": 950,
    "like_count": 180,
    "price": 0.0,
    "publish_date": "2024-05-10 16:00:00",
    "is_published": True,
    "location": {"lat": 22.5431, "lon": 114.0579},
    "suggest": {"input": ["Docker部署", "容器化", "DevOps"]},
}
es.index(index=INDEX_NAME, id="article_004", document=doc4)

# 刷新索引（使文档立即可搜索，默认1秒自动刷新）
es.indices.refresh(index=INDEX_NAME)

# 获取文档
doc = es.get(index=INDEX_NAME, id="article_001")
print(f"\n获取文档: {doc['_source']['title']}")

# 检查文档是否存在
exists = es.exists(index=INDEX_NAME, id="article_001")
print(f"文档存在: {exists}")

# 更新文档（部分更新）
es.update(
    index=INDEX_NAME,
    id="article_001",
    body={
        "doc": {
            "view_count": 1600,
            "summary": "本文全面讲解Python异步编程的核心概念和实战技巧。",
        }
    }
)

# 脚本更新（使用Painless脚本）
es.update(
    index=INDEX_NAME,
    id="article_001",
    body={
        "script": {
            "source": "ctx._source.view_count += params.count; ctx._source.last_viewed = params.time",
            "lang": "painless",
            "params": {
                "count": 10,
                "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            }
        }
    }
)

# 删除文档
# es.delete(index=INDEX_NAME, id="article_004")

# 按查询删除
# es.delete_by_query(index=INDEX_NAME, body={"query": {"match_all": {}}})

搜索DSL查询

# 1. 全文检索 - match查询
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "match": {
                "title": "Python异步编程"
            }
        }
    }
)
print(f"\n全文检索结果:")
for hit in result["hits"]["hits"]:
    score = hit["_score"]
    title = hit["_source"]["title"]
    print(f"  [{score:.2f}] {title}")

# 2. 多字段搜索 - multi_match
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "multi_match": {
                "query": "缓存 分布式",
                "fields": ["title^3", "content^1", "tags^2"],  # ^N表示权重提升
                "type": "best_fields",  # 取最高分
            }
        },
        "size": 5,
    }
)
print(f"\n多字段搜索:")
for hit in result["hits"]["hits"]:
    print(f"  [{hit['_score']:.2f}] {hit['_source']['title']}")

# 3. 精确匹配 - term查询
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "term": {
                "category": "数据库技术"  # keyword字段精确匹配
            }
        }
    }
)
print(f"\n精确匹配（数据库技术）:")
for hit in result["hits"]["hits"]:
    print(f"  {hit['_source']['title']}")

# 4. 多值匹配 - terms查询
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "terms": {
                "author": ["dev_coder", "search_expert"]
            }
        }
    }
)

# 5. 范围查询 - range
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "range": {
                "view_count": {
                    "gte": 1000,
                    "lte": 3000,
                }
            }
        },
        "sort": [{"view_count": {"order": "desc"}}],
    }
)
print(f"\n浏览量1000-3000的文章:")
for hit in result["hits"]["hits"]:
    print(f"  {hit['_source']['title']} (浏览:{hit['_source']['view_count']})")

# 6. 布尔查询 - bool（must/should/must_not/filter）
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "bool": {
                "must": [
                    {"match": {"content": "分布式"}}
                ],
                "filter": [
                    {"term": {"is_published": True}},
                    {"range": {"view_count": {"gte": 500}}},
                ],
                "must_not": [
                    {"term": {"category": "运维技术"}}
                ],
                "should": [
                    {"match": {"tags": "Redis"}},
                ],
                "minimum_should_match": 0,
            }
        }
    }
)
print(f"\n布尔查询结果:")
for hit in result["hits"]["hits"]:
    print(f"  [{hit['_score']:.2f}] {hit['_source']['title']}")

# 7. 短语匹配 - match_phrase
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "match_phrase": {
                "content": "倒排索引"  # 要求词项按顺序连续出现
            }
        }
    }
)

# 8. 前缀查询 - prefix
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "prefix": {
                "title.keyword": "Python"
            }
        }
    }
)

# 9. 通配符查询 - wildcard
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "wildcard": {
                "author": "dev*"
            }
        }
    }
)

# 10. 模糊查询 - fuzzy（允许编辑距离）
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "fuzzy": {
                "title": {
                    "value": "Pythn",  # 模糊匹配"Python"
                    "fuzziness": "AUTO",
                }
            }
        }
    }
)

# 11. 地理查询 - geo_distance
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "geo_distance": {
                "distance": "1000km",
                "location": {"lat": 39.9, "lon": 116.4},
            }
        }
    }
)
print(f"\n1000km范围内的文章:")
for hit in result["hits"]["hits"]:
    print(f"  {hit['_source']['title']}")

# 12. 分页查询 - from/size
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {"match_all": {}},
        "from": 0,
        "size": 2,
        "sort": [{"publish_date": {"order": "desc"}}],
    }
)
print(f"\n分页查询（第1页）:")
for hit in result["hits"]["hits"]:
    print(f"  {hit['_source']['title']} ({hit['_source']['publish_date']})")

# 13. 搜索结果高亮
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {
            "match": {"content": "缓存"}
        },
        "highlight": {
            "fields": {
                "content": {
                    "pre_tags": ["<em class='highlight'>"],
                    "post_tags": ["</em>"],
                    "fragment_size": 150,
                    "number_of_fragments": 3,
                },
                "title": {}
            }
        }
    }
)
print(f"\n搜索高亮:")
for hit in result["hits"]["hits"]:
    print(f"  标题: {hit['highlight'].get('title', [hit['_source']['title']])[0]}")
    if "content" in hit["highlight"]:
        print(f"  内容: {hit['highlight']['content'][0]}")

# 14. 自动补全建议
result = es.search(
    index=INDEX_NAME,
    body={
        "suggest": {
            "my_suggest": {
                "prefix": "Py",
                "completion": {
                    "field": "suggest",
                    "size": 5,
                    "fuzzy": {
                        "fuzziness": "AUTO"
                    }
                }
            }
        }
    }
)
print(f"\n自动补全建议:")
for option in result["suggest"]["my_suggest"][0]["options"]:
    print(f"  {option['_source']['title']}")

# 15. 源字段过滤
result = es.search(
    index=INDEX_NAME,
    body={
        "query": {"match_all": {}},
        "_source": ["title", "author", "view_count"],  # 只返回指定字段
        # "_source": {"excludes": ["content"]},  # 排除指定字段
    }
)

聚合分析

# 1. 桶聚合 - 按分类分组统计
result = es.search(
    index=INDEX_NAME,
    body={
        "size": 0,  # 不返回文档，只返回聚合结果
        "aggs": {
            "by_category": {
                "terms": {
                    "field": "category",
                    "size": 10,
                },
                "aggs": {
                    "avg_views": {"avg": {"field": "view_count"}},
                    "total_likes": {"sum": {"field": "like_count"}},
                }
            }
        }
    }
)
print(f"\n分类聚合:")
for bucket in result["aggregations"]["by_category"]["buckets"]:
    print(f"  {bucket['key']}: {bucket['doc_count']}篇, "
          f"平均浏览{bucket['avg_views']['value']:.0f}, "
          f"总点赞{bucket['total_likes']['value']}")

# 2. 按作者统计
result = es.search(
    index=INDEX_NAME,
    body={
        "size": 0,
        "aggs": {
            "by_author": {
                "terms": {"field": "author"},
                "aggs": {
                    "categories": {
                        "terms": {"field": "category"}
                    },
                    "max_views": {"max": {"field": "view_count"}},
                }
            }
        }
    }
)
print(f"\n作者聚合:")
for bucket in result["aggregations"]["by_author"]["buckets"]:
    cats = [c["key"] for c in bucket["categories"]["buckets"]]
    print(f"  {bucket['key']}: {bucket['doc_count']}篇, 分类{cats}, "
          f"最高浏览{bucket['max_views']['value']}")

# 3. 日期直方图聚合
result = es.search(
    index=INDEX_NAME,
    body={
        "size": 0,
        "aggs": {
            "by_month": {
                "date_histogram": {
                    "field": "publish_date",
                    "calendar_interval": "month",
                    "format": "yyyy-MM",
                },
                "aggs": {
                    "total_views": {"sum": {"field": "view_count"}},
                }
            }
        }
    }
)
print(f"\n按月统计:")
for bucket in result["aggregations"]["by_month"]["buckets"]:
    print(f"  {bucket['key_as_string']}: {bucket['doc_count']}篇, "
          f"总浏览{bucket['total_views']['value']}")

# 4. 范围聚合
result = es.search(
    index=INDEX_NAME,
    body={
        "size": 0,
        "aggs": {
            "view_ranges": {
                "range": {
                    "field": "view_count",
                    "ranges": [
                        {"to": 1000},
                        {"from": 1000, "to": 2000},
                        {"from": 2000, "to": 3000},
                        {"from": 3000},
                    ]
                }
            }
        }
    }
)
print(f"\n浏览量范围:")
for bucket in result["aggregations"]["view_ranges"]["buckets"]:
    print(f"  {bucket['key']}: {bucket['doc_count']}篇")

# 5. 统计聚合
result = es.search(
    index=INDEX_NAME,
    body={
        "size": 0,
        "aggs": {
            "view_stats": {"stats": {"field": "view_count"}},
            "price_stats": {"stats": {"field": "price"}},
        }
    }
)
print(f"\n统计信息:")
print(f"  浏览量: {result['aggregations']['view_stats']}")
print(f"  价格: {result['aggregations']['price_stats']}")

# 6. 嵌套聚合 - 按分类统计标签
result = es.search(
    index=INDEX_NAME,
    body={
        "size": 0,
        "aggs": {
            "categories": {
                "terms": {"field": "category"},
                "aggs": {
                    "tags": {
                        "terms": {"field": "tags", "size": 5}
                    }
                }
            }
        }
    }
)
print(f"\n分类标签聚合:")
for cat_bucket in result["aggregations"]["categories"]["buckets"]:
    tags = [f"{t['key']}({t['doc_count']})" for t in cat_bucket["tags"]["buckets"]]
    print(f"  {cat_bucket['key']}: {tags}")

批量操作

# 批量索引文档
def generate_docs(count=100):
    """生成批量文档"""
    categories = ["编程技术", "数据库技术", "运维技术", "前端技术", "AI技术"]
    authors = ["dev_coder", "search_expert", "ops_master", "ai_researcher"]
    for i in range(count):
        yield {
            "_index": INDEX_NAME,
            "_id": f"batch_{i:04d}",
            "_source": {
                "title": f"批量文章第{i+1}篇",
                "content": f"这是第{i+1}篇批量索引的测试文章，内容涉及{categories[i % len(categories)]}领域。",
                "category": categories[i % len(categories)],
                "tags": [categories[i % len(categories)], "批量测试"],
                "author": authors[i % len(authors)],
                "view_count": i * 10,
                "like_count": i * 2,
                "price": float(i) * 0.5,
                "publish_date": datetime(2024, 1, 1).strftime("%Y-%m-%d %H:%M:%S"),
                "is_published": True,
            }
        }

# 使用bulk API批量索引
success_count, errors = bulk(es, generate_docs(100), raise_on_error=False)
print(f"\n批量索引: 成功{success_count}条, 错误{len(errors)}条")

es.indices.refresh(index=INDEX_NAME)
print(f"当前文档总数: {es.count(index=INDEX_NAME)['count']}")

# 批量获取
result = es.mget(
    index=INDEX_NAME,
    body={
        "ids": ["article_001", "article_002", "article_003"]
    }
)
print(f"\n批量获取:")
for doc in result["docs"]:
    if doc["found"]:
        print(f"  {doc['_source']['title']}")

# 批量更新
bulk_body = [
    {"_op_type": "update", "_index": INDEX_NAME, "_id": "batch_0000",
     "doc": {"view_count": 9999}},
    {"_op_type": "update", "_index": INDEX_NAME, "_id": "batch_0001",
     "doc": {"view_count": 8888}},
    {"_op_type": "update", "_index": INDEX_NAME, "_id": "batch_0002",
     "doc": {"view_count": 7777}},
]
bulk(es, bulk_body, raise_on_error=False)

# 批量删除
bulk_delete = [
    {"_op_type": "delete", "_index": INDEX_NAME, "_id": f"batch_{i:04d}"}
    for i in range(90, 100)
]
bulk(es, bulk_delete, raise_on_error=False)

# 使用scroll API遍历大量数据
def scroll_search(index_name, query=None, scroll_size=1000):
    """使用scroll API遍历所有匹配文档"""
    if query is None:
        query = {"match_all": {}}
    
    result = es.search(
        index=index_name,
        body={"query": query, "size": scroll_size},
        scroll="2m",  # scroll上下文保持2分钟
    )
    
    scroll_id = result["_scroll_id"]
    total = result["hits"]["total"]["value"]
    hits = result["hits"]["hits"]
    
    count = len(hits)
    while hits:
        for hit in hits:
            yield hit
        result = es.scroll(scroll_id=scroll_id, scroll="2m")
        scroll_id = result["_scroll_id"]
        hits = result["hits"]["hits"]
        count += len(hits)
    
    # 清理scroll上下文
    es.clear_scroll(scroll_id=scroll_id)

# 使用scroll遍历
print(f"\nScroll遍历:")
for i, hit in enumerate(scroll_search(INDEX_NAME, scroll_size=50)):
    if i < 5:
        print(f"  {hit['_id']}: {hit['_source']['title']}")
    if i >= 4:
        print(f"  ... (更多结果省略)")
        break

分词器测试

# 测试分词器效果
def test_analyzer(text, analyzer="ik_max_word"):
    """测试分词器对文本的分词结果"""
    result = es.indices.analyze(
        body={
            "analyzer": analyzer,
            "text": text,
        }
    )
    tokens = [token["token"] for token in result["tokens"]]
    return tokens

# 测试不同分词器
test_text = "Python异步编程完全指南"
print(f"\n分词器测试:")
print(f"  原文: {test_text}")
print(f"  ik_max_word: {test_analyzer(test_text, 'ik_max_word')}")
print(f"  ik_smart: {test_analyzer(test_text, 'ik_smart')}")
print(f"  standard: {test_analyzer(test_text, 'standard')}")

test_text2 = "Elasticsearch全文检索引擎"
print(f"\n  原文: {test_text2}")
print(f"  ik_max_word: {test_analyzer(test_text2, 'ik_max_word')}")

# 测试自定义分析器
test_text3 = "这是一个测试的句子，包含了一些停用词"
print(f"\n  原文: {test_text3}")
print(f"  my_analyzer: {test_analyzer(test_text3, 'my_analyzer')}")

进阶技巧

搜索相关性优化

搜索结果的相关性排序是全文检索的核心挑战。Elasticsearch默认使用BM25算法计算文档得分，考虑词频（TF）、逆文档频率（IDF）和文档长度归一化。优化相关性的方法包括：使用boost参数调整字段权重（如标题权重高于内容），使用function_score查询根据业务规则调整得分（如按浏览量、发布时间、用户偏好加权），使用rescore在初始检索后对top-N结果重新打分。

同义词扩展是提升搜索召回率的有效手段。通过配置同义词过滤器，将"手机"和"移动电话"视为同义词，搜索任一词都能匹配两者。同义词可以在索引时扩展（增加索引体积但搜索更快）或搜索时扩展（索引不变但每次搜索需扩展）。推荐在搜索时使用同义词，便于动态更新。

索引别名与零停机重建

索引别名（Alias）是生产环境的重要实践。为索引创建别名后，应用通过别名访问索引。当需要修改映射或重建索引时，创建新索引、导入数据、验证无误后将别名指向新索引、删除旧索引，整个过程对应用透明，实现零停机迁移。别名还支持过滤别名（自动添加filter条件）和路由别名（指定routing值）。

深度分页优化

传统的from/size分页在深度分页时性能急剧下降——Elasticsearch需要从每个分片获取from+size条数据，在协调节点合并排序后返回size条。当from=10000时，每个分片需要返回10000+size条数据，内存和CPU开销巨大。解决方案是使用search_after参数，基于上一页最后一条文档的排序值获取下一页，每次查询都是独立的高效查询。对于需要导出全部数据的场景，使用scroll API。

最佳实践

索引设计方面，主分片数量根据数据量预估，单个分片建议不超过50GB。副本数量根据可用性要求设置，生产环境至少1个副本。避免过度映射——只映射需要搜索和聚合的字段，动态映射（dynamic mapping）在生产环境建议关闭或设为strict模式，防止意外字段污染映射。

查询优化方面，filter查询不计算得分且结果可缓存，优先使用filter替代must进行条件过滤。避免使用通配符开头的查询（如*word），这会导致全索引扫描。聚合查询设置size: 0跳过文档返回，只返回聚合结果。使用routing参数将相关文档路由到同一分片，减少跨分片查询。

集群运维方面，监控JVM堆内存使用率，保持在75%以下。设置合理的刷新间隔（index.refresh_interval），批量写入时可以临时设为-1（禁用刷新）或30s提升写入性能。使用索引生命周期管理（ILM）自动管理索引的滚动、收缩和删除。定期强制合并（force merge）旧索引减少段数量，提升搜索性能。

总结

Elasticsearch作为分布式全文检索引擎，以其倒排索引结构、丰富的查询DSL和强大的聚合分析能力，成为搜索场景的首选方案。本文从elasticsearch-py客户端的基本连接出发，系统讲解了索引创建与映射配置、文档CRUD操作、搜索DSL的各种查询类型（match、term、range、bool、fuzzy、geo等）、分词器配置与测试、搜索结果高亮、聚合分析（桶聚合、指标聚合、嵌套聚合）以及批量操作和scroll遍历。在实际应用中，搜索相关性优化是核心挑战——通过字段权重boost、function_score自定义打分、同义词扩展等手段提升搜索质量。索引别名实现零停机重建，search_after解决深度分页性能问题。分词器选择对中文搜索至关重要，IK分词器的ik_max_word和ik_smart模式分别适用于索引和搜索。掌握Elasticsearch的关键在于理解倒排索引的工作原理，合理设计映射和分析器，根据业务场景选择合适的查询类型和聚合方式，并通过别名、ILM等机制保证集群的可运维性。