Elasticsearch 全文搜索与数据分析实战指南

FST（Finite State Transducer，有限状态传感器）是 Lucene 对传统前缀树的优化实现，用于存储词条索引。相比传统前缀树，FST 具有更高的空间效率和查询效率。

FST 的核心思想是将词条词典构建为一个有限状态自动机，共享公共前缀和后缀。通过状态转移实现词条查找，同时支持输出值的计算。这种结构使得词条索引可以常驻内存，大幅减少磁盘 IO。

FST 的优势体现在三个方面：空间压缩、快速查找和前缀匹配。空间压缩通过共享前后缀实现，内存占用可减少 50% 以上。快速查找通过状态转移实现，时间复杂度为 O(m)，m 为词条长度。前缀匹配支持自动补全等场景，无需额外数据结构。

# FST 构建示例：模拟词条索引的构建过程
class FSTNode:
    """FST 节点类"""
    def __init__(self):
        self.transitions = {}  # 状态转移表
        self.output = 0  # 输出值
        self.is_final = False  # 是否为终止状态

class FST:
    """简化的 FST 实现"""
    def __init__(self):
        self.root = FSTNode()

    def insert(self, term, doc_ids):
        """插入词条及其对应的文档 ID 列表"""
        node = self.root
        for char in term:
            if char not in node.transitions:
                node.transitions[char] = FSTNode()
            node = node.transitions[char]
        node.is_final = True
        node.output = doc_ids  # 存储倒排表指针

    def search(self, term):
        """查找词条对应的文档 ID 列表"""
        node = self.root
        for char in term:
            if char not in node.transitions:
                return None
            node = node.transitions[char]
        return node.output if node.is_final else None

# 构建示例
fst = FST()
fst.insert("elasticsearch", [1, 3])
fst.insert("搜索", [1, 2])
fst.insert("引擎", [2])
fst.insert("优化", [2])
fst.insert("实战", [3])

# 查询示例
result = fst.search("搜索")
print(f"'搜索' 对应的文档 ID: {result}")  # 输出：[1, 2]

上述代码展示了 FST 的基本结构和操作。在实际的 Lucene 实现中，FST 采用了更加复杂的压缩算法和序列化格式，但核心思想是一致的。通过 FST，Elasticsearch 能够在内存中快速定位词条，然后从磁盘读取对应的倒排表，实现高效的全文检索。

3. 全文搜索基础：核心查询语法

Elasticsearch 提供了丰富的查询 DSL（Domain Specific Language），支持从简单到复杂的各种搜索场景。掌握核心查询语法，是构建高效搜索系统的基础。

3.1 match 查询：全文检索的主力

match 查询是 Elasticsearch 中最常用的全文检索查询类型。它会对查询文本进行分词，然后查找包含任意或所有分词结果的文档。

// match 查询示例：搜索商品名称
GET /products/_search {
  "query": {
    "match": {
      "name": {
        "query": "智能手机",
        "operator": "or", // 匹配任意一个词即可
        "minimum_should_match": "75%", // 至少匹配 75% 的词
        "analyzer": "ik_max_word" // 使用 IK 分词器
      }
    }
  },
  "highlight": {
    "fields": {
      "name": {}
    }
  },
  "size": 20
}

上述查询展示了 match 查询的完整语法。query 字段指定查询文本，operator 控制匹配逻辑（or/and），minimum_should_match 设置最小匹配比例，analyzer 指定分词器。查询结果会高亮显示匹配部分，便于用户快速定位。

match 查询有多个变体：match_phrase 用于短语匹配，要求词序一致；match_phrase_prefix 支持前缀匹配，适用于搜索建议；multi_match 支持多字段查询，适用于需要同时搜索多个字段的场景。

3.2 term 查询：精确匹配

term 查询用于精确匹配，不对查询文本进行分词处理。适用于关键词、ID、状态码等精确值字段的查询。

// term 查询示例：精确匹配商品分类
GET /products/_search {
  "query": {
    "bool": {
      "must": [
        {"term": {"category_id": {"value": "electronics"}}},
        {"terms": {"brand_id": ["apple", "samsung", "huawei"]}} // 多值匹配
      ],
      "filter": [
        {"range": {"price": {"gte": 1000, "lte": 5000}}}
      ]
    }
  }
}

