# Python连接Elasticsearch 8.x完整指南 ## 1. 环境准备与依赖安装 ### 1.1 安装Elasticsearch Python客户端 要连接Elasticsearch 8.x，首先需要安装官方的Python客户端库： ```python # 安装elasticsearch库 pip install elasticsearch # 如果需要更高级的功能，可以安装完整套件 pip install elasticsearch[async] ``` ### 1.2 版本兼容性说明 Elasticsearch 8.x在安全配置方面有重大变化，需要特别注意版本兼容性： | ES版本 | Python客户端版本 | 主要特性 | |--------|------------------|----------| | 8.0-8.4 | 8.x | 强制TLS，内置安全 | | 8.5+ | 8.5+ | 增强的安全特性 | ## 2. 连接配置方式 ### 2.1 基础连接（关闭安全模式） 在开发环境中，可以暂时关闭Elasticsearch的安全特性进行连接： ```python from elasticsearch import Elasticsearch # 基础连接（安全关闭时使用） es = Elasticsearch( ["http://localhost:9200"], verify_certs=False # 不验证证书 ) ``` **配置Elasticsearch关闭安全模式**（需要在ES配置文件中设置）： ```yaml # elasticsearch.yml xpack.security.enabled: false xpack.security.enrollment.enabled: false ``` ### 2.2 用户名密码认证连接 在生产环境中，推荐使用用户名密码进行认证： ```python from elasticsearch import Elasticsearch # 用户名密码认证 es = Elasticsearch( "https://localhost:9200", http_auth=('elastic', 'your_password'), verify_certs=False ) ``` ### 2.3 HTTPS与证书认证 Elasticsearch 8.x默认启用HTTPS，可以使用证书指纹或证书文件进行连接： ```python from elasticsearch import Elasticsearch # 使用证书指纹连接 es = Elasticsearch( "https://localhost:9200", http_auth=('elastic', 'your_password'), ssl_assert_fingerprint="your_fingerprint" ) # 使用CA证书连接 es = Elasticsearch( "https://localhost:9200", http_auth=('elastic', 'your_password'), ca_certs="/path/to/http_ca.crt" ) ``` ### 2.4 API Key认证方式 ```python from elasticsearch import Elasticsearch # API Key认证 es = Elasticsearch( "https://localhost:9200", api_key=('api_key_id', 'api_key_secret') ) ``` ## 3. 完整操作示例 ### 3.1 创建索引与映射 ```python # 创建索引并定义映射 index_name = "test_index" mapping = { "mappings": { "properties": { "title": {"type": "text"}, "content": {"type": "text"}, "timestamp": {"type": "date"}, "views": {"type": "integer"} } } } # 创建索引 if not es.indices.exists(index=index_name): es.indices.create(index=index_name, body=mapping) print(f"索引 {index_name} 创建成功") ``` ### 3.2 文档操作 #### 插入单个文档 ```python # 插入文档 document = { "title": "Python连接ES教程", "content": "这是一篇关于如何使用Python连接Elasticsearch 8.x的教程", "timestamp": "2024-01-01T10:00:00", "views": 100 } response = es.index(index=index_name, document=document) print(f"文档插入成功，ID: {response['_id']}") ``` #### 批量插入文档 ```python from elasticsearch.helpers import bulk # 批量插入数据 documents = [ {"_index": index_name, "_source": {"title": f"文档{i}", "content": f"内容{i}", "views": i}} for i in range(10) ] success, failed = bulk(es, documents) print(f"成功插入: {success}, 失败: {failed}") ``` ### 3.3 查询操作 #### 基础查询 ```python # 匹配查询 query = { "query": { "match": { "title": "Python" } } } response = es.search(index=index_name, body=query) print(f"找到 {response['hits']['total']['value']} 个文档") for hit in response['hits']['hits']: print(f"ID: {hit['_id']}, 分数: {hit['__score']}") print(f"内容: {hit['_source']}") ``` #### 复合查询 ```python # 布尔查询 bool_query = { "query": { "bool": { "must": [ {"match": {"title": "教程"}} ], "filter": [ {"range": {"views": {"gte": 50}}} ] } } } response = es.search(index=index_name, body=bool_query) ``` ### 3.4 更新与删除 ```python # 更新文档 update_body = { "doc": { "views": 150, "updated": True } } es.update(index=index_name, id="document_id", body=update_body) # 删除文档 es.delete(index=index_name, id="document_id") ``` ## 4. 高级功能 ### 4.1 异步客户端 ```python from elasticsearch import AsyncElasticsearch import asyncio async def async_example(): es = AsyncElasticsearch("https://localhost:9200") # 异步查询 response = await es.search(index=index_name, body={"query": {"match_all": {}}}) print(response['hits']['total']['value']) await es.close() # 运行异步示例 asyncio.run(async_example()) ``` ### 4.2 连接池与集群配置 ```python from elasticsearch import Elasticsearch # 多节点集群配置 es = Elasticsearch([ {"host": "node1", "port": 9200}, {"host": "node2", "port": 9200}, {"host": "node3", "port": 9200} ], http_auth=('elastic', 'password')) # 连接池配置 es = Elasticsearch( ["https://localhost:9200"], http_auth=('elastic', 'password'), sniff_on_start=True, # 启动时嗅探节点 sniff_on_connection_fail=True, # 连接失败时嗅探 sniffer_timeout=60 # 嗅探超时时间 ) ``` ## 5. 错误处理与监控 ### 5.1 异常处理 ```python from elasticsearch import Elasticsearch, NotFoundError, ConnectionError try: es = Elasticsearch("https://localhost:9200") # 测试连接 if not es.ping(): raise ValueError("无法连接到Elasticsearch") # 执行操作 info = es.info() print(f"集群名称: {info['cluster_name']}") print(f"ES版本: {info['version']['number']}") except ConnectionError as e: print(f"连接错误: {e}") except NotFoundError as e: print(f"文档不存在: {e}") except Exception as e: print(f"其他错误: {e}") ``` ### 5.2 健康检查与监控 ```python # 检查集群健康状态 health = es.cluster.health() print(f"集群状态: {health['status']}") print(f"节点数: {health['number_of_nodes']}") # 检查索引状态 stats = es.indices.stats(index=index_name) print(f"文档总数: {stats['_all']['primaries']['docs']['count']}") ``` ## 6. 实际应用场景 ### 6.1 日志分析系统 ```python class LogAnalyzer: def __init__(self, es_client): self.es = es_client self.log_index = "application_logs" def index_log(self, level, message, timestamp): log_doc = { "level": level, "message": message, "@timestamp": timestamp, "host": "server01" } return self.es.index(index=self.log_index, document=log_doc) def search_logs(self, level=None, time_range=None): query = {"query": {"bool": {"must": []}}} if level: query["query"]["bool"]["must"].append({"term": {"level": level}}) if time_range: query["query"]["bool"]["must"].append({ "range": {"@timestamp": time_range} }) return self.es.search(index=self.log_index, body=query) ``` ### 6.2 电商搜索实现 ```python class ProductSearch: def __init__(self, es_client): self.es = es_client self.product_index = "products" def search_products(self, keyword, category=None, price_range=None): query = { "query": { "bool": { "must": [ {"multi_match": { "query": keyword, "fields": ["name", "description", "brand"] }} ], "filter": [] } }, "sort": [{"_score": {"order": "desc"}}] } if category: query["query"]["bool"]["filter"].append({"term": {"category": category}}) if price_range: query["query"]["bool"]["filter"].append({ "range": {"price": price_range} }) return self.es.search(index=self.product_index, body=query) ``` 通过以上完整的配置和示例，您可以成功使用Python连接和操作Elasticsearch 8.x。记得根据您的具体环境调整安全配置和连接参数，特别是在生产环境中要确保使用适当的安全措施。