数据库

Elasticsearch 完全指南

相关文档：PostgreSQL完全指南(/postgresql-wan-quan-zhi-nan/) MongoDB完全指南(/mongodb-wan-quan-zhi-nan/) Redis完全指南(/redis-wan-quan-zhi-nan/) Elasticsearch 是基于 Apache Lucene 的分布式全文搜索引擎，RESTful API 驱动，支持近实时搜索和分析。 Mapping 定义字段类型，一旦设置不可更改（只能新增字段）。聚合类似 SQL 的 GROUP BY + 统计函数。 Mapping 提前定义，不依赖动态映射：E

官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html
适用版本：Elasticsearch 8.x（2026-05-07 核实）

1. 基础概念

Elasticsearch 是什么

Elasticsearch 是基于 Apache Lucene 的分布式全文搜索引擎，RESTful API 驱动，支持近实时搜索和分析。

概念	说明	类比（关系型数据库）
Index	存储文档的逻辑命名空间	数据库
Document	JSON 格式的数据单元	行
Field	文档的键值对	列
Mapping	字段类型定义	Schema
Shard	Index 的物理分片	分区
Replica	Shard 的副本（高可用）	主从复制

核心特性

倒排索引：term → [doc_id, position] 映射，全文检索极快
近实时（NRT）：默认 1 秒刷新，写入后约 1 秒可搜索
分布式：数据自动分片，水平扩展
丰富查询 DSL：bool/match/range/geo/aggregation 等

安装

# Docker 单节点（开发用）
docker run -d \
  --name elasticsearch \
  -p 9200:9200 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  docker.elastic.co/elasticsearch/elasticsearch:8.12.0

# Python 客户端
pip install elasticsearch[async]   # 异步客户端（推荐）

2. 基础 CRUD

from elasticsearch import AsyncElasticsearch

es = AsyncElasticsearch("http://localhost:9200")

# 创建 / 更新文档
await es.index(
    index="articles",
    id="1",
    document={
        "title": "Elasticsearch 入门",
        "content": "Elasticsearch 是分布式搜索引擎...",
        "author": "Alice",
        "created_at": "2026-01-01T00:00:00",
        "tags": ["search", "database"],
        "views": 100,
    }
)

# 获取文档
doc = await es.get(index="articles", id="1")
print(doc["_source"])

# 更新文档（部分更新）
await es.update(
    index="articles",
    id="1",
    doc={"views": 101},
)

# 删除文档
await es.delete(index="articles", id="1")

# 关闭连接
await es.close()

3. Mapping 定义

Mapping 定义字段类型，一旦设置不可更改（只能新增字段）。

# 创建索引并定义 Mapping
await es.indices.create(
    index="articles",
    body={
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 0,
            "analysis": {
                "analyzer": {
                    "ik_smart_analyzer": {
                        "type": "custom",
                        "tokenizer": "ik_smart",  # 需安装 IK 分词插件
                    }
                }
            }
        },
        "mappings": {
            "properties": {
                "title": {
                    "type": "text",
                    "analyzer": "ik_smart",       # 索引时分词
                    "search_analyzer": "ik_smart", # 搜索时分词
                    "fields": {
                        "keyword": {              # 同时保留 keyword 类型（精确匹配/排序）
                            "type": "keyword",
                            "ignore_above": 256,
                        }
                    }
                },
                "content": {"type": "text", "analyzer": "ik_smart"},
                "author": {"type": "keyword"},
                "created_at": {"type": "date"},
                "tags": {"type": "keyword"},
                "views": {"type": "integer"},
                "location": {"type": "geo_point"},
            }
        }
    }
)

常用字段类型

类型	说明	用途
`text`	分词后建立倒排索引	全文搜索
`keyword`	不分词，精确值	过滤、排序、聚合
`integer/long/float`	数值	范围查询、聚合
`date`	日期（ISO 8601 或时间戳）	时间范围查询
`boolean`	布尔值	过滤
`nested`	嵌套对象数组（保留对象关系）	复杂嵌套查询
`geo_point`	经纬度坐标	地理位置查询

4. 查询 DSL

全文搜索