上述查询使用 bool 查询组合多个条件。must 中的条件必须满足，filter 中的条件用于过滤但不参与评分。term 查询适用于 keyword 类型字段，对于 text 类型字段，建议使用 match 查询。

3.3 bool 查询：复杂条件组合

bool 查询是 Elasticsearch 中最强大的查询类型，支持通过逻辑组合构建复杂查询条件。

bool 查询的四个子句各有特点：must 表示必须匹配，参与评分计算；should 表示可选匹配，参与评分计算，可通过 minimum_should_match 控制最少匹配数；must_not 表示必须不匹配，不参与评分；filter 表示必须匹配，但不参与评分，性能更优。

# 使用 Python Elasticsearch 客户端构建 bool 查询
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

# 连接 Elasticsearch
es = Elasticsearch(["http://localhost:9200"], basic_auth=("elastic", "your_password"))

# 构建复杂的 bool 查询
def search_products(keyword, category=None, min_price=None, max_price=None, brands=None, page=1, size=20):
    """商品搜索函数：支持关键词、分类、价格区间、品牌等多条件组合"""
    query = {
        "query": {
            "bool": {
                "must": [
                    {"match": {"name": {"query": keyword, "operator": "and", "boost": 2.0}}},
                    {"match": {"description": keyword}}
                ],
                "should": [
                    {"term": {"is_hot": {"value": True, "boost": 1.5}}}
                ],
                "filter": []
            }
        },
        "sort": ["_score", {"sales_count": "desc"}, {"created_at": "desc"}],
        "from": (page - 1) * size,
        "size": size,
        "highlight": {
            "fields": {
                "name": {},
                "description": {"fragment_size": 150}
            }
        }
    }

    # 添加过滤条件
    if category:
        query["query"]["bool"]["filter"].append({"term": {"category_id": category}})
    if min_price is not None or max_price is not None:
        price_range = {}
        if min_price is not None:
            price_range["gte"] = min_price
        if max_price is not None:
            price_range["lte"] = max_price
        query["query"]["bool"]["filter"].append({"range": {"price": price_range}})
    if brands:
        query["query"]["bool"]["filter"].append({"terms": {"brand_id": brands}})

    # 执行查询
    response = es.search(index="products", body=query)

    # 处理结果
    results = []
    for hit in response["hits"]["hits"]:
        product = hit["_source"]
        product["_score"] = hit["_score"]
        if "highlight" in hit:
            product["highlight"] = hit["highlight"]
        results.append(product)
    
    return {
        "total": response["hits"]["total"]["value"],
        "results": results,
        "page": page,
        "size": size
    }

# 调用示例
result = search_products(
    keyword="智能手机",
    category="electronics",
    min_price=1000,
    max_price=5000,
    brands=["apple", "huawei"],
    page=1,
    size=20
)
print(f"找到 {result['total']} 个商品")

上述代码展示了如何使用 Python 客户端构建复杂的 bool 查询。查询包含 must、should、filter 三个子句，实现了关键词搜索、分类过滤、价格区间筛选、品牌过滤等功能。同时支持分页、排序和高亮显示，是一个完整的商品搜索实现。

4. 聚合查询详解：数据分析利器

聚合（Aggregations）是 Elasticsearch 强大的数据分析功能，支持对数据进行统计、分组、计算等操作。聚合查询可以与搜索查询结合使用，实现"搜索 + 分析"一体化。

4.1 聚合类型概览

Elasticsearch 提供了三大类聚合：指标聚合（Metric Aggregations）、桶聚合（Bucket Aggregations）和管道聚合（Pipeline Aggregations）。

4.2 terms 聚合：分组统计

terms 聚合是最常用的桶聚合，按照字段的值将文档分组，并返回每个分组的文档数量。

