答案：Python Elasticsearch DSL 提供了更便捷的面向对象方式操作 Elasticsearch，通过安装 elasticsearch-dsl 库并连接服务后，可定义 Document 模型映射字段与索引设置，调用 init() 创建索引，使用 save() 添加文档，Search 类构建 match、term、range 等查询，支持 bool 组合与聚合分析，还可通过 get() 更新或 delete() 删除文档，结合 bulk 实现高效批量写入，适用于复杂检索场景。
Python Elasticsearch DSL 是一个用于与 Elasticsearch 进行交互的高级库，它封装了官方的 elasticsearch-py 客户端，提供了更直观、更 Pythonic 的方式来构建查询和操作数据。下面介绍如何使用它进行常见操作。
安装依赖
首先需要安装 elasticsearch-dsl：
pip install elasticsearch-dsl确保你已经运行了 Elasticsearch 服务（例如本地启动在 http://localhost:9200）。
连接到 Elasticsearch
使用 connections 模块配置连接：
立即学习“Python免费学习笔记（深入）”；
from elasticsearch_dsl import connections# 连接到本地 ES 实例
connections.create_connection(hosts=['localhost'], port=9200, timeout=20)
也可以传入多个节点或使用 URL：
connections.create_connection(hosts=['http://user:pass@localhost:9200'])定义文档模型（Document）
通过继承 Document 类定义索引结构和字段类型：
class Article(Document):
title = Text(analyzer='ik_max_word', search_analyzer='ik_smart')
content = Text(analyzer='ik_max_word')
author = Keyword()
views = Integer()
created_at = Date()
class Index:
name = 'articles' # 索引名称
settings = {
"number_of_shards": 1,
"number_of_replicas": 0
}
说明：
-
Text：会被分词，适合全文检索 -
Keyword：不分词，用于精确匹配（如作者名） -
Date、Integer等对应基本类型 -
Index.name指定索引名 - 可设置分词器（如中文推荐 ik 分词器）
创建索引（映射）：
Article.init() # 创建索引并应用 mapping添加和保存文档
创建并保存一条记录：
article = Article(title="Python 使用 Elasticsearch",
content="本文介绍 elasticsearch-dsl 的用法",
author="张三",
views=100,
created_at="2025-04-05"
)
article.save() # 写入 ES
指定 ID 保存：
article.save(id=1)查询数据（Search API）
使用 Search 类进行复杂查询：
s = Search().index('articles')
s = s.query("match", title="Python")
response = s.execute()
遍历结果：
for hit in response:
print(hit.title, hit.author, hit.meta.score)
常用查询类型：
-
query("match", field="value")：全文匹配 -
query("term", author="张三")：精确匹配（Keyword 字段） -
query("range", views={"gte": 50})：范围查询 -
query(~Q("match", title="Elasticsearch"))：使用~表示否定
组合查询（bool 查询）：
from elasticsearch_dsl import Qs = Search().index('articles')
s = s.query('bool',
must=[Q('match', title='Python')],
filter=[Q('term', author='张三')])
聚合操作（Aggregations）
执行统计聚合：
s.aggs.bucket('by_author', 'terms', field='author')s.aggs.metric('avg_views', 'avg', field='views')
获取聚合结果：
response = s.execute()for bucket in response.aggregations.by_author.buckets:
print(bucket.key, bucket.doc_count)
更新和删除文档
根据 ID 获取并修改：
article = Article.get(id=1)article.views = 150
article.save()
部分更新（仅更新某些字段）：
article.update(views=200)删除文档：
article.delete()删除整个索引：
Article._index.delete()批量操作（Bulk Operations）
使用 bulk 批量写入提升性能：
from elasticsearch_dsl import Index
def bulk_index(data_list):
actions = []
for data in data_list:
action = {
"_index": "articles",
"_source": data
}
actions.append(action)
bulk(client=connections.get_connection(), actions=actions)
或者使用 DocType 的批量支持（较旧版本）或自定义生成器。
小贴士
- 中文分词建议安装并启用 IK 分词器插件
- 生产环境注意设置合理的超时和重试机制
- 可以结合 Django 或 Flask 使用，有相应集成包
- 使用
.scan()遍历大量数据（替代 from/size 分页） - 调试时可用
print(s.to_dict())查看生成的 DSL 查询语句
基本上就这些。掌握定义模型、增删改查、构造查询和聚合，就能高效使用 elasticsearch-dsl。
