Начало работы со сводками

Для использования функции сводок необходимо создать одно или несколько «заданий сводки». Эти задания выполняются непрерывно в фоновом режиме и сводят указанный индекс или индексы, помещая свёрнутые документы во вторичный индекс (также по вашему выбору).

Представьте, что у вас есть серия ежедневных индексов, содержащих данные датчиков (sensor-2017-01-01, sensor-2017-01-02 и т.д.). Пример документа может выглядеть так:

{
  "timestamp": 1516729294000,
  "temperature": 200,
  "voltage": 5.2,
  "node": "a"
}

Мы хотели бы свести эти документы в сводки по часам, что позволит нам генерировать отчеты и панели мониторинга с любым интервалом времени в час или более. Задание сводки может выглядеть так:

resp = client.rollup.put_job(
    id="sensor",
    index_pattern="sensor-*",
    rollup_index="sensor_rollup",
    cron="*/30 * * * * ?",
    page_size=1000,
    groups={
        "date_histogram": {
            "field": "timestamp",
            "fixed_interval": "60m"
        },
        "terms": {
            "fields": [
                "node"
            ]
        }
    },
    metrics=[
        {
            "field": "temperature",
            "metrics": [
                "min",
                "max",
                "sum"
            ]
        },
        {
            "field": "voltage",
            "metrics": [
                "avg"
            ]
        }
    ],
)
print(resp)

const response = await client.rollup.putJob({
  id: "sensor",
  index_pattern: "sensor-*",
  rollup_index: "sensor_rollup",
  cron: "*/30 * * * * ?",
  page_size: 1000,
  groups: {
    date_histogram: {
      field: "timestamp",
      fixed_interval: "60m",
    },
    terms: {
      fields: ["node"],
    },
  },
  metrics: [
    {
      field: "temperature",
      metrics: ["min", "max", "sum"],
    },
    {
      field: "voltage",
      metrics: ["avg"],
    },
  ],
});
console.log(response);

PUT _rollup/job/sensor
{
  "index_pattern": "sensor-*",
  "rollup_index": "sensor_rollup",
  "cron": "*/30 * * * * ?",
  "page_size": 1000,
  "groups": {
    "date_histogram": {
      "field": "timestamp",
      "fixed_interval": "60m"
    },
    "terms": {
      "fields": [ "node" ]
    }
  },
  "metrics": [
    {
      "field": "temperature",
      "metrics": [ "min", "max", "sum" ]
    },
    {
      "field": "voltage",
      "metrics": [ "avg" ]
    }
  ]
}

Мы даём заданию идентификатор «sensor» (в URL: PUT _rollup/job/sensor) и говорим ему свести шаблон индекса "sensor-*". Это задание найдёт и свведёт любой индекс, соответствующий этому шаблону. Сводки затем хранятся в индексе "sensor_rollup".

Параметр cron управляет временем и частотой активации задания. Когда планировщик задания сводки срабатывает, он начинает сводить данные с места, где он остановился после последней активации. Таким образом, если вы настроите планировщик на выполнение каждые 30 секунд, задание будет обрабатывать последние 30 секунд данных, которые были индексированы в индексы sensor-*.

Если планировщик был настроен на выполнение один раз в день в полночь, задание обработает последние 24 часа данных. Выбор в значительной степени зависит от того, насколько «реального времени» вы хотите сводки и хотите ли вы обрабатывать данные непрерывно или перенести это на непиковые часы.

Далее, мы определяем набор groups. По сути, мы определяем измерения, на которых мы хотим фокусироваться в будущем при запросе данных. Группировка в этом задании позволяет использовать агрегации date_histogram по полю timestamp, свёрнутые с часовыми интервалами. Это также позволяет выполнять агрегации `terms` по полю node.

