使用 Elasticsearch 构建搜索引擎

很多人对 Elasticsearch 的第一印象是“搜索神器”，但它本质上是一个基于 Apache Lucene 优化的分布式实时文档存储+分析引擎——存储只是基础，检索才是核心优势。今天就来快速上手，用 Python 客户端搭一个基础的中文全文检索系统原型。

1. 先理清核心逻辑：类比关系型数据库

很多新手容易被 ES 的一堆专有名词绕晕，这里先用大家熟悉的 RDBMS 做个简单但够用的对应（注意不是严格等价）：

RDBMS 结构Elasticsearch 对应简单说明
数据库(Database)索引(Index)数据的顶层管理容器，必须全小写
表(Table)—7.x+已完全弃用 Type，现在一个 Index 只存一类文档
行(Row)文档(Document)JSON 格式的核心数据单元
列(Column)字段(Field)JSON 中的键值对，支持文本、数值、地理坐标等多种类型
主键(Primary Key)文档ID(Document ID)可以自动生成，也可以业务方指定（比如用订单号、新闻ID）

2. 5分钟搭好本地测试环境

不用纠结复杂的集群配置，单节点 Docker 容器是新手入门最快的方式：

2.1 拉取并启动 ES

# 拉取 8.12.0 版本（如果版本变了，后面的 IK 分词器也要对应）
docker pull docker.elastic.co/elasticsearch/elasticsearch:8.12.0

# 启动单节点（关闭集群发现、禁用安全配置方便测试，生产环境绝对不能这么做！）
docker run -d \
  -p 9200:9200 \
  -p 9300:9300 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "xpack.security.enrollment.enabled=false" \
  docker.elastic.co/elasticsearch/elasticsearch:8.12.0

启动后等30秒左右，访问 http://localhost:9200，如果返回类似 JSON 的集群信息，说明成功了。

2.2 安装 Python 客户端

官方推荐的 elasticsearch-py 7.x+ 支持 ES 7.x/8.x 系列，直接 pip 安装即可：

pip install elasticsearch==8.12.0  # 建议和 ES 版本号一致，避免兼容性问题

3. 准备工作：索引定义（Mapping）

ES 默认会根据你插入的第一条文档自动推断字段类型，但中文全文检索场景必须手动指定 Mapping——比如给文本字段加中文分词器，给 URL、状态码这种不需要分词的字段设为 keyword 类型。

这里我们用新闻搜索做场景，先插入 IK 分词器（默认 ES 只有英文分词器，中文会拆成单个汉字）：

# 先找到容器 ID：docker ps
docker exec -it <你的容器ID> bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v8.12.0/elasticsearch-analysis-ik-8.12.0.zip

# 安装后重启容器
docker restart <你的容器ID>

重启后写 Python 代码创建索引：

from elasticsearch import Elasticsearch

# 连接本地 ES
es = Elasticsearch("http://localhost:9200")
if not es.ping():
    raise ConnectionError("无法连接到 Elasticsearch，请检查容器是否正常运行！")

# 定义索引 Mapping
news_mapping = {
    "mappings": {
        "properties": {
            "title": {  # 新闻标题：需要细粒度分词，支持“中国高考”“高考政策”等组合搜索
                "type": "text",
                "analyzer": "ik_max_word",  # 最细粒度分词：比如“中国学生”会拆成“中国、国中、学生、国学生、中国学生”
                "search_analyzer": "ik_smart"  # 搜索时粗粒度：只拆出“中国、学生”，避免无意义的匹配
            },
            "content": {  # 新闻内容：同标题
                "type": "text",
                "analyzer": "ik_max_word",
                "search_analyzer": "ik_smart"
            },
            "url": {  # 新闻链接：不需要分词，只需要精准匹配或前缀匹配
                "type": "keyword"
            },
            "publish_date": {  # 发布时间：需要范围过滤
                "type": "date"
            },
            "category": {  # 分类：精准匹配
                "type": "keyword"
            }
        }
    }
}

# 创建索引，ignore=400 表示如果索引已存在就跳过（不会报错）
response = es.indices.create(index="chinese_news", body=news_mapping, ignore=400)
print("创建索引结果：", response)

4. 核心功能实战

4.1 文档操作

ES 提供了单条和批量操作的接口，生产环境优先用 bulk API 批量插入/更新/删除，性能比单条快几十倍。

单条操作

# 1. 指定 ID 插入（如果 ID 已存在会报错）
doc1 = {
    "title": "2024年高考报名时间公布：多省提前启动",
    "content": "近日，教育部发布通知，2024年全国普通高等学校招生统一考试报名工作将在部分省份提前启动，具体时间以当地教育考试院公告为准...",
    "url": "https://example.com/news/20240520/1",
    "publish_date": "2024-05-20",
    "category": "教育"
}
res = es.create(index="chinese_news", id="20240520_1", body=doc1)
print("指定ID插入结果：", res["result"])  # 成功返回 created

# 2. 自动生成 ID 插入（幂等性差，不建议核心数据用）
doc2 = doc1.copy()
doc2["url"] = "https://example.com/news/20240520/2"
res = es.index(index="chinese_news", body=doc2)
print("自动ID插入结果：", res["result"], res["_id"])

# 3. 更新文档（部分更新/全量更新都用这个接口，部分更新只需传要改的字段）
update_body = {
    "doc": {
        "category": "高考政策"  # 部分更新分类
    }
}
res = es.update(index="chinese_news", id="20240520_1", body=update_body)
print("部分更新结果：", res["result"])  # 成功返回 updated

# 4. 删除文档
res = es.delete(index="chinese_news", id=res["_id"])  # 删除刚才自动生成的那条
print("删除结果：", res["result"])

4.2 搜索功能

搜索是 ES 的灵魂，这里介绍3个最常用的场景。

场景1：简单的全文匹配

比如搜索“2024高考报名”，ES 会返回标题或内容包含相关关键词的文档，并按相关性排序：

simple_query = {
    "query": {
        "multi_match": {  # 多字段匹配
            "query": "2024高考报名",
            "fields": ["title^3", "content"]  # title的权重是content的3倍
        }
    },
    "size": 10,  # 只返回前10条（默认也是10）
    "_source": ["title", "url", "publish_date", "category"]  # 只返回需要的字段，减少传输量
}

res = es.search(index="chinese_news", body=simple_query)
print("搜索到的文档数：", res["hits"]["total"]["value"])
for hit in res["hits"]["hits"]:
    print(f"标题：{hit['_source']['title']} | 相关性分数：{hit['_score']}")

场景2：带条件的布尔查询

布尔查询是 ES 最灵活的查询方式，支持 must（必须满足，算分）、must_not（必须不满足，不算分）、should（满足即可加分）、filter（必须满足，不算分但会走缓存，性能最好）。

比如搜索“2023-2024年发布的，分类是‘高考政策’或‘教育’，标题或内容包含‘报名’但不包含‘成人高考’的新闻”：

bool_query = {
    "query": {
        "bool": {
            "must": [
                {"multi_match": {"query": "报名", "fields": ["title^3", "content"]}}
            ],
            "must_not": [
                {"match": {"title": "成人高考"}}
            ],
            "should": [
                {"term": {"category": "高考政策"}},  # term是精准匹配keyword类型
                {"term": {"category": "教育"}}
            ],
            "filter": [
                {"range": {"publish_date": {"gte": "2023-01-01", "lte": "2024-12-31"}}}  # 范围过滤
            ],
            "minimum_should_match": 1  # should里至少满足1个
        }
    },
    "sort": [
        {"publish_date": {"order": "desc"}},  # 先按发布时间倒序
        {"_score": {"order": "desc"}}  # 再按相关性分数倒序
    ],
    "size": 10,
    "_source": ["title", "url", "publish_date", "category"]
}

res = es.search(index="chinese_news", body=bool_query)
print("过滤后的搜索结果：")
for hit in res["hits"]["hits"]:
    print(f"{hit['_source']['publish_date']} | {hit['_source']['category']} | {hit['_source']['title']}")

5. 新手必看的3个最佳实践

  1. 索引设计要前置规划：Mapping 创建后，除了新增字段，其他类型/分词器的修改都需要重建索引。可以用别名(Alias) 管理索引，重建时只需要切换别名，业务完全无感知。
  2. 性能优化从细节入手：
    • 避免大文档（>100MB），建议拆成多个小文档；
    • 批量操作时，bulk 请求的大小控制在 5-15MB；
    • filter 查询多的场景，要合理利用 ES 的缓存机制。
  3. 生产环境必须开启安全配置：前面我们为了测试关闭了 xpack.security，生产环境要启用 HTTPS、用户名密码认证、RBAC 访问控制。

6. 总结与扩展

今天我们快速过了 ES 的核心概念、环境搭建、索引定义、文档操作和常用搜索场景，用 Python 就能轻松搭一个基础的中文全文检索原型。

如果想深入学习，可以看看这些官方/权威资源：