# match — 单字段全文搜索（分词后搜索）
response = await es.search(
    index="articles",
    body={
        "query": {
            "match": {
                "title": "Elasticsearch 搜索引擎"
            }
        }
    }
)

# multi_match — 多字段全文搜索
response = await es.search(
    index="articles",
    body={
        "query": {
            "multi_match": {
                "query": "搜索引擎",
                "fields": ["title^2", "content"],  # title 权重加倍
                "type": "best_fields",
            }
        }
    }
)

# match_phrase — 短语搜索（词序和位置都匹配）
response = await es.search(
    index="articles",
    body={
        "query": {
            "match_phrase": {
                "content": "分布式搜索引擎"
            }
        }
    }
)

bool 组合查询

response = await es.search(
    index="articles",
    body={
        "query": {
            "bool": {
                "must": [                         # 必须匹配（影响评分）
                    {"match": {"title": "Elasticsearch"}}
                ],
                "filter": [                       # 必须匹配（不影响评分，有缓存）
                    {"term": {"author": "Alice"}},
                    {"range": {"created_at": {"gte": "2026-01-01"}}},
                    {"terms": {"tags": ["search", "database"]}},
                ],
                "should": [                       # 可选（匹配则提升评分）
                    {"match": {"content": "倒排索引"}},
                ],
                "must_not": [                     # 不能匹配
                    {"term": {"author": "Bob"}},
                ],
            }
        }
    }
)

term / range / exists

# term — 精确匹配（keyword 类型）
{"term": {"author": "Alice"}}
{"terms": {"author": ["Alice", "Bob"]}}

# range — 范围查询
{"range": {"views": {"gte": 100, "lte": 1000}}}
{"range": {"created_at": {"gte": "2026-01-01", "lt": "2027-01-01"}}}

# exists — 字段存在
{"exists": {"field": "tags"}}

# wildcard — 通配符（性能较差，慎用）
{"wildcard": {"title.keyword": "Elastic*"}}

# prefix — 前缀匹配
{"prefix": {"title.keyword": "Elastic"}}

5. 聚合（Aggregation）

聚合类似 SQL 的 GROUP BY + 统计函数。

response = await es.search(
    index="articles",
    body={
        "size": 0,  # 不返回文档，只返回聚合结果
        "aggs": {
            # Bucket 聚合 — 分组
            "by_author": {
                "terms": {
                    "field": "author",
                    "size": 10,
                    "order": {"avg_views": "desc"},
                },
                "aggs": {
                    # 子聚合
                    "avg_views": {"avg": {"field": "views"}},
                    "total_views": {"sum": {"field": "views"}},
                }
            },
            # Metric 聚合 — 统计
            "views_stats": {
                "stats": {"field": "views"}
                # 返回：count/min/max/avg/sum
            },
            # Date Histogram — 按时间分组
            "by_month": {
                "date_histogram": {
                    "field": "created_at",
                    "calendar_interval": "month",
                    "format": "yyyy-MM",
                },
                "aggs": {
                    "article_count": {"value_count": {"field": "_id"}}
                }
            },
            # Range 聚合
            "views_range": {
                "range": {
                    "field": "views",
                    "ranges": [
                        {"to": 100},
                        {"from": 100, "to": 1000},
                        {"from": 1000},
                    ]
                }
            }
        }
    }
)

# 获取聚合结果
buckets = response["aggregations"]["by_author"]["buckets"]
for bucket in buckets:
    print(bucket["key"], bucket["avg_views"]["value"])

6. 批量操作

from elasticsearch.helpers import async_bulk

# bulk 写入（高效批量索引）
async def bulk_index(documents: list[dict]):
    actions = [
        {
            "_index": "articles",
            "_id": doc["id"],
            "_source": doc,
        }
        for doc in documents
    ]
    success, failed = await async_bulk(es, actions, chunk_size=500)
    return success, failed

# 批量查询
response = await es.mget(
    index="articles",
    body={"ids": ["1", "2", "3"]},
)

7. 分页