// terms 聚合示例：统计各品牌商品数量和平均价格
GET /products/_search {
  "size": 0,
  "aggs": {
    "brands": {
      "terms": {
        "field": "brand_id",
        "size": 10,
        "order": {"avg_price": "desc"}
      },
      "aggs": {
        "avg_price": {"avg": {"field": "price"}},
        "total_sales": {"sum": {"field": "sales_count"}},
        "price_ranges": {
          "range": {
            "field": "price",
            "ranges": [
              {"to": 1000, "key": "低价位"},
              {"from": 1000, "to": 3000, "key": "中价位"},
              {"from": 3000, "key": "高价位"}
            ]
          }
        }
      }
    }
  }
}

上述查询展示了 terms 聚合的嵌套使用。外层 terms 聚合按品牌分组，内层嵌套了 avg、sum 和 range 三种聚合，分别计算平均价格、总销量和价格区间分布。这种嵌套聚合是 Elasticsearch 强大的分析能力体现。

4.3 histogram 聚合：区间统计

histogram 聚合将数值字段按指定间隔分组，适用于价格区间、年龄分布等场景。date_histogram 是 histogram 的时间版本，适用于时间序列数据分析。

# 时间序列聚合分析示例
def analyze_sales_trend(index="orders", interval="day", start_date=None, end_date=None):
    """分析销售趋势：按时间维度统计订单量和销售额"""
    query = {
        "size": 0,
        "query": {
            "range": {
                "created_at": {
                    "gte": start_date,
                    "lte": end_date
                }
            }
        },
        "aggs": {
            "sales_over_time": {
                "date_histogram": {
                    "field": "created_at",
                    "calendar_interval": interval,
                    "format": "yyyy-MM-dd",
                    "min_doc_count": 0
                },
                "aggs": {
                    "total_revenue": {"sum": {"field": "total_amount"}},
                    "avg_order_value": {"avg": {"field": "total_amount"}},
                    "revenue_derivative": {"derivative": {"buckets_path": "total_revenue"}},
                    "cumulative_revenue": {"cumulative_sum": {"buckets_path": "total_revenue"}}
                }
            }
        }
    }
    response = es.search(index=index, body=query)

    # 解析结果
    buckets = response["aggregations"]["sales_over_time"]["buckets"]
    trend_data = []
    for bucket in buckets:
        trend_data.append({
            "date": bucket["key_as_string"],
            "order_count": bucket["doc_count"],
            "total_revenue": bucket["total_revenue"]["value"],
            "avg_order_value": bucket["avg_order_value"]["value"],
            "revenue_change": bucket.get("revenue_derivative", {}).get("value"),
            "cumulative_revenue": bucket["cumulative_revenue"]["value"]
        })
    return trend_data

# 调用示例
trend = analyze_sales_trend(
    index="orders",
    interval="day",
    start_date="2024-01-01",
    end_date="2024-01-31"
)

上述代码展示了时间序列聚合的完整实现。使用 date_histogram 按天分组，嵌套计算总销售额、平均订单金额，并使用 derivative 管道聚合计算销售额变化率，cumulative_sum 计算累计销售额。这种多层次的聚合分析，是构建数据仪表盘的核心能力。

5. 相关性评分机制：精准排序的核心

相关性评分（Relevance Scoring）是搜索引擎的核心能力，决定了搜索结果的排序质量。Elasticsearch 通过评分算法计算每个文档与查询的相关性分数，并按分数降序返回结果。

5.1 TF-IDF 算法原理

TF-IDF（Term Frequency-Inverse Document Frequency）是 Elasticsearch 早期版本使用的评分算法，至今仍是理解相关性评分的基础。

**词频（TF）**衡量词条在文档中出现的频率，出现次数越多，相关性越高。计算公式为：TF = √(词频)，使用平方根是为了避免词频过高时权重过大。

**逆文档频率（IDF）**衡量词条的区分度，出现越少的词条区分度越高。计算公式为：IDF = 1 + log(总文档数 / 包含该词的文档数)。

字段长度归一化考虑字段长度的影响，较短字段中出现的词条权重更高。计算公式为：norm = 1 / √(字段长度)。

最终的 TF-IDF 评分公式为：

score = Σ (TF × IDF × norm) = Σ (√词频 × (1 + log(N/df)) × (1/√字段长度))