После определения групп, которые должны быть сгенерированы для данных, вы настраиваете метрики, которые должны быть собраны. По умолчанию собираются только doc_counts для каждой группы. Чтобы сводка была полезной, вы часто добавляете метрики, такие как средние значения, минимальные и максимальные значения и т. д. В этом примере метрики довольно простые: мы хотим сохранить мин/макс/сумму поля temperature и среднее значение поля voltage.

Более подробную информацию о синтаксисе задания см. в Создание задач сводки.

После выполнения вышеприведённой команды и создания задания вы получите следующий ответ:

{
  "acknowledged": true
}

После создания задание будет находиться в неактивном состоянии. Задания нужно запускать, прежде чем они начнут обрабатывать данные (это позволяет остановить их позже, как способ временной паузы без удаления конфигурации).

Для запуска задания выполните эту команду:

resp = client.rollup.start_job(
    id="sensor",
)
print(resp)

response = client.rollup.start_job(
  id: 'sensor'
)
puts response

const response = await client.rollup.startJob({
  id: "sensor",
});
console.log(response);

POST _rollup/job/sensor/_start

После того как задание выполнилось и обработало некоторые данные, мы можем использовать конечную точку поиска сводки для поиска. Функция сводки разработана таким образом, что вы можете использовать знакомый синтаксис Query DSL… он просто работает с свёрнутыми данными вместо исходных данных.

Например, рассмотрим этот запрос:

resp = client.rollup.rollup_search(
    index="sensor_rollup",
    size=0,
    aggregations={
        "max_temperature": {
            "max": {
                "field": "temperature"
            }
        }
    },
)
print(resp)

response = client.rollup.rollup_search(
  index: 'sensor_rollup',
  body: {
    size: 0,
    aggregations: {
      max_temperature: {
        max: {
          field: 'temperature'
        }
      }
    }
  }
)
puts response

const response = await client.rollup.rollupSearch({
  index: "sensor_rollup",
  size: 0,
  aggregations: {
    max_temperature: {
      max: {
        field: "temperature",
      },
    },
  },
});
console.log(response);

GET /sensor_rollup/_rollup_search
{
  "size": 0,
  "aggregations": {
    "max_temperature": {
      "max": {
        "field": "temperature"
      }
    }
  }
}

Это простая агрегация, которая вычисляет максимальное значение поля temperature. Но обратите внимание, что она отправляется в индекс sensor_rollup вместо исходных индексов sensor-*. И вы также заметите, что используется конечная точка _rollup_search. В остальном синтаксис точно такой же, как и ожидалось.

Если вы выполните этот запрос, вы получите результат, похожий на обычный ответ агрегации:

{
  "took" : 102,
  "timed_out" : false,
  "terminated_early" : false,
  "_shards" : ... ,
  "hits" : {
    "total" : {
        "value": 0,
        "relation": "eq"
    },
    "max_score" : 0.0,
    "hits" : [ ]
  },
  "aggregations" : {
    "max_temperature" : {
      "value" : 202.0
    }
  }
}

Единственное заметное отличие заключается в том, что результаты поиска сводки имеют нулевые hits, так как мы больше не ищем исходные, текущие данные. В остальном синтаксис идентичен.

Отсюда можно сделать несколько интересных выводов. Во-первых, даже несмотря на то, что данные были свёрнуты с часовыми интервалами и разбиты по имени узла, запрос, который мы выполнили, просто вычисляет максимальную температуру по всем документам. groups, которые были настроены в задании, не являются обязательными элементами запроса, а являются дополнительными измерениями, по которым вы можете разбивать данные. Во-вторых, синтаксис запроса и ответа почти идентичен обычному синтаксису DSL, что упрощает интеграцию в панели мониторинга и приложения.

Наконец, мы можем использовать эти поля группировки, которые мы определили, чтобы сконструировать более сложный запрос:

resp = client.rollup.rollup_search(
    index="sensor_rollup",
    size=0,
    aggregations={
        "timeline": {
            "date_histogram": {
                "field": "timestamp",
                "fixed_interval": "7d"
            },
            "aggs": {
                "nodes": {
                    "terms": {
                        "field": "node"
                    },
                    "aggs": {
                        "max_temperature": {
                            "max": {
                                "field": "temperature"
                            }
                        },
                        "avg_voltage": {
                            "avg": {
                                "field": "voltage"
                            }
                        }
                    }
                }
            }
        }
    },
)
print(resp)

response = client.rollup.rollup_search(
  index: 'sensor_rollup',
  body: {
    size: 0,
    aggregations: {
      timeline: {
        date_histogram: {
          field: 'timestamp',
          fixed_interval: '7d'
        },
        aggregations: {
          nodes: {
            terms: {
              field: 'node'
            },
            aggregations: {
              max_temperature: {
                max: {
                  field: 'temperature'
                }
              },
              avg_voltage: {
                avg: {
                  field: 'voltage'
                }
              }
            }
          }
        }
      }
    }
  }
)
puts response

const response = await client.rollup.rollupSearch({
  index: "sensor_rollup",
  size: 0,
  aggregations: {
    timeline: {
      date_histogram: {
        field: "timestamp",
        fixed_interval: "7d",
      },
      aggs: {
        nodes: {
          terms: {
            field: "node",
          },
          aggs: {
            max_temperature: {
              max: {
                field: "temperature",
              },
            },
            avg_voltage: {
              avg: {
                field: "voltage",
              },
            },
          },
        },
      },
    },
  },
});
console.log(response);

GET /sensor_rollup/_rollup_search
{
  "size": 0,
  "aggregations": {
    "timeline": {
      "date_histogram": {
        "field": "timestamp",
        "fixed_interval": "7d"
      },
      "aggs": {
        "nodes": {
          "terms": {
            "field": "node"
          },
          "aggs": {
            "max_temperature": {
              "max": {
                "field": "temperature"
              }
            },
            "avg_voltage": {
              "avg": {
                "field": "voltage"
              }
            }
          }
        }
      }
    }
  }
}

Что возвращает соответствующий ответ:

{
   "took" : 93,
   "timed_out" : false,
   "terminated_early" : false,
   "_shards" : ... ,
   "hits" : {
     "total" : {
        "value": 0,
        "relation": "eq"
     },
     "max_score" : 0.0,
     "hits" : [ ]
   },
   "aggregations" : {
     "timeline" : {
       "buckets" : [
         {
           "key_as_string" : "2018-01-18T00:00:00.000Z",
           "key" : 1516233600000,
           "doc_count" : 6,
           "nodes" : {
             "doc_count_error_upper_bound" : 0,
             "sum_other_doc_count" : 0,
             "buckets" : [
               {
                 "key" : "a",
                 "doc_count" : 2,
                 "max_temperature" : {
                   "value" : 202.0
                 },
                 "avg_voltage" : {
                   "value" : 5.1499998569488525
                 }
               },
               {
                 "key" : "b",
                 "doc_count" : 2,
                 "max_temperature" : {
                   "value" : 201.0
                 },
                 "avg_voltage" : {
                   "value" : 5.700000047683716
                 }
               },
               {
                 "key" : "c",
                 "doc_count" : 2,
                 "max_temperature" : {
                   "value" : 202.0
                 },
                 "avg_voltage" : {
                   "value" : 4.099999904632568
                 }
               }
             ]
           }
         }
       ]
     }
   }
}

Помимо того, что он более сложный (гистограмма дат и агрегация `terms` плюс дополнительная метрика среднего значения), вы заметите, что гистограмма дат использует интервал 7d вместо 60m.

Этот быстрый старт должен был дать краткое описание основных функций, которые предоставляет Rollup. Больше советов и моментов, которые следует учитывать при настройке сводок, вы найдёте в остальной части этого раздела. Вы также можете изучить API REST для обзора доступных возможностей.