# 普通分页（from + size，深度分页性能差，建议 from < 10000）
response = await es.search(
    index="articles",
    body={
        "from": 0,
        "size": 20,
        "query": {"match_all": {}},
        "sort": [{"created_at": "desc"}, {"_score": "desc"}],
    }
)

# Search After — 游标分页（深度分页推荐）
response = await es.search(
    index="articles",
    body={
        "size": 20,
        "query": {"match_all": {}},
        "sort": [{"created_at": "desc"}, {"_id": "asc"}],
        "search_after": ["2026-01-15T10:00:00", "abc123"],  # 上一页最后一条的 sort 值
    }
)

# Scroll API — 全量导出（不适合实时搜索）
response = await es.search(
    index="articles",
    scroll="2m",      # scroll 上下文保留 2 分钟
    size=1000,
    body={"query": {"match_all": {}}}
)
scroll_id = response["_scroll_id"]

while True:
    hits = response["hits"]["hits"]
    if not hits:
        break
    process(hits)
    response = await es.scroll(scroll_id=scroll_id, scroll="2m")

await es.clear_scroll(scroll_id=scroll_id)

8. 高亮显示

response = await es.search(
    index="articles",
    body={
        "query": {"match": {"content": "搜索引擎"}},
        "highlight": {
            "fields": {
                "content": {
                    "fragment_size": 150,         # 每个片段的字符数
                    "number_of_fragments": 3,      # 最多返回几个片段
                    "pre_tags": ["<em>"],
                    "post_tags": ["</em>"],
                }
            }
        }
    }
)

for hit in response["hits"]["hits"]:
    print(hit["highlight"]["content"])  # ['...关键词<em>搜索引擎</em>...']

9. 与 FastAPI 集成

# src/core/elasticsearch.py
from contextlib import asynccontextmanager
from elasticsearch import AsyncElasticsearch
from fastapi import FastAPI

es: AsyncElasticsearch = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global es
    es = AsyncElasticsearch("http://localhost:9200")
    yield
    await es.close()

app = FastAPI(lifespan=lifespan)

# 搜索 API
from fastapi import Query

@app.get("/search")
async def search_articles(
    q: str = Query(..., description="搜索关键词"),
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    author: str | None = None,
):
    filters = []
    if author:
        filters.append({"term": {"author": author}})

    response = await es.search(
        index="articles",
        body={
            "from": (page - 1) * size,
            "size": size,
            "query": {
                "bool": {
                    "must": [{"multi_match": {"query": q, "fields": ["title^2", "content"]}}],
                    "filter": filters,
                }
            },
            "highlight": {
                "fields": {"title": {}, "content": {"fragment_size": 200}}
            },
        }
    )

    hits = response["hits"]["hits"]
    return {
        "total": response["hits"]["total"]["value"],
        "results": [
            {**h["_source"], "highlight": h.get("highlight", {})}
            for h in hits
        ]
    }

10. 常用代码段

检查索引是否存在

exists = await es.indices.exists(index="articles")
if not exists:
    await create_index()

更新 Mapping（新增字段）

await es.indices.put_mapping(
    index="articles",
    body={
        "properties": {
            "new_field": {"type": "keyword"}
        }
    }
)

删除并重建索引（Reindex）

# 重建索引（修改不可变的 mapping 时使用）
await es.reindex(body={
    "source": {"index": "articles"},
    "dest": {"index": "articles_v2"},
})
await es.indices.put_alias(index="articles_v2", name="articles_alias")

统计文档数

count = await es.count(index="articles", body={"query": {"match_all": {}}})
print(count["count"])

11. 最佳实践

Mapping 设计原则

对需要全文搜索的字段用 text，同时用 .keyword 子字段支持排序和聚合
对不需要搜索、只需过滤的字段用 keyword（如 ID、状态）
用 index: false 关闭不需要搜索的字段的索引，减少存储

"fields": {
    "status": {"type": "keyword"},
    "raw_content": {"type": "text", "index": False},  # 只存储不搜索
}

filter vs must

过滤条件（不影响评分）用 filter，ES 会缓存 filter 结果，性能更好
只有需要影响相关性评分的条件才放入 must

批量写入

单条写入延迟高，批量用 async_bulk，每批 500~1000 条
写入时关闭 refresh_interval（设为 -1），完成后手动触发 refresh

