Python Elasticsearch DSL如何使用-Python教程-PHP中文网

Python Elasticsearch DSL如何使用

舞夢輝影

发布： 2025-11-15 23:29:02

原创

458人浏览过

答案：Python Elasticsearch DSL 提供了更便捷的面向对象方式操作 Elasticsearch，通过安装 elasticsearch-dsl 库并连接服务后，可定义 Document 模型映射字段与索引设置，调用 init() 创建索引，使用 save() 添加文档，Search 类构建 match、term、range 等查询，支持 bool 组合与聚合分析，还可通过 get() 更新或 delete() 删除文档，结合 bulk 实现高效批量写入，适用于复杂检索场景。

python elasticsearch dsl如何使用

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 类定义索引结构和字段类型：

from elasticsearch_dsl import Document, Text, Keyword, Integer, Date

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 类进行复杂查询：

from elasticsearch_dsl import Search

s = Search().index('articles')
s = s.query("match", title="Python")
response = s.execute()

遍历结果：

如知AI笔记

如知笔记——支持markdown的在线笔记，支持ai智能写作、AI搜索，支持DeepseekR1满血大模型

查看详情

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 Q

s = 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.helpers import 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 的批量支持（较旧版本）或自定义生成器。