ElasticSearch DBAPI

elasticsearch-dbapi 实现了一个DBAPI（PEP-249）和SQLAlchemy方言，它允许对Elasticsearch集群进行仅查询访问的SQL访问。

在Elasticsearch上：使用Elastic X-Pack SQL API

在AWS ES、opendistro Elasticsearch上： Open Distro SQL

此库支持Elasticsearch 7.X版本。

安装

$ pip install elasticsearch-dbapi

要安装AWS Elasticsearch Service / Open Distro的支持

$ pip install elasticsearch-dbapi[opendistro]

用法

使用DBAPI

from es.elastic.api import connect

conn = connect(host='localhost')
curs = conn.cursor()
curs.execute(
    "select * from flights LIMIT 10"
)
print([row for row in curs])

使用SQLAlchemy execute

from sqlalchemy.engine import create_engine

engine = create_engine("elasticsearch+http://localhost:9200/")
rows = engine.connect().execute(
    "select * from flights LIMIT 10"
)
print([row for row in rows])

使用SQLAlchemy

from sqlalchemy import func, select
from sqlalchemy.engine import create_engine
from sqlalchemy.schema import MetaData, Table


engine = create_engine("elasticsearch+http://localhost:9200/")
logs = Table("flights", MetaData(bind=engine), autoload=True)
count = select([func.count("*")], from_obj=logs).scalar()
print(f"COUNT: {count}")

使用SQLAlchemy反射

from sqlalchemy.engine import create_engine
from sqlalchemy.schema import Table, MetaData

engine = create_engine("elasticsearch+http://localhost:9200/")
logs = Table("flights", MetaData(bind=engine), autoload=True)
print(engine.table_names())

metadata = MetaData()
metadata.reflect(bind=engine)
print([table for table in metadata.sorted_tables])
print(logs.columns)

连接参数

elasticsearch-py 用于建立连接和传输，这是官方的Elastic Python库。 Elasticsearch 构造函数接受多个可选参数，可用于正确配置您的连接，例如安全、性能和高可用性。这些可选参数可以在连接字符串中设置，例如

   elasticsearch+http://localhost:9200/?http_compress=True&timeout=100

将传输设置为使用gzip（http_compress）并将超时设置为10秒。

有关配置选项的更多信息，请参阅elasticsearch-py的文档

• 传输选项
• HTTP传输

连接字符串遵循RFC-1738，要支持多个节点，应使用sniff_*参数

获取大小

默认情况下，单个查询获取的最大行数限制为10000。这可以通过fetch_size参数进行调整

from es.elastic.api import connect

conn = connect(host="localhost", fetch_size=1000)
curs = conn.cursor()

如果需要获取超过10000行的数据，则还需要调整max_result_window。

时区

默认情况下，elasticsearch查询时区默认为Z（UTC）。这可以通过time_zone参数进行调整

from es.elastic.api import connect

conn = connect(host="localhost", time_zone="Asia/Shanghai")
curs = conn.cursor()

测试

要运行单元测试，请启动elasticsearch和kibana（kibana不是必需的，但作为附加功能很受欢迎）

$ docker-compose up -d
$ nosetests -v

sql opendistro端点（AWS ES）的特殊情况

AWS ES公开了opendistro SQL插件，并遵循不同的SQL方言。使用odelasticsearch驱动程序

from sqlalchemy.engine import create_engine

engine = create_engine(
    "odelasticsearch+https://search-SOME-CLUSTER.us-west-2.es.amazonaws.com:443/"
)
rows = engine.connect().execute(
    "select count(*), Carrier from flights GROUP BY Carrier"
)
print([row for row in rows])

或使用DBAPI

from es.opendistro.api import connect

conn = connect(host='localhost',port=9200,path="", scheme="http")

curs = conn.cursor().execute(
    "select * from flights LIMIT 10"
)

print([row for row in curs])

Opendistro（AWS ES）基本身份验证

基本身份验证配置在URI的字段中，如预期的那样

from sqlalchemy.engine import create_engine

engine = create_engine(
    "odelasticsearch+https://my_user:my_password@search-SOME-CLUSTER.us-west-2.es.amazonaws.com:443/"
)

IAM AWS身份验证密钥通过URI基本认证位置传递，并通过设置aws_keys

查询字符串键包括

• aws_keys
• aws_region

from sqlalchemy.engine import create_engine

engine = create_engine(
    "odelasticsearch+https://<AWS_ACCESS_KEY>:<AWS_SECRET_KEY>@search-SOME-CLUSTER.us-west-2.es.amazonaws.com:443/?aws_keys=1&&aws_region=<AWS_REGION>"
)

IAM AWS配置文件作为查询参数名称aws_profile在URI上配置。键的值提供AWS区域

from sqlalchemy.engine import create_engine

engine = create_engine(
    "odelasticsearch+https://search-SOME-CLUSTER.us-west-2.es.amazonaws.com:443/?aws_profile=us-west-2"
)

使用新的SQL引擎

Opendistro 1.13.0引入（默认启用）了一个新的SQL引擎，具有许多改进和修复。请参阅发行说明

此DBAPI必须对SQL v1和SQL v2的行为略有不同，默认情况下我们遵守v1，要启用v2支持，请将v2=true作为查询参数传递。

odelasticsearch+https://search-SOME-CLUSTER.us-west-2.es.amazonaws.com:443/?aws_profile=us-west-2&v2=true

要连接到提供的Opendistro ES上的docker-compose，请使用以下URI：odelasticsearch+https://admin:admin@localhost:9400/?verify_certs=False

已知限制

此库尚未支持以下功能

• 不支持数组类型列。Elasticsearch SQL也不支持它们。SQLAlchemy的get_columns将排除它们。

• object和nested列类型支持不佳，并转换为字符串

• 以.开头的索引名称

• GEO点目前不支持，并转换为字符串

• AWS ES（opendistro elasticsearch）受支持（仍处于beta阶段），已知限制包括

  • 您只能使用GROUP BY关键字字段（新的实验性 opendistro SQL已经支持此功能）
  • 不支持以点开头的索引（如'audit_log.2021.01.20'），在这些情况下我们建议使用别名