> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# Elasticsearch 完全指南
- URL: https://blog.vercanti.com/elasticsearch-wan-quan-zhi-nan/
- Published: 2026-08-28T14:35:36.000Z
- Updated: 2026-08-28T14:59:08.000Z
- Description: 相关文档：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
- Author: yellowdog
- Tags: 数据库

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

相关文档：[PostgreSQL完全指南](https://blog.vercanti.com/postgresql-wan-quan-zhi-nan/) [MongoDB完全指南](https://blog.vercanti.com/mongodb-wan-quan-zhi-nan/) [Redis完全指南](https://blog.vercanti.com/redis-wan-quan-zhi-nan/)

---

## 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 等

### 安装

```bash
# 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

```python
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 定义字段类型，一旦设置不可更改（只能新增字段）。

```python
# 创建索引并定义 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

### 全文搜索

```python
# 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 组合查询

```python
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

```python
# 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 + 统计函数。

```python
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\. 批量操作

```python
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\. 分页

```python
# 普通分页（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\. 高亮显示

```python
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 集成

```python
# 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\. 常用代码段

### 检查索引是否存在

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

```

### 更新 Mapping（新增字段）

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

```

### 删除并重建索引（Reindex）

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

```

### 统计文档数

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

```

---

## 11\. 最佳实践

### Mapping 设计原则

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

```python
"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

```python
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` 用于精确匹配、聚合、排序。两者通常同时需要。

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

```

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

```python
# 关闭自动 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"`（搜索分词）。

```bash
./bin/elasticsearch-plugin install analysis-ik

```

```json
"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 将数据从旧索引迁移到新索引，更新别名后切换流量。

```python
# 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 子字段）。

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

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

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

```

---

## 参见

- [MongoDB完全指南](https://blog.vercanti.com/mongodb-wan-quan-zhi-nan/)
- [MySQL基础完全指南](https://blog.vercanti.com/mysql-ji-chu-wan-quan-zhi-nan/)
- [PostgreSQL完全指南](https://blog.vercanti.com/postgresql-wan-quan-zhi-nan/)
- [Redis完全指南](https://blog.vercanti.com/redis-wan-quan-zhi-nan/)