1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
| """
Elasticsearch 常用操作完整 Demo
依赖: pip install elasticsearch
运行: python es_demo.py
"""
import json
import asyncio
from datetime import datetime
from elasticsearch import AsyncElasticsearch
from elasticsearch.exceptions import NotFoundError
ES_URL = "http://localhost:9200"
ES_USER = "elastic"
ES_PASSWORD = "elastic123"
DOC_INDEX = "demo_documents"
class ESDemo:
"""Elasticsearch 常用操作演示类"""
def __init__(self):
auth = (ES_USER, ES_PASSWORD) if ES_USER else None
self.client = AsyncElasticsearch(
hosts=[ES_URL],
http_auth=auth,
verify_certs=False,
ssl_show_warn=False,
)
self.doc_index = DOC_INDEX
async def close(self):
await self.client.close()
async def create_index(self, index_name: str, mapping: dict = None):
exists = await self.client.indices.exists(index=index_name)
if exists:
print(f"ℹ️ 索引已存在: {index_name}")
return
if mapping:
await self.client.indices.create(index=index_name, body=mapping)
else:
await self.client.indices.create(index=index_name)
print(f"✅ 创建索引: {index_name}")
async def delete_index(self, index_name: str):
await self.client.indices.delete(index=index_name)
print(f"🗑️ 删除索引: {index_name}")
async def index_document(self, index: str, doc_id, body: dict):
result = await self.client.index(index=index, id=doc_id, body=body)
print(f"📝 索引文档: id={doc_id}, result={result['result']}")
await self.client.indices.refresh(index=index)
return result
async def get_document(self, index: str, doc_id):
try:
result = await self.client.get(index=index, id=doc_id)
return result["_source"]
except NotFoundError:
return None
async def update_document(self, index: str, doc_id, update_data: dict):
result = await self.client.update(index=index, id=doc_id, body={"doc": update_data})
await self.client.indices.refresh(index=index)
return result
async def multi_match_search(self, index: str, query: str, fields: list = None):
"""多字段全文搜索 + 高亮 + 模糊匹配"""
if fields is None:
fields = ["title^2", "content"]
body = {
"query": {
"multi_match": {
"query": query,
"fields": fields,
"type": "best_fields",
"fuzziness": "AUTO"
}
},
"highlight": {
"fields": {f: {"fragment_size": 80} for f in fields}
},
"size": 5
}
result = await self.client.search(index=index, body=body)
print(f"\n🔍 搜索: \"{query}\" (共 {result['hits']['total']['value']} 条结果)")
for hit in result["hits"]["hits"]:
print(f" 🌟 高亮: {hit.get('highlight', {})}")
return result
async def similar_search(self, index: str, doc_id, size: int = 3):
"""相似文档推荐 (More Like This)"""
body = {
"query": {
"more_like_this": {
"fields": ["title", "content"],
"like": [{"_index": index, "_id": doc_id}],
"min_term_freq": 1,
"max_query_terms": 12,
}
},
"size": size
}
result = await self.client.search(index=index, body=body)
print(f"\n🔍 相似文档推荐 (id={doc_id}):")
for hit in result["hits"]["hits"]:
print(f" - [{hit['_score']:.2f}] {hit['_source'].get('title', '')}")
async def aggregate_by_field(self, index: str, field: str):
"""按字段聚合统计 (类似 SQL 的 Group By)"""
body = {
"size": 0,
"aggs": {
f"by_{field}": {"terms": {"field": field, "size": 10}}
}
}
result = await self.client.search(index=index, body=body)
buckets = result["aggregations"][f"by_{field}"]["buckets"]
print(f"\n📊 聚合统计: {field}")
for bucket in buckets:
print(f" - {bucket['key']}: {bucket['doc_count']} 条")
async def run_full_demo(self):
print("="*50)
print("🚀 Elasticsearch 异步操作演示开始")
print("="*50)
doc_mapping = {
"mappings": {
"properties": {
"title": {"type": "text", "analyzer": "standard"},
"content": {"type": "text", "analyzer": "standard"},
"category": {"type": "keyword"},
"author": {"type": "keyword"},
"price": {"type": "float"},
"created_at": {"type": "date"},
}
},
"settings": {"number_of_shards": 1, "number_of_replicas": 0}
}
await self.create_index(self.doc_index, doc_mapping)
docs = [
{"id": 1, "title": "Python 异步编程指南", "content": "asyncio 是 Python 标准库中的异步 I/O 框架。", "category": "编程", "author": "张三", "price": 39.9},
{"id": 2, "title": "Elasticsearch 权威指南", "content": "ES 是一个分布式搜索和分析引擎。", "category": "编程", "author": "李四", "price": 59.9},
{"id": 3, "title": "机器学习入门", "content": "机器学习是人工智能的一个分支。", "category": "AI", "author": "王五", "price": 49.9},
]
for doc in docs:
body = {k: v for k, v in doc.items() if k != "id"}
body["created_at"] = datetime.now().isoformat()
await self.index_document(self.doc_index, doc["id"], body)
await self.multi_match_search(self.doc_index, "asyncio 框架")
await self.similar_search(self.doc_index, 1)
await self.aggregate_by_field(self.doc_index, "category")
await self.delete_index(self.doc_index)
print("\n🎉 ES 演示完成!")
async def main():
demo = ESDemo()
try:
await demo.run_full_demo()
finally:
await demo.close()
if __name__ == "__main__":
asyncio.run(main())
|
