在我之前的文章 “Elasticsearch：ES|QL 查询展示”，我展示了如何在 Kibana 中使用 ES|QL 对索引来进行查询及统计。在很多的情况下，我们需要在客户端中来对数据进行查询，那么我们该怎么办呢？我们需要使用到 Elasticsearch 的客户端。在今天的文章中，我们来展示如何使用 Python 来对数据进行查询。
注意：为了使用 ES|QL，我们的 Elastic Stack 版本至少在 8.12 及以上。
安装
如果你还没有安装好自己的 Elasticsearch 及 Kibana，请参考如下的链接来进行安装：
- 如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch
- Kibana：如何在 Linux，MacOS 及 Windows上安装 Elastic 栈中的 Kibana
在安装的时候，我们选择 Elastic Stack 8.x 来进行安装。特别值得指出的是：ES|QL 只在 Elastic Stack 8.11 及以后得版本中才有。你需要下载 Elastic Stack 8.11 及以后得版本来进行安装。
在首次启动 Elasticsearch 的时候，我们可以看到如下的输出：
我们需要记下 Elasticsearch 超级用户 elastic 的密码。
我们还需要安装 Elasticsearch 的 python 依赖包：
pip3 install elasticsearch==8.12.1$ pip3 list | grep elasticsearch
elasticsearch 8.12.1准备数据
我们参考之前的文章 “Elasticsearch：ES|QL 查询展示” 来创建索引：
PUT sample_data
{
"mappings": {
"properties": {
"client.ip": {
"type": "ip"
},
"message": {
"type": "keyword"
}
}
}
}PUT sample_data/_bulk
{"index": {}}
{"@timestamp": "2023-10-23T12:15:03.360Z", "client.ip": "172.21.2.162", "message": "Connected to 10.1.0.3", "event.duration": 3450233}
{"index": {}}
{"@timestamp": "2023-10-23T12:27:28.948Z", "client.ip": "172.21.2.113", "message": "Connected to 10.1.0.2", "event.duration": 2764889}
{"index": {}}
{"@timestamp": "2023-10-23T13:33:34.937Z", "client.ip": "172.21.0.5", "message": "Disconnected", "event.duration": 1232382}
{"index": {}}
{"@timestamp": "2023-10-23T13:51:54.732Z", "client.ip": "172.21.3.15", "message": "Connection error", "event.duration": 725448}
{"index": {}}
{"@timestamp": "2023-10-23T13:52:55.015Z", "client.ip": "172.21.3.15", "message": "Connection error", "event.duration": 8268153}
{"index": {}}
{"@timestamp": "2023-10-23T13:53:55.832Z", "client.ip": "172.21.3.15", "message": "Connection error", "event.duration": 5033755}
{"index": {}}
{"@timestamp": "2023-10-23T13:55:01.543Z", "client.ip": "172.21.3.15", "message": "Connected to 10.1.0.1", "event.duration": 1756467}使用 Elasticsearch 客户端来进行查询
Elasticsearch 查询语言 (ES|QL) 提供了一种强大的方法来过滤、转换和分析 Elasticsearch 中存储的数据。 它旨在易于最终用户、SRE 团队、应用程序开发人员和管理员学习和使用。 但它也非常适合熟悉 Pandas 和其他基于数据框的框架的数据科学家。
事实上，ES|QL 查询会生成带有命名列的表，即数据帧。 但是如何使用 Python 处理这些数据呢？ ES|QL 目前没有 Apache Arrow 输出，但 CSV 输出是一个很好的开始。
我们使用如下的测试程序：
esql.py
from io import StringIO
import numpy as np
import os
from elasticsearch import Elasticsearch
import pandas as pd
endpoint = os.getenv("ES_SERVER")
username = os.getenv("ES_USER")
password = os.getenv("ES_PASSWORD")
fingerprint = os.getenv("ES_FINGERPRINT")
url = f"https://{endpoint}:9200"
es = Elasticsearch( url ,
basic_auth = (username, password),
ssl_assert_fingerprint = fingerprint,
http_compress = True )
# print(es.info())
response = es.esql.query(query="FROM sample_data", format="csv")
df = pd.read_csv(StringIO(response.body))
print(df)
print("==================================================================")
response = es.esql.query(
query="""
FROM sample_data
| LIMIT 5
| sort @timestamp desc
| WHERE event.duration > 3000000
| WHERE message LIKE "Connection *"
""",
format="csv"
)
df = pd.DataFrame = pd.read_csv(StringIO(response.body))
print(df)
print("==================================================================")
response = es.esql.query(
query="""
FROM sample_data
| STATS avg=AVG(event.duration), count=COUNT(*) BY client.ip
| SORT count
""",
format="csv"
)
df = pd.DataFrame = pd.read_csv(
StringIO(response.body),
dtype={"count":"Int64", "avg":np.float64}
)
print(df)
print("==================================================================")在运行上面的代码之前，我们需要在 terminal 中设置相应的环境变量：
export ES_SERVER="localhost"
export ES_USER="elastic"
export ES_PASSWORD="q2rqAIphl-fx9ndQ36CO"
export ES_FINGERPRINT="bce66ed55097f255fc8e4420bdadafc8d609cc8027038c2dd09d805668f3459e"然后，我们使用如下的命令来运行：
python3 esql.py$ python3 esql.py
/Users/liuxg/python/esql/esql.py:22: ElasticsearchWarning: No limit defined, adding default limit of [500]
response = es.esql.query(query="FROM sample_data", format="csv")
@timestamp client.ip event.duration message
0 2023-10-23T12:15:03.360Z 172.21.2.162 3450233 Connected to 10.1.0.3
1 2023-10-23T12:27:28.948Z 172.21.2.113 2764889 Connected to 10.1.0.2
2 2023-10-23T13:33:34.937Z 172.21.0.5 1232382 Disconnected
3 2023-10-23T13:51:54.732Z 172.21.3.15 725448 Connection error
4 2023-10-23T13:52:55.015Z 172.21.3.15 8268153 Connection error
5 2023-10-23T13:53:55.832Z 172.21.3.15 5033755 Connection error
6 2023-10-23T13:55:01.543Z 172.21.3.15 1756467 Connected to 10.1.0.1
==================================================================
@timestamp client.ip event.duration message
0 2023-10-23T13:52:55.015Z 172.21.3.15 8268153 Connection error
==================================================================
/Users/liuxg/python/esql/esql.py:44: ElasticsearchWarning: No limit defined, adding default limit of [500]
response = es.esql.query(
avg count client.ip
0 1232382.00 1 172.21.0.5
1 3450233.00 1 172.21.2.162
2 2764889.00 1 172.21.2.113
3 3945955.75 4 172.21.3.15
==================================================================很显然，我们得到了最终的结果。