5.2 BM25 算法：更先进的评分模型

从 Elasticsearch 5.0 开始，BM25（Best Matching 25）成为默认的评分算法。BM25 是对 TF-IDF 的改进，解决了 TF-IDF 的一些缺陷。

BM25 的主要改进包括：

饱和度控制：BM25 使用饱和函数控制词频的增长，避免词频过高时权重无限增长。公式为：TF_BM25 = ((k1 + 1) × tf) / (k1 × (1 - b + b × dl/avgdl) + tf)，其中 k1 控制饱和速度，默认值为 1.2。

字段长度归一化改进：BM25 将字段长度与平均长度比较，而非直接使用字段长度。参数 b 控制归一化程度，默认值为 0.75。

IDF 计算调整：BM25 的 IDF 计算考虑了文档频率的边界情况，公式为：IDF_BM25 = log(1 + (N - df + 0.5) / (df + 0.5))。

// 查看 BM25 评分详情
GET /products/_explain/{document_id}
{
  "query": {"match": {"name": "智能手机"}}
}

// 自定义 BM25 参数
PUT /products/_mapping {
  "properties": {
    "name": {"type": "text", "similarity": "my_bm25"}
  }
}

PUT /products/_settings {
  "index": {
    "similarity": {
      "my_bm25": {
        "type": "BM25",
        "k1": 1.5,
        "b": 0.5
      }
    }
  }
}

5.3 自定义评分策略

在实际业务中，单纯的文本相关性评分往往不能满足需求。电商搜索需要考虑销量、评价、价格等因素；内容搜索需要考虑时效性、热度等因素。Elasticsearch 提供了多种自定义评分方式。

function_score 查询是最灵活的自定义评分方式，支持多种评分函数：

# function_score 自定义评分示例
def search_with_custom_score(keyword, page=1, size=20):
    """使用 function_score 实现多因素综合评分"""
    query = {
        "query": {
            "function_score": {
                "query": {
                    "multi_match": {
                        "query": keyword,
                        "fields": ["name^2", "description", "tags^1.5"],
                        "type": "best_fields"
                    }
                },
                "functions": [
                    {
                        "field_value_factor": {
                            "field": "sales_count",
                            "factor": 0.1,
                            "modifier": "log1p",
                            "missing": 1
                        }
                    },
                    {
                        "field_value_factor": {
                            "field": "rating",
                            "factor": 2,
                            "modifier": "sqrt",
                            "missing": 3
                        }
                    },
                    {
                        "gauss": {
                            "created_at": {
                                "origin": "now",
                                "scale": "30d",
                                "decay": 0.5
                            }
                        }
                    },
                    {
                        "filter": {"term": {"is_hot": True}},
                        "weight": 1.5
                    },
                    {
                        "filter": {"terms": {"brand_id": ["apple", "huawei"]}},
                        "weight": 1.2
                    }
                ],
                "score_mode": "sum",
                "boost_mode": "multiply",
                "max_boost": 10
            }
        },
        "from": (page - 1) * size,
        "size": size
    }
    return es.search(index="products", body=query)

# 调用示例
result = search_with_custom_score("智能手机")
for hit in result["hits"]["hits"]:
    print(f"商品：{hit['_source']['name']}, 评分：{hit['_score']:.2f}")

上述代码展示了 function_score 的完整用法。通过 field_value_factor 使用字段值影响评分，通过 gauss 实现时间衰减，通过 filter+weight 实现条件权重提升。score_mode 控制各函数评分的组合方式，boost_mode 控制与原评分的组合方式。

script_score 允许使用 Painless 脚本实现更复杂的评分逻辑：

// script_score 自定义评分
GET /products/_search {
  "query": {
    "script_score": {
      "query": {"match": {"name": "智能手机"}},
      "script": {
        "source": """
          double score = _score;
          score += Math.log10(doc['sales_count'].value + 1) * 0.5;
          score += doc['rating'].value * 0.3;
          double priceFactor = Math.max(0, 1 - doc['price'].value / 10000);
          score += priceFactor * 0.2;
          long daysSinceCreated = (new Date().getTime() - doc['created_at'].value) / 86400000;
          if (daysSinceCreated < 7) {
            score += (7 - daysSinceCreated) * 0.1;
          }
          return score;
        """
      }
    }
  }
}

