Elasticsearch Python 索引与搜索

目标

使用 Python elasticsearch 客户端：创建索引、写入文档、全文搜索、聚合查询。

环境准备

pip install elasticsearch
# 确保 ES 运行中
docker run -d --name es -p 9200:9200 -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" elasticsearch:8.12.0

完整代码

基础操作（es_basic.py）

from elasticsearch import Elasticsearch

es = Elasticsearch('http://localhost:9200')

# 创建索引
mapping = {
    "mappings": {
        "properties": {
            "title": {"type": "text", "analyzer": "ik_max_word"},
            "content": {"type": "text", "analyzer": "ik_max_word"},
            "price": {"type": "float"},
            "category": {"type": "keyword"},
            "created_at": {"type": "date"}
        }
    }
}
es.indices.create(index='products', body=mapping, ignore=400)

# 批量写入
from elasticsearch.helpers import bulk

docs = [
    {"_index": "products", "_id": 1, "title": "STM32F103开发板", "content": "ARM Cortex-M3 72MHz", "price": 49.9, "category": "开发板"},
    {"_index": "products", "_id": 2, "title": "树莓派4B", "content": "四核 Cortex-A72 4GB RAM", "price": 299, "category": "开发板"},
    {"_index": "products", "_id": 3, "title": "USB转TTL模块", "content": "CH340G 3.3V/5V", "price": 9.9, "category": "连接器"},
]
bulk(es, docs)

# 全文搜索
query = {
    "query": {
        "multi_match": {
            "query": "Cortex ARM",
            "fields": ["title^2", "content"]  # title 权重翻倍
        }
    },
    "highlight": {
        "fields": {"content": {}}
    }
}
result = es.search(index='products', body=query)
for hit in result['hits']['hits']:
    print(f"⭐ {hit['_source']['title']} (评分: {hit['_score']:.2f})")

# 聚合查询
agg_query = {
    "size": 0,
    "aggs": {
        "by_category": {
            "terms": {"field": "category"},
            "aggs": {
                "avg_price": {"avg": {"field": "price"}},
                "price_range": {
                    "range": {"field": "price", "ranges": [
                        {"to": 50}, {"from": 50, "to": 200}, {"from": 200}
                    ]}
                }
            }
        }
    }
}
result = es.search(index='products', body=agg_query)
for bucket in result['aggregations']['by_category']['buckets']:
    print(f"{bucket['key']}: {bucket['doc_count']} 个商品, 均价 ¥{bucket['avg_price']['value']:.2f}")

预期输出

搜索 "Cortex ARM" 会按相关度排序返回开发板。聚合查询按分类汇总商品数量和均价。