Python实战之Elasticsearch的高级实现详解

pip install elasticsearch

from elasticsearch import Elasticsearch

# 连接到本地Elasticsearch集群
es = Elasticsearch(['localhost:9200'])

# 创建名为"my_index"的索引
index_name = "my_index"
es.indices.create(index=index_name, ignore=400)

# 添加文档到索引中
doc = {
    "title": "Python实战Elasticsearch",
    "author": "John Doe",
    "content": "Elasticsearch是一个功能强大的搜索引擎。",
}
es.index(index=index_name, body=doc)

# 执行搜索查询
query = {
    "query": {
        "match": {
            "content": "搜索引擎"
        }
    }
}
result = es.search(index=index_name, body=query)
for hit in result['hits']['hits']:
    print(hit['_source'])

# 更新文档
update_doc = {
    "doc": {
        "content": "Elasticsearch是一个功能强大的搜索和分析引擎。"
    }
}
es.update(index=index_name, id='1', body=update_doc)

# 删除文档
es.delete(index=index_name, id='1')

# 删除索引
es.indices.delete(index=index_name)

# 使用批量操作
actions = [
    { "index": { "_index": index_name }},
    { "title": "文档1", "author": "Alice" },
    { "index": { "_index": index_name }},
    { "title": "文档2", "author": "Bob" },
]
es.bulk(body=actions)

try:
    # 尝试执行Elasticsearch操作
    # ...
except Exception as e:
    # 处理异常情况
    print("发生异常：", e)

# 复杂查询示例：布尔查询
query = {
    "query": {
        "bool": {
            "must": [
                { "match": { "title": "Python" }},
                { "match": { "content": "Elasticsearch" }}
            ],
            "filter": [
                { "range": { "views": { "gte": 100 }}}
            ]
        }
    }
}
result = es.search(index=index_name, body=query)

# 聚合查询示例：按作者统计文档数量
aggs_query = {
    "aggs": {
        "authors": {
            "terms": { "field": "author.keyword" }
        }
    }
}
result = es.search(index=index_name, body=aggs_query)

from elasticsearch_dsl import Search, Q

# 使用Elasticsearch DSL构建查询
s = Search(using=es, index=index_name)
s = s.query(Q("match", title="Python") & Q("match", content="Elasticsearch"))
response = s.execute()

import logging

from elasticsearch import logger as es_logger# 设置日志级别为DEBUG
logging.basicConfig(level=logging.DEBUG)

# 打印Elasticsearch客户端的日志
es_logger.setLevel(logging.DEBUG)

# 使用安全连接协议HTTPS
es = Elasticsearch(['https://localhost:9200'])