› › ›

API создания потоков данных

Создаёт поток данных.

Запрос

PUT _ml/datafeeds/<feed_id>

Предварительные условия

Прежде чем создавать поток данных, необходимо создать задание обнаружения аномалий.
Требуются следующие разрешения:
- cluster: manage_ml (роль machine_learning_admin предоставляет это разрешение)
- источник индекса, настроенный в потоке данных: read

Описание

Потоки данных извлекают данные из Elasticsearch для анализа заданием обнаружения аномалий. Одному заданию обнаружения аномалий можно назначить только один поток данных.

Поток данных содержит запрос, выполняемый через определённый интервал (frequency). Если вас беспокоит задержка данных, вы можете добавить задержку (query_delay) на каждом интервале. См. Обработка задержки данных.

Для создания потока данных необходимо использовать Kibana, этот API или API создания заданий обнаружения аномалий. Не добавляйте поток данных напрямую в индекс .ml-config с помощью API индексов Elasticsearch. Если включены функции безопасности Elasticsearch, не предоставляйте пользователям права доступа write к индексу .ml-config.
Когда включены функции безопасности Elasticsearch, поток данных запоминает роли пользователя, создавшего его, в момент создания, и выполняет запрос с использованием тех же ролей. Если вы предоставляете заголовки вторичной авторизации, используются эти данные авторизации.

Параметры пути

<feed_id>: (Обязательный, строка) Числовая строка, уникально идентифицирующая поток данных. Этот идентификатор может содержать строчные буквенно-цифровые символы (a-z и 0-9), дефисы и подчёркивания. Он должен начинаться и заканчиваться буквенно-цифровыми символами.

Параметры запроса

allow_no_indices

(Необязательный, булево) Если true, выражения с подстановкой индексов, которые не разрешаются в конкретные индексы, игнорируются. Включая строку _all или когда индексы не указаны. По умолчанию true.

expand_wildcards

(Необязательный, строка) Тип индекса, с которым могут совпадать шаблоны с подстановкой. Если запрос может быть направлен на потоки данных, этот параметр определяет, будут ли шаблоны с подстановкой соответствовать скрытым потокам данных. Поддерживает значения через запятую, такие как open,hidden. Допустимые значения:

all: Соответствует любому потоку данных или индексу, включая скрытые.
open: Соответствует открытым индексам, не являющимся скрытыми. Также соответствует любым нескрытым потокам данных.
closed: Соответствует закрытым индексам, не являющимся скрытыми. Также соответствует любым нескрытым потокам данных. Потоки данных закрыть нельзя.
hidden: Соответствует скрытым потокам данных и скрытым индексам. Должно использоваться совместно с open, closed или с обоими.
none: Шаблоны с подстановкой не принимаются.

По умолчанию open.

ignore_throttled

(Необязательный, булево) Если true, конкретные, расширенные или алиасированные индексы игнорируются при замораживании. По умолчанию true.

[7.16.0] Устарело в версии 7.16.0.

ignore_unavailable

(Необязательный, булево) Если true, недоступные индексы (отсутствующие или закрытые) игнорируются. По умолчанию false.

Тело запроса

aggregations

(Необязательно, объект) Если задано, поток данных выполняет агрегационные запросы. Поддержка агрегаций ограничена и должна использоваться только с данными низкой кардинальности. Дополнительную информацию см. в статье Агрегирование данных для повышения производительности.

chunking_config

(Необязательно, объект) Потоки данных могут потребоваться для поиска за длительные периоды времени, в течение нескольких месяцев или лет. Этот поиск разбивается на временные сегменты, чтобы обеспечить контроль нагрузки на Elasticsearch. Конфигурация сегментирования управляет тем, как рассчитывается размер этих временных сегментов, и является расширенным параметром конфигурации.

Свойства chunking_config

mode

(строка) Доступно три режима:

auto: Размер сегмента динамически вычисляется. Это значение по умолчанию и рекомендуется, когда поток данных не использует агрегации.
manual: Сегментирование применяется в соответствии с указанным time_span. Используйте этот режим, когда поток данных использует агрегации.
off: Сегментирование не применяется.

time_span

(единицы измерения времени) Временной интервал, который будет опрашиваться в каждом запросе. Этот параметр применим только в режиме manual. Например: 3h.

delayed_data_check_config

(Необязательно, объект) Указывает, проверяет ли поток данных наличие пропущенных данных и размер окна. Например: {"enabled": true, "check_window": "1h"}.

Поток данных может (по выбору) искать в индексах, которые уже были прочитаны, чтобы определить, были ли впоследствии добавлены данные в индекс. Если пропущенные данные найдены, это хороший признак того, что параметр query_delay установлен слишком низко, и данные индексируются после того, как поток данных пропустил этот момент во времени. См. Работа с отстающими данными.

Эта проверка выполняется только для потоков данных в реальном времени.

Свойства delayed_data_check_config

check_window: (единицы измерения времени) Окно времени, в котором ищутся отстающие данные. Это окно заканчивается последним завершённым ведром. По умолчанию значение null, что приводит к вычислению соответствующего check_window при запуске потока данных в реальном времени. В частности, расчёт значения check_window по умолчанию основан на максимуме значений 2h или 8 * bucket_span.
enabled: (Булево) Указывает, периодически ли поток данных проверяет отстающие данные. По умолчанию true.

frequency

(Необязательно, единицы измерения времени) Интервал, с которым запланированные запросы выполняются во время работы потока данных в реальном времени. Значение по умолчанию — это интервал ведра для коротких ведёр или разумная часть интервала ведра для длинных ведёр. Например: 150s. Когда frequency короче, чем интервал ведра, промежуточные результаты для последнего (частичного) ведра записываются, а затем в конечном итоге перезаписываются полными результатами ведра. Если поток данных использует агрегации, это значение должно быть кратно интервалу агрегации гистограммы дат.

indices

(Обязательно, массив) Массив имён индексов. Поддерживаются подстановочные знаки. Например: ["it_ops_metrics", "server*"].

Если какие-либо индексы находятся в удалённых кластерах, то узлы-мастера и узлы машинного обучения должны иметь роль remote_cluster_client.

indices_options

(Необязательно, объект) Указывает параметры расширения индекса, используемые во время поиска.

Например:

{
   "expand_wildcards": ["all"],
   "ignore_unavailable": true,
   "allow_no_indices": "false",
   "ignore_throttled": true
}

Дополнительную информацию об этих параметрах см. в разделе Синтаксис для работы с несколькими целевыми объектами.

job_id

(Обязательно, строка) Идентификатор задачи обнаружения аномалий.

max_empty_searches

(Необязательно, целое число) Если поток данных в реальном времени никогда не видел данных (включая период начальной подготовки), он автоматически остановится и закроет связанную задачу после заданного количества запросов в реальном времени, не вернувших документов. Другими словами, он остановится после frequency раз max_empty_searches работы в режиме реального времени. Если параметр не задан, то поток данных без конечной даты, не получивший данных, будет оставаться запущенным до явного останова. По умолчанию этот параметр не задан.

query

(Необязательно, объект) Язык запросов Elasticsearch (DSL). Это значение соответствует объекту запроса в теле POST-запроса Elasticsearch. Можно использовать все параметры, поддерживаемые Elasticsearch, так как этот объект передаётся в Elasticsearch в неизменённом виде. По умолчанию этот параметр имеет значение: {"match_all": {"boost": 1}}.

query_delay

(Необязательно, единицы измерения времени) Количество секунд, отстающих от реального времени, для запроса данных. Например, если данные с 10:04 утра могут быть доступны для поиска в Elasticsearch только в 10:06 утра, установите этот параметр в 120 секунд. Значение по умолчанию выбирается случайным образом между 60s и 120s. Эта случайность улучшает производительность запросов при наличии нескольких задач, выполняющихся на одном узле. Дополнительную информацию см. в статье Обработка отстающих данных.

runtime_mappings

(Необязательно, объект) Указывает поля runtime для поиска потока данных.

Например:

{
  "day_of_week": {
    "type": "keyword",
    "script": {
      "source": "emit(doc['@timestamp'].value.dayOfWeekEnum.getDisplayName(TextStyle.FULL, Locale.ENGLISH))"
    }
  }
}

script_fields

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

scroll_size

(Необязательно, целое без знака) Параметр size, используемый в запросах Elasticsearch, когда поток данных не использует агрегации. Значение по умолчанию — 1000. Максимальное значение — значение index.max_result_window, по умолчанию равное 10 000.

Примеры

Создайте поток данных для задачи обнаружения аномалий (test-job):

resp = client.ml.put_datafeed(
    datafeed_id="datafeed-test-job",
    pretty=True,
    indices=[
        "kibana_sample_data_logs"
    ],
    query={
        "bool": {
            "must": [
                {
                    "match_all": {}
                }
            ]
        }
    },
    job_id="test-job",
)
print(resp)

const response = await client.ml.putDatafeed({
  datafeed_id: "datafeed-test-job",
  pretty: "true",
  indices: ["kibana_sample_data_logs"],
  query: {
    bool: {
      must: [
        {
          match_all: {},
        },
      ],
    },
  },
  job_id: "test-job",
});
console.log(response);

PUT _ml/datafeeds/datafeed-test-job?pretty
{
  "indices": [
    "kibana_sample_data_logs"
  ],
  "query": {
    "bool": {
      "must": [
        {
          "match_all": {}
        }
      ]
    }
  },
  "job_id": "test-job"
}

При создании потока данных вы получите следующие результаты:

{
  "datafeed_id" : "datafeed-test-job",
  "job_id" : "test-job",
  "authorization" : {
    "roles" : [
      "superuser"
    ]
  },
  "query_delay" : "91820ms",
  "chunking_config" : {
    "mode" : "auto"
  },
  "indices_options" : {
    "expand_wildcards" : [
      "open"
    ],
    "ignore_unavailable" : false,
    "allow_no_indices" : true,
    "ignore_throttled" : true
  },
  "query" : {
    "bool" : {
      "must" : [
        {
          "match_all" : { }
        }
      ]
    }
  },
  "indices" : [
    "kibana_sample_data_logs"
  ],
  "scroll_size" : 1000,
  "delayed_data_check_config" : {
    "enabled" : true
  }
}

© 2023-2025 Elasticsearch
As of September 2024, Elasticsearch is available under a choice of three licenses: the Server Side Public License (SSPL), the Elastic License, or the AGPLv3 (OSI approved).
Elasticsearch and the Elasticsearch logo are trademarks of Elasticsearch B.V., registered in the U.S. and in other countries.
https://www.elastic.co/guide/en/elasticsearch/reference/8.17/ml-put-datafeed.html