6. 性能优化：构建高性能搜索系统

性能优化是 Elasticsearch 生产部署的关键环节。从索引设计、查询优化到缓存策略，每个环节都需要精心调优。

6.1 索引设计优化

分片策略是索引设计的核心。分片数量影响并行度和资源消耗，需要根据数据量和查询特点合理规划。

字段类型选择直接影响存储效率和查询性能。keyword 类型适用于精确匹配和聚合，text 类型适用于全文检索。对于不需要搜索的字段，可以设置 "index": false 节省存储。

// 优化的索引映射设置
PUT /products {
  "settings": {
    "number_of_shards": 5,
    "number_of_replicas": 1,
    "refresh_interval": "5s",
    "index": {
      "max_result_window": 100000,
      "analysis": {
        "analyzer": {
          "ik_smart_analyzer": {
            "type": "custom",
            "tokenizer": "ik_smart"
          }
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "fields": {
          "keyword": {"type": "keyword", "ignore_above": 256}
        }
      },
      "description": {"type": "text", "analyzer": "ik_max_word"},
      "price": {"type": "scaled_float", "scaling_factor": 100},
      "sales_count": {"type": "integer", "doc_values": true},
      "category_id": {"type": "keyword"},
      "brand_id": {"type": "keyword"},
      "tags": {"type": "keyword"},
      "is_hot": {"type": "boolean"},
      "created_at": {"type": "date", "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"},
      "updated_at": {"type": "date", "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"}
    }
  }
}

6.2 查询优化策略

使用 filter 代替 query：filter 不计算评分，结果可被缓存，性能更优。对于精确匹配条件，应优先使用 filter。

避免深度分页：from+size 的分页方式在深度分页时性能急剧下降。推荐使用 search_after 或 scroll API。

批量操作：使用 bulk API 进行批量写入，减少网络开销。合理设置批量大小，一般建议 5-15MB。

# 批量写入优化示例
from elasticsearch.helpers import bulk

def bulk_index_products(products, index="products", chunk_size=500):
    """批量索引商品数据"""
    actions = []
    for product in products:
        action = {
            "_index": index,
            "_id": product["id"],
            "_source": product
        }
        actions.append(action)

        if len(actions) >= chunk_size:
            success, failed = bulk(es, actions, chunk_size=chunk_size, request_timeout=60)
            print(f"成功：{success}, 失败：{len(failed)}")
            actions = []

    if actions:
        success, failed = bulk(es, actions)
        print(f"成功：{success}, 失败：{len(failed)}")

    # 禁用 refresh 后批量写入，完成后手动刷新
    es.indices.put_settings(index="products", body={"index": {"refresh_interval": "-1"}})
    # 执行批量写入...
    # 完成后恢复 refresh
    es.indices.put_settings(index="products", body={"index": {"refresh_interval": "5s"}})
    es.indices.refresh(index="products")

6.3 缓存策略

Elasticsearch 提供了多层缓存机制：查询缓存（Query Cache）、请求缓存（Request Cache）和节点查询缓存（Node Query Cache）。

查询缓存缓存查询结果，适用于重复查询场景。请求缓存缓存完整的搜索结果，适用于仪表盘等场景。节点查询缓存缓存倒排表数据，加速 filter 查询。

// 启用请求缓存
GET /products/_search?request_cache=true
{
  "size": 0,
  "aggs": {
    "brands": {
      "terms": {"field": "brand_id"}
    }
  }
}

// 查看缓存统计
GET /products/_stats?request_cache=true

// 清除缓存
POST /products/_cache/clear

7. 实战案例：电商商品搜索系统

7.1 系统架构设计

本节以电商商品搜索系统为例，展示 Elasticsearch 在实际项目中的应用。系统需要支持关键词搜索、分类筛选、价格区间、品牌过滤、排序、分页等功能。

前端通过 API 网关访问搜索服务，搜索服务调用 Elasticsearch 进行查询。数据通过 Canal 监听 MySQL binlog，实时同步到 Elasticsearch。系统采用主从架构，主节点负责写入，从节点负责查询，实现读写分离。

7.2 核心代码实现

# 完整的电商搜索服务实现
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
from dataclasses import dataclass
from typing import List, Optional, Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class SearchParams:
    """搜索参数封装"""
    keyword: str
    category_id: Optional[str] = None
    brand_ids: Optional[List[str]] = None
    min_price: Optional[float] = None
    max_price: Optional[float] = None
    tags: Optional[List[str]] = None
    sort_by: str = "relevance"
    page: int = 1
    size: int = 20

class ProductSearchService:
    """商品搜索服务"""
    def __init__(self, es_hosts: List[str], index_name: str = "products"):
        self.es = Elasticsearch(es_hosts)
        self.index_name = index_name

    def search(self, params: SearchParams) -> Dict[str, Any]:
        """执行商品搜索"""
        query = self._build_query(params)
        try:
            response = self.es.search(
                index=self.index_name,
                body=query,
                request_cache=True
            )
            return self._parse_response(response, params)
        except Exception as e:
            logger.error(f"搜索失败：{e}")
            raise

    def _build_query(self, params: SearchParams) -> Dict[str, Any]:
        """构建查询 DSL"""
        query = {
            "query": {
                "bool": {
                    "must": self._build_must_clauses(params),
                    "should": self._build_should_clauses(params),
                    "filter": self._build_filter_clauses(params)
                }
            },
            "sort": self._build_sort(params),
            "from": (params.page - 1) * params.size,
            "size": params.size,
            "highlight": {
                "fields": {
                    "name": {},
                    "description": {"fragment_size": 150, "number_of_fragments": 3}
                },
                "pre_tags": ["<em>"],
                "post_tags": ["</em>"]
            },
            "aggs": {
                "brands": {"terms": {"field": "brand_id", "size": 20}},
                "price_ranges": {
                    "range": {
                        "field": "price",
                        "ranges": [
                            {"to": 100, "key": "0-100"},
                            {"from": 100, "to": 500, "key": "100-500"},
                            {"from": 500, "to": 1000, "key": "500-1000"},
                            {"from": 1000, "key": "1000+"}
                        ]
                    }
                }
            }
        }
        return query

    def _build_must_clauses(self, params: SearchParams) -> List[Dict]:
        clauses = []
        if params.keyword:
            clauses.append({
                "multi_match": {
                    "query": params.keyword,
                    "fields": ["name^2", "description", "tags^1.5"],
                    "type": "best_fields",
                    "operator": "and"
                }
            })
        return clauses

    def _build_should_clauses(self, params: SearchParams) -> List[Dict]:
        clauses = [
            {"term": {"is_hot": {"value": True, "boost": 1.3}}},
            {"range": {"sales_count": {"gte": 1000, "boost": 1.2}}}
        ]
        return clauses

    def _build_filter_clauses(self, params: SearchParams) -> List[Dict]:
        clauses = []
        if params.category_id:
            clauses.append({"term": {"category_id": params.category_id}})
        if params.brand_ids:
            clauses.append({"terms": {"brand_id": params.brand_ids}})
        if params.min_price is not None or params.max_price is not None:
            price_range = {}
            if params.min_price is not None:
                price_range["gte"] = params.min_price
            if params.max_price is not None:
                price_range["lte"] = params.max_price
            clauses.append({"range": {"price": price_range}})
        if params.tags:
            clauses.append({"terms": {"tags": params.tags}})
        return clauses

    def _build_sort(self, params: SearchParams) -> List[Dict]:
        sort_map = {
            "relevance": ["_score", {"sales_count": "desc"}],
            "price_asc": [{"price": "asc"}, "_score"],
            "price_desc": [{"price": "desc"}, "_score"],
            "sales": [{"sales_count": "desc"}, "_score"],
            "newest": [{"created_at": "desc"}, "_score"]
        }
        return sort_map.get(params.sort_by, sort_map["relevance"])

    def _parse_response(self, response: Dict, params: SearchParams) -> Dict:
        hits = response["hits"]
        products = []
        for hit in hits["hits"]:
            product = hit["_source"]
            product["_score"] = hit["_score"]
            if "highlight" in hit:
                product["highlight"] = hit["highlight"]
            products.append(product)

        aggregations = {}
        if "aggregations" in response:
            aggs = response["aggregations"]
            aggregations = {
                "brands": [{"key": b["key"], "count": b["doc_count"]} for b in aggs["brands"]["buckets"]],
                "price_ranges": [{"key": r["key"], "count": r["doc_count"]} for r in aggs["price_ranges"]["buckets"]]
            }
        return {
            "total": hits["total"]["value"],
            "products": products,
            "aggregations": aggregations,
            "page": params.page,
            "size": params.size,
            "has_more": hits["total"]["value"] > params.page * params.size
        }

# 使用示例
if __name__ == "__main__":
    service = ProductSearchService(
        es_hosts=["http://localhost:9200"],
        index_name="products"
    )
    params = SearchParams(
        keyword="智能手机",
        category_id="electronics",
        min_price=1000,
        max_price=5000,
        sort_by="sales",
        page=1,
        size=20
    )
    result = service.search(params)
    print(f"找到 {result['total']} 个商品")
    for product in result['products'][:5]:
        print(f"- {product['name']} (¥{product['price']})")

上述代码实现了完整的商品搜索服务。通过 SearchParams 类封装搜索参数，ProductSearchService 类提供搜索功能。支持关键词搜索、多条件过滤、多种排序、分页、高亮和聚合统计。代码结构清晰，易于扩展和维护。

8. 常见问题与解决方案

8.1 深度分页问题

问题描述：使用 from+size 进行深度分页时，性能急剧下降。Elasticsearch 默认限制 from+size 不超过 10000。

解决方案：使用 search_after 或 scroll API。

// search_after 分页
GET /products/_search {
  "size": 20,
  "sort": [{"created_at": "desc"}, "_id"],
  "search_after": ["2024-01-15T10:00:00", "abc123"]
}

8.2 数据同步延迟

问题描述：MySQL 数据更新后，Elasticsearch 中数据存在延迟。

解决方案：使用 Canal 监听 binlog 实时同步，或使用消息队列解耦。

8.3 聚合结果不准确

问题描述：terms 聚合返回的数量与实际不符。

解决方案：调整 shard_size 参数，或使用 terms 聚合的 execution_hint 参数。

// 调整 shard_size 提高准确性
GET /products/_search {
  "size": 0,
  "aggs": {
    "brands": {
      "terms": {
        "field": "brand_id",
        "size": 10,
        "shard_size": 50
      }
    }
  }
}

9. 总结

本文从倒排索引原理、全文搜索基础、聚合查询、相关性评分机制、性能优化等多个维度，全面解析了 Elasticsearch 的核心技术。通过电商商品搜索系统的实战案例，展示了 Elasticsearch 在实际项目中的应用方法。

核心要点回顾：

1. 倒排索引是 Elasticsearch 高效检索的基石，理解词条词典、倒排表、FST 等核心组件，有助于优化搜索性能。
2. 查询语法需要根据场景选择：match 用于全文检索，term 用于精确匹配，bool 用于复杂条件组合。
3. 聚合查询提供了强大的数据分析能力，terms、histogram 等聚合类型可以满足大多数分析场景。
4. 相关性评分是搜索排序的核心，BM25 算法相比 TF-IDF 更加合理，function_score 支持多因素综合评分。
5. 性能优化需要从索引设计、查询优化、缓存策略等多方面入手，构建高性能搜索系统。

思考题：

1. 在你的业务场景中，如何平衡搜索相关性与商业目标（如销量、利润）的权重？
2. 面对海量数据（TB 级别），如何设计 Elasticsearch 的索引策略以保证查询性能？
3. 如何实现搜索结果的个性化排序，让不同用户看到不同的搜索结果？

参考资料

• Elasticsearch 官方文档
• Lucene FST 实现原理
• BM25 算法论文
• Elasticsearch 权威指南
• Canal 数据同步方案