大型数据集上的 ElasticSearch 聚合

Question

我使用 Python ElasticSearch API。 我的数据集太大而无法使用 search() 检索。 我可以用 helpers.scan() 检索它，但数据太大，无法用 pandas 快速处理。

所以我学会了如何使用 ElasticSearch 进行聚合以压缩数据，但仍然使用 search() 我无法检索所有数据。我知道聚合是在“通常”的搜索大小上完成的，即使聚合会给出一行？

最后我尝试了聚合 + 扫描或滚动，但我知道 scan() 或滚动() 不能用于聚合，因为这些请求适用于数据集的子集，然后聚合在子集上是无意义的。

在非常大的数据集上进行聚合的好方法是什么？ 我在网上找不到任何相关的解决方案。

更明确地说，我的情况是： 我有 X 数千个移动传感器每小时传输最后一个停止位置，即新的停止位置。从最后一站到新站的移动可能需要几天时间，所以在几天内我没有每小时采集的相关信息。 作为 ElasticSearch 搜索输出，我只需要格式的每个唯一行： sensor_id / last_stop / new_stop

Answer 1

如果您在 pandas 中使用 elastic，您可以尝试 eland 一个新的官方弹性库，该库旨在更好地集成它们。试试：

es = Elasticsearch() 

body = {
  "size": 0,
  "aggs": {
    "getAllSensorId": {
      "terms": {
        "field": "sensor_id",
        "size": 10000
      },
      "aggs": {
        "getAllTheLastStop": {
          "terms": {
            "field": "last_stop",
            "size": 10000
          },
      "aggs": {
        "getAllTheNewStop": {
          "terms": {
            "field": "new_stop",
            "size": 10000
          }
        }
      }
        }
      }
    }
  }
}
list_of_results = []
result = es.search(index="my_index", body=body)
for sensor in result["aggregations"]["getAllTheSensorId"]["buckets"]:
    for last in sensor["getAllTheLastStop"]["buckets"]:
        for new in last["getAllTheNewStop"]["buckets"]:
            record = {"sensor": sensor['key'], "last_stop": last['key'], "new_stop": new['key']}
            list_of_results.append(record)