如何在 Elasticsearch Ruby 客户端中使用 ES|QL Helper

作者：来自 Elastic Fernando Briano

了解如何使用 Elasticsearch Ruby 客户端编写 ES|QL 查询并处理其结果。

简介

Elasticsearch Ruby 客户端可用于编写 EQ|QL 查询，使处理从 esql.query 返回的数据更加容易。ES|QL 允许开发人员通过查询过滤、转换和分析存储在 Elasticsearch 中的数据。它使用 “管道”(|) 逐步处理数据。

自版本 8.11.0 中作为实验性版本推出以来，esql.query API 一直受 Elasticsearch Ruby 客户端支持。

你可以使用以下代码执行 ES|QL 请求：

client = Elasticsearch::Client.new
query = <<ESQL
        FROM sample_data
        | EVAL duration_ms = ROUND(event.duration / 1000000.0, 1)
ESQL

response = client.esql.query(body: { query: query })

默认响应是从 JSON 解析的（你也可以通过传入格式参数来获取 CSV 或文本），它看起来像这样：

puts response

{"columns"=>[
  {"name"=>"@timestamp", "type"=>"date"},
  {"name"=>"client.ip", "type"=>"ip"},
  {"name"=>"event.duration", "type"=>"long"},
  {"name"=>"message", "type"=>"keyword"},
  {"name"=>"duration_ms", "type"=>"double"}
],
"values"=>[
  ["2023-10-23T12:15:03.360Z", "172.21.2.162", 3450233, "Connected to 10.1.0.3", 3.5],
  ["2023-10-23T12:27:28.948Z", "172.21.2.113", 2764889, "Connected to 10.1.0.2", 2.8],
  ["2023-10-23T13:33:34.937Z", "172.21.0.5", 1232382, "Disconnected", 1.2],
  ["2023-10-23T13:51:54.732Z", "172.21.3.15", 725448, "Connection error", 0.7],
  ["2023-10-23T13:52:55.015Z", "172.21.3.15", 8268153, "Connection error", 8.3],
  ["2023-10-23T13:53:55.832Z", "172.21.3.15", 5033755, "Connection error", 5.0],
  ["2023-10-23T13:55:01.543Z", "172.21.3.15", 1756467, "Connected to 10.1.0.1", 1.8]
]}

ES|QL 助手

在 Elasticsearch Ruby v8.13.0 中，客户端为 esql.query API 引入了 ES|QL 助手。助手返回的不是默认响应，而是以列作为键并返回相应值的哈希数组，而不是默认的 JSON 值。

此外，你可以遍历响应值并通过传入列 => Proc 值的哈希来转换数据。例如，你可以使用它将 @timestamp 列值转换为 DateTime 对象。我们将通过示例数据了解如何使用它。

设置和提取数据

在此示例中，我们使用来自 TheGamesDB（一个社区驱动的众包游戏信息网站）的 JSON dump。下载 JSON 文件后，我们可以使用 Ruby 客户端的另一个助手 Bulk Helper 将其提取到 Elasticsearch 中。

数据中包含数据库中所有游戏的列表，位于 data.games 键中。数据还包括平台和封面信息，但在本示例中，我们只使用游戏数据。BulkHelper 提供了一种将 JSON 文件直接导入到 Elasticsearch 的方法。

要使用该助手，我们需要在代码中 require 它，并使用客户端和索引对其进行实例化，以便执行批量操作（我们可以稍后在已实例化的助手上更改索引）。我们可以使用 ingest_json 并传入 JSON 文件、它可以找到数据的键，然后切片以将文档分批分离，然后再将它们发送到 Elasticsearch：

require 'elasticsearch/helpers/bulk_helper'
file = './database-latest.json'
index = 'videogames'

bulk_helper = Elasticsearch::Helpers::BulkHelper.new(client, index)
bulk_helper.ingest_json(file, keys: ['data', 'games'], slice: 100)

这会将所有游戏标题及其各自的信息纳入视频游戏索引中。

使用 ES|QL 助手

加载数据后，我们现在可以使用 ES|QL 进行查询：

require 'elasticsearch/helpers/esql_helper'

query = <<~ESQL
  FROM videogames
  | WHERE game_title LIKE "*Turrican*"
  | LIMIT 100
  | SORT release_date
ESQL

如果我们直接使用 esql.query API 运行此查询，我们将获得列/值结果：

response = client.esql.query(body: { query: query })
response['columns']
[
  {"name"=>"alternates", "type"=>"text"},
  {"name"=>"alternates.keyword", "type"=>"keyword"},
  {"name"=>"coop", "type"=>"text"},
  {"name"=>"coop.keyword", "type"=>"keyword"},
  {"name"=>"country_id", "type"=>"long"},
  {"name"=>"developers", "type"=>"long"},
  {"name"=>"game_title", "type"=>"text"},
 ...
]

response['values']
[
  [nil, nil, "No", "No", 0, 6970, "Turrican",  ...],
  [nil, nil, "No", "No", 0, nil, "Turrican II: The Final Fight", ...]

但是，助手会返回一个哈希数组，其中列作为键，并带有相应的值。因此，我们可以使用响应，并以列的名称作为键来访问数组中每个哈希的值：

response = Elasticsearch::Helpers::ESQLHelper.query(client, query)
[
  {
    "alternates"=>nil,
    "alternates.keyword"=>nil,
    "coop"=>"No",
    "coop.keyword"=>"No",
    "country_id"=>0,
    "developers"=>6970,
    "game_title"=>"Turrican",
    ...
  },
  ...
]

response.map { |game| "#{game['game_title']} - 📅 #{game['release_date']}" }
[
  "Turrican - 📅 1990-01-01T00:00:00.000Z",
  "Turrican II: The Final Fight - 📅 1990-01-01T00:00:00.000Z",
  ...
]

ESQLHelper 还提供了转换响应中的数据的功能。我们可以通过传入 column => Proc 值的哈希来实现这一点。例如，假设我们想要格式化上一个查询中的发布日期以显示更人性化的日期。我们可以运行以下命令：

parser = proc do |t|
  DateTime.parse(t).strftime('%B %d, %Y')
end

response = Elasticsearch::Helpers::ESQLHelper.query(client, query, parser: { 'release_date' => parser })

如果我们运行之前相同的代码，我们将得到以下结果：

response.map { |game| "#{game['game_title']} - 📅 #{game['release_date']}" }
[
  "Turrican - 📅 January 01, 1990",
  "Turrican II: The Final Fight - 📅 January 01, 1990",
  ...
]

你可以传入与响应中的列数相同的 Procs。例如，数据包含一个 youtube 字段，其中有时存储 YouTube 视频的 URL，有时仅存储视频哈希（例如 U4bKxcV5hsg）。YouTube 视频的 URL 遵循惯例 https://youtube.com/watch?v=VIDEOHASH。因此，我们还可以添加一个解析器，将 URL 添加到仅包含哈希的值的前面：

parser = {
  'release_date' => proc { |t| DateTime.parse(t).strftime('%B %d, %Y') },
  'youtube' => proc do |video|
    if video =~ URI::DEFAULT_PARSER.make_regexp
      video
    elsif video.nil? || video.empty?
      nil
    else
      "https://www.youtube.com/watch?v=#{video}"
    end
  end
}
# We run the query again and pass the new parser to the helper:
response = Elasticsearch::Helpers::ESQLHelper.query(client, query, parser: parser)

如果我们随后运行 response.map { |a| a['youtube'] }.compact，我们将获得我们正在寻找的视频游戏的 YouTube 视频的 URL。