await es.indices.put_settings(index="articles", body={"refresh_interval": "-1"})
# ... 批量写入 ...
await es.indices.refresh(index="articles")
await es.indices.put_settings(index="articles", body={"refresh_interval": "1s"})

最佳实践

Mapping 提前定义，不依赖动态映射：ES 的动态映射会猜测字段类型，把数字字符串（如 "123"）映射为 text 或 long，后续难以修改。生产环境必须在创建索引时显式定义所有字段的 Mapping。

text 字段同时配置 .keyword 子字段：text 用于全文搜索，keyword 用于精确匹配、聚合、排序。两者通常同时需要。

"title": {
  "type": "text",
  "analyzer": "ik_max_word",
  "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } }
}

批量写入关闭自动刷新以提升吞吐：默认每 1 秒 refresh 一次（数据可见），大批量导入时关闭可提升写入速度 2–5 倍，完成后手动 refresh。

# 关闭自动 refresh
await es.indices.put_settings(index="articles", body={"refresh_interval": "-1"})
# 批量写入...
await es.helpers.async_bulk(es, actions)
# 恢复
await es.indices.refresh(index="articles")
await es.indices.put_settings(index="articles", body={"refresh_interval": "1s"})

bool 查询中将过滤条件放入 filter，不放 must：filter 子句不参与相关性评分且结果可被缓存；must 参与评分，每次都要计算 TF-IDF/BM25。能用 filter 的条件（状态、时间范围）一律放 filter。

分片数量在创建时确定，不可修改：主分片数在索引创建后不可更改。索引创建前根据数据量估算：每个分片 20–50GB 为宜，不要追求过多分片（增加集群协调开销）。

使用 search_after 替代 from+size 实现深度分页：from + size > 10000 时 ES 默认拒绝，且即使放开限制，深度分页需要协调所有分片的 top-N，开销随深度线性增长。

常见陷阱

陷阱：中文搜索按单字切分，结果不准确

现象： 搜索"机器学习"只能搜到包含"机"、"器"、"学"、"习"单字的文档，不能按词搜索。

原因： ES 内置分析器（standard/whitespace）对中文按 Unicode 码点切分，不做词法分析。

解决： 安装 IK 分词插件，Mapping 中指定 "analyzer": "ik_max_word"（建库分词）和 "search_analyzer": "ik_smart"（搜索分词）。

./bin/elasticsearch-plugin install analysis-ik

"content": {
  "type": "text",
  "analyzer": "ik_max_word",
  "search_analyzer": "ik_smart"
}

陷阱：字段类型一旦建立无法修改

现象： 需要把 price 从 keyword 改为 float，执行 PUT /_mapping 报错 mapper cannot be changed from type [keyword] to [float]。

原因： ES 的倒排索引结构决定了已有字段的类型在索引建立后不可修改（修改需要重建倒排索引）。

解决： 新建索引，用 _reindex API 将数据从旧索引迁移到新索引，更新别名后切换流量。

# 1. 新建索引（新 Mapping）
await es.indices.create(index="articles_v2", body={...})

# 2. 重索引
await es.reindex(body={
    "source": {"index": "articles_v1"},
    "dest": {"index": "articles_v2"}
})

# 3. 切换别名
await es.indices.update_aliases(body={
    "actions": [
        {"remove": {"index": "articles_v1", "alias": "articles"}},
        {"add": {"index": "articles_v2", "alias": "articles"}},
    ]
})

陷阱：text 字段直接排序或聚合报错

现象： {"sort": [{"title": "asc"}]} 报错 Text fields are not optimised for operations that require per-document field data。

原因： text 字段已经分词，每个词条都是独立的 token，无法对原始字符串排序或聚合。

解决： 使用 .keyword 子字段进行排序和聚合（需在 Mapping 中提前定义 keyword 子字段）。

# 错误
"sort": [{"title": "asc"}]

# 正确：使用 keyword 子字段
"sort": [{"title.keyword": "asc"}]

# 聚合也一样
"aggs": {"by_title": {"terms": {"field": "title.keyword"}}}