Elasticsearch：使用 Python elasticsearch-dsl-py 库对 Elasticsearch 进行查询

Posted 2021-09-16 Elastic 中国社区官方博客

在我之前的几篇文章中：

• Elasticsearch：Elasticsearch 开发入门 - Python

• Elasticsearch：运用 Python 来实现对搜索结果的分页

• Elasticsearch：从零开始安装 Elasticsearch 并使用 Python 装载一个 CSV 并读写它

我详细地描述了如何使用 elasticsearch-py 这个库来对 Elasticsearch 进行访问。 elasticsearch-py 库是一个官方发布的低级客户端库。它其实就是针对 Elasticsearch 的 REST API 接口进行了包装。

在今天的文章中，我将介绍如何使用 elasticsearch-dsl-py 这个高级库来对 Elasticsearch 进行访问。这个库是建立于 elasticsearch-py 之上的。它提供用于将文档作为 Python 对象处理的可选包装器。

有关更多关于 Elasticsearch 的客户端开发，可以参阅官方链接 https://www.elastic.co/guide/en/elasticsearch/client/index.html

在本次练习中，我将使用 Elastic Stack 7.14 来进行展示。有些版本的界面可能和这个有所不同。

安装

Elasticsearch

我们可以参阅文章 “如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch”  来安装 Elasticsearch。

Kibana

我们可以参阅文章 “Kibana：如何在 Linux，MacOS 及 Windows上安装 Elastic 栈中的 Kibana” 来安装 Kibana。

Python

我们需要安装 Python 的环境。同时为了能够访问 Elasticsearch，我们必须安装如下的库：

pip install elasticsearch
pip install elasticsearch-dsl

为了能够使得我们很方便地展示，我们也安装 ipython。当然你也可以使用 Jupyter Notebook 来进行展示。你可以参考我之前的文章 “Elasticsearch：使用 Jupyter Notebook 创建 Python 应用导入 CSV 文件”。

我们使用如下的命令来安装 ipython：

pip install ipython

准备数据

在本次练习中，我将使用 Kibana 自带的索引来进行练习。

 

 

这样我们就导入了索引 kibana_sample_data_logs。

我们们在 Kibana 的 Dev Tools 中可以使用如下的查询： 

显然这是一个比较复杂的 compound DSL 查询。如果大家对这个查询还是不很熟悉的话，请参阅我之前的文章 “开始使用 Elasticsearch （2）”。在那里，有详细的关于 DSL 查询以及 compound query 的具体描述。

上面的查询结果将返回 8 个结果的查询：

 在接下来的介绍中，我们将展示如何使用 elastiicsearch-dsl 来实现上面的查询。

elaticsearch-dsl 实践

我们在 terminal 中打入如下的命令：

ipython

 我们接着打入如下的句子：

from elasticsearch import Elasticsearch
es = Elasticsearch(hosts=['localhost:9200'])
es.info()

我们可以看到：

上面表明我们的连接到 Elasticsearch 是成功的。如果你的 Elasticsearch 是 HTTPS 的那么你需要使用如下的格式来进行连接：

from elasticsearch import Elasticsearch

es = Elasticsearch(
      hosts=['YourElasticsearchServer:9200'],
      http_auth=('YourUserName', 'YourPassword'),
      scheme="https",
      use_ssl=True,
      verify_certs=False,
      ssl_show_warn=False
  )

我们接下来 import elasticsearch-dsl 库：

from elasticsearch_dsl import Search
s = Search(index="kibana_sample_data_logs")
s = s.using(es)

我们接着打入如下的命令：

s.to_dict()

 上面的命令显示：

 显然这个查询和我们在 Kibana 中如下的搜索是一样的：

GET kibana_sample_data_logs/_search
{
  "query": {
    "match": {
      "extension": "gz"
    }
  }
}

我们接下来添加另外一个 match query：

s = s.query("match", host__keyword="artifacts.elastic.co")
s.to_dict()

请注意上面的 host__keyword。中间是两个 _。这是因为在 Python 中 . 是一个保留符号。它表示 host.keyword。经过上面的添加后，我们可以查看现在的 query：

很显然这是一个 compound query。它和如下格式的查询是一样的：

GET kibana_sample_data_logs/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "match": {
            "host.keyword": "artifacts.elastic.co"
          }
        },
        {
          "match": {
            "extension": "gz"
          }
        }
      ]
    }
  }
}

我们接着添加一个 filter 的 clause：

s=s.filter("term", geo__dest="LK")

 再次提醒请注意上面的 __ 符号。geo__dest 表示的是 geo.dest。

上面的 query 在 Kibana 中是这样的:

GET kibana_sample_data_logs/_search
{
  "query": {
    "bool": {
      "filter": [
        {
          "term": {
            "geo.dest": "LK"
          }
        }
      ],
      "must": [
        {
          "match": {
            "host.keyword": "artifacts.elastic.co"
          }
        },
        {
          "match": {
            "extension": "gz"
          }
        }
      ]
    }
  }
}

我们接下来添加 from + size 来选择我们想要的数据：

s=s[2:4]

这个类似如下的查询：

GET kibana_sample_data_logs/_search
{
  "from": 2,
  "size": 2, 
  "query": {
    "bool": {
      "filter": [
        {
          "term": {
            "geo.dest": "LK"
          }
        }
      ],
      "must": [
        {
          "match": {
            "host.keyword": "artifacts.elastic.co"
          }
        },
        {
          "match": {
            "extension": "gz"
          }
        }
      ]
    }
  }
}

最后，我们可以甚至添加 aggregation：

s.aggs.bucket("source_country","terms",field="geo__src")

在上面，我们针对 geo.src 做一个 terms 的 bucket 聚合：

上面的查询就是如下的查询：

GET kibana_sample_data_logs/_search
{
  "from": 2,
  "size": 2, 
  "query": {
    "bool": {
      "filter": [
        {
          "term": {
            "geo.dest": "LK"
          }
        }
      ],
      "must": [
        {
          "match": {
            "host.keyword": "artifacts.elastic.co"
          }
        },
        {
          "match": {
            "extension": "gz"
          }
        }
      ]
    }
  },
  "aggs": {
    "source_country": {
      "terms": {
        "field": "geo.src"
      }
    }
  }
}

我们可以使用如下的方法来打印查询的结果：

for hit in s:
   print (hit.clientip)

在上面，我们打印出来 clientip 的地址：

好了，今天的展示就到这里。在某种程度上，elasticsearch-dsl 库给我们开发者带来更多的方便及便捷。

更多关于 elasticsearch-dsl 的资料，请参考官方文档：https://elasticsearch-dsl.readthedocs.io/en/latest/ 及源码 https://github.com/elastic/elasticsearch-dsl-py