Python Elasticsearch DSL使用教程 - 高效操作Elasticsearch指南

pip install elasticsearch elasticsearch-dsl

from elasticsearch import Elasticsearch
from elasticsearch_dsl import connections

# 创建默认连接
connections.create_connection(hosts=['localhost:9200'])

# 或者使用Elasticsearch客户端
es = Elasticsearch(['localhost:9200'])

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

class Article(Document):
    title = Text(analyzer='snowball', fields={'raw': Keyword()})
    author = Keyword()
    content = Text(analyzer='snowball')
    published_at = Date()
    tags = Keyword(multi=True)
    views = Integer()
    
    class Index:
        name = 'blog'
        settings = {
            "number_of_shards": 2,
            "number_of_replicas": 1
        }
        http://inanxing.com
http://lyfezx.com

# 创建索引（如果不存在）
Article.init()

article = Article(
    title='Python Elasticsearch DSL教程',
    author='张三',
    content='本教程将教你如何使用Elasticsearch DSL库...',
    published_at='2023-06-15',
    tags=['python', 'elasticsearch', '教程'],
    views=1500
)

# 保存文档到Elasticsearch
article.save()

from elasticsearch.helpers import bulk
from elasticsearch_dsl import Index

# 批量导入文档
articles = [
    Article(title=f"文章 {i}", author=f"作者 {i%5}") 
    for i in range(100)
]

for a in articles:
    a.save()

from elasticsearch_dsl import Search

# 创建Search对象
s = Search(index='blog')

# 查询所有文档
response = s.execute()

for hit in response:
    print(hit.title)

from elasticsearch_dsl import Q

# 创建多条件查询
s = Search(index='blog').query(
    Q('match', title='python') | 
    Q('match', content='elasticsearch')
).filter('range', views={'gte': 100})

# 分页设置
s = s[0:10]

# 执行查询
response = s.execute()

from elasticsearch_dsl import A

# 按作者聚合文章数量
s = Search(index='blog')
a = A('terms', field='author.keyword', size=5)
s.aggs.bucket('authors', a)

response = s.execute()

# 打印聚合结果
for author in response.aggregations.authors.buckets:
    print(f"{author.key}: {author.doc_count}篇文章")

# 获取文档
article = Article.get(id='some_id')

# 更新字段
article.views += 1
article.save()

# 删除单个文档
article = Article.get(id='some_id')
article.delete()

# 按查询删除
s = Search(index='blog').query('match', author='张三')
s.delete()