Elasticsearch 8.x + Python 官方客户端实战教程

Elasticsearch 8.x + Python 官方客户端实战教程

一、Elasticsearch 基础概念（工程视角）

1. Index（索引）

• 类似数据库中的表
• 一个 Index 通常对应一个业务实体集合，例如：
  • user_logs
  • order_records

⚠️工程建议：

• 一个索引只服务一种查询模式
• 不要把“所有数据”塞进一个 index

2. Document（文档）

• 一条 JSON 数据
• 每条文档都有_id

{"user_id":"u123","action":"login","timestamp":"2025-01-01T10:00:00"}

3. Mapping（字段结构）

• 相当于表结构定义
• 决定：
  • 是否可搜索
  • 是否可聚合
  • 是否支持排序

❗一旦字段类型确定，几乎不可修改

4. Shard / Replica（分片 & 副本）

• shard：数据拆分单元（影响写入 & 查询性能）
• replica：副本（高可用 & 查询吞吐）

通用建议：

• 中小业务：1~3shards
• 副本数：1

二、Elasticsearch 8.x 架构与安全机制

从8.x 开始默认开启安全机制：

• HTTPS
• 用户认证
• API 访问权限

👉Python 客户端必须显式配置认证

三、Python 官方客户端安装（8.x）

pipinstallelasticsearch>=8.12.0

验证版本：

fromelasticsearchimport__version__print(__version__)

四、Python 连接 Elasticsearch（生产可用）

1. 基础连接（用户名 + 密码）

fromelasticsearchimportElasticsearch es=Elasticsearch(hosts=["https://localhost:9200"],basic_auth=("elastic","your_password"),verify_certs=False# 本地测试可关闭，生产请开启)print(es.info())

2. HTTPS + CA 证书（生产推荐）

es=Elasticsearch(hosts=["https://es-prod.example.com:9200"],basic_auth=("elastic","password"),ca_certs="/etc/elasticsearch/certs/http_ca.crt")

五、索引设计与创建（真实工程示例）

1. 索引 Mapping 设计

INDEX_NAME="user_action_logs"mapping={"settings":{"number_of_shards":2,"number_of_replicas":1},"mappings":{"properties":{"user_id":{"type":"keyword"},"action":{"type":"keyword"},"message":{"type":"text"},"ip":{"type":"ip"},"created_at":{"type":"date"}}}}

2. 创建索引（幂等）

ifnotes.indices.exists(index=INDEX_NAME):es.indices.create(index=INDEX_NAME,body=mapping)

六、数据写入（单条 & 批量）

1. 单条写入（index）

es.index(index=INDEX_NAME,id="u123_001",document={"user_id":"u123","action":"login","message":"user login success","ip":"127.0.0.1","created_at":"2025-01-01T10:00:00"})

2. 批量写入（bulk，生产必用）

fromelasticsearch.helpersimportbulk actions=[]foriinrange(1000):actions.append({"_index":INDEX_NAME,"_id":f"u123_{i}","_source":{"user_id":"u123","action":"click","message":f"click{i}","ip":"127.0.0.1","created_at":"2025-01-01T10:00:00"}})bulk(es,actions,request_timeout=60)

工程建议：

• 单批 500~2000 条
• 明确_id，保证幂等

七、查询（核心能力）

1. 基础查询

res=es.search(index=INDEX_NAME,query={"term":{"action":"login"}})

2. Bool 查询（真实业务）

res=es.search(index=INDEX_NAME,query={"bool":{"must":[{"term":{"user_id":"u123"}}],"filter":[{"range":{"created_at":{"gte":"now-1d"}}}]}},size=20,sort=[{"created_at":"desc"}])

3. 聚合查询

res=es.search(index=INDEX_NAME,size=0,aggs={"action_count":{"terms":{"field":"action"}}})

八、深分页解决方案

❌ from + size（禁止大页）

• 超过 10k 会严重影响性能

search_after（推荐）

res=es.search(index=INDEX_NAME,size=10,sort=[{"created_at":"desc"},{"_id":"desc"}])last_sort=res["hits"]["hits"][-1]["sort"]res_next=es.search(index=INDEX_NAME,size=10,search_after=last_sort,sort=[{"created_at":"desc"},{"_id":"desc"}])

九、工程级 Client 封装示例

classESClient:def__init__(self,hosts,username,password,ca_certs=None):self.client=Elasticsearch(hosts=hosts,basic_auth=(username,password),ca_certs=ca_certs,)defindex_doc(self,index,doc_id,doc):returnself.client.index(index=index,id=doc_id,document=doc)defsearch(self,index,query,size=10):returnself.client.search(index=index,query=query,size=size)

十、常见踩坑总结（8.x）

1. ❌ 使用body=写 search（已废弃）
2. ❌ 忽略 HTTPS / 认证
3. ❌ 动态 Mapping 失控
4. ❌ 大分页 from + size

十一、结语

如果你：

• 正确设计索引
• 使用 bulk 写入
• 使用 bool + filter 查询
• 合理分页