文档首页/ 云搜索服务 CSS/ 用户指南/ Elasticsearch/ 访问Elasticsearch集群/ 通过Python客户端接入Elasticsearch集群
更新时间：2026-01-09 GMT+08:00
查看PDF
分享

通过Python客户端接入Elasticsearch集群

在使用CSS服务的Elasticsearch集群时，开发者通常需要通过编程语言进行数据访问和管理操作。然而，如果直接使用Elasticsearch的HTTP API，不仅代码复杂度高，而且容易出错。为了简化开发流程，CSS服务的Elasticsearch集群支持通过官方Elasticsearch Python客户端进行数据访问和管理操作。该客户端库（elasticsearch-py）提供了完整的Elasticsearch API封装，开发者可通过Python原生接口实现索引管理、文档CRUD、搜索查询等操作。详细使用指导请参见Elasticsearch官方Python客户端文档。

前提条件

  • Elasticsearch集群处于可用状态。
  • 运行Python代码的服务器与Elasticsearch集群之间网络互通。
  • 根据集群的网络配置方式，获取集群的访问地址，操作指导请参见网络配置。
  • 确认服务器已安装Python 3。

安装Python客户端库

在运行Python代码的服务器中安装Python客户端库。为确保更好的兼容性，建议使用与Elasticsearch集群同版本的Python客户端库。

使用时需要将7.10替换为实际Python客户端的版本号。

pip install Elasticsearch==7.10

接入集群

当使用Python客户端访问不同安全类型的Elasticsearch集群时，示例代码会有差异，请根据实际业务场景选择参考文档。

表1 接入集群的场景介绍

Elasticsearch集群安全类型

参考文档

非安全模式

Elasticsearch Python连接非安全集群

安全模式+HTTP协议

Elasticsearch Python连接HTTP协议的安全集群

安全模式+HTTPS协议

Elasticsearch Python连接HTTPS协议的安全集群

Elasticsearch Python连接非安全集群

使用Elasticsearch Python客户端连接非安全模式的Elasticsearch集群，并查询索引“test”是否存在。

使用集群连接信息创建客户端实例的代码示例如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
 
from elasticsearch import Elasticsearch

class ElasticFactory(object):

    def __init__(self, host: list, port: str, username: str, password: str):
        self.port = port
        self.host = host
        self.username = username
        self.password = password

    def create(self) -> Elasticsearch:
        addrs = []
        for host in self.host:
            addr = {'host': host, 'port': self.port}
            addrs.append(addr)

        if self.username and self.password:
            elasticsearch = Elasticsearch(addrs, http_auth=(self.username, self.password))
        else:
            elasticsearch = Elasticsearch(addrs)
        return elasticsearch

es = ElasticFactory(["{集群访问地址}"], "9200", None, None).create()
print(es.indices.exists(index='test'))

该程序为判断集群是否存在test索引，当返回“true”（索引存在）或“false”（索引不存在）时，表示正常返回查询结果，集群连接成功。

Elasticsearch Python连接HTTP协议的安全集群

使用Elasticsearch Python客户端连接安全模式+HTTP协议的Elasticsearch集群，并查询索引“test”是否存在。

使用集群连接信息创建客户端实例的代码示例如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
 
from elasticsearch import Elasticsearch

class ElasticFactory(object):

    def __init__(self, host: list, port: str, username: str, password: str):
        self.port = port
        self.host = host
        self.username = username
        self.password = password

    def create(self) -> Elasticsearch:
        addrs = []
        for host in self.host:
            addr = {'host': host, 'port': self.port}
            addrs.append(addr)

        if self.username and self.password:
            elasticsearch = Elasticsearch(addrs, http_auth=(self.username, self.password))
        else:
            elasticsearch = Elasticsearch(addrs)
        return elasticsearch

es = ElasticFactory(["xxx.xxx.xxx.xxx"], "9200", "username", "password").create()
print(es.indices.exists(index='test'))
表2 函数中的变量说明

参数

描述

host

集群的访问地址，当存在多个IP地址时，中间用“,”隔开。

port

集群的连接端口，填写“9200”。

username

访问集群的用户名。

password

用户名对应的密码。

该程序为判断集群是否存在test索引，当返回“true”（索引存在）或“false”（索引不存在）时，表示正常返回查询结果，集群连接成功。

Elasticsearch Python连接HTTPS协议的安全集群

使用Elasticsearch Python客户端连接安全模式+HTTPS协议的Elasticsearch集群，并查询索引“test”是否存在。

使用集群连接信息创建客户端实例的代码示例如下：

 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
 
from elasticsearch import Elasticsearch
import ssl

class ElasticFactory(object):

    def __init__(self, host: list, port: str, username: str, password: str):
        self.port = port
        self.host = host
        self.username = username
        self.password = password

    def create(self) -> Elasticsearch:
        context = ssl._create_unverified_context()

        addrs = []
        for host in self.host:
            addr = {'host': host, 'port': self.port}
            addrs.append(addr)

        if self.username and self.password:
            elasticsearch = Elasticsearch(addrs, http_auth=(self.username, self.password), scheme="https", ssl_context=context)
        else:
            elasticsearch = Elasticsearch(addrs)
        return elasticsearch

es = ElasticFactory(["xxx.xxx.xxx.xxx"], "9200", "username", "password").create()
print(es.indices.exists(index='test'))
表3 函数中的变量说明

参数

描述

host

集群的访问地址，当存在多个IP地址时，中间用“,”隔开。

port

集群的连接端口，填写“9200”。

username

访问集群的用户名。

password

用户名对应的密码。

该程序为判断集群是否存在test索引，当返回“true”（索引存在）或“false”（索引不存在）时，表示正常返回查询结果，集群连接成功。

相关文档