Как работают контрольные точки преобразования

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

Если ваше преобразование выполняется только один раз, то логически существует только одна контрольная точка. Однако, если преобразование выполняется непрерывно, оно создаёт контрольные точки по мере поглощения и преобразования новых исходных данных. Свойство sync конфигурации преобразования настраивает создание контрольных точек, определяя поле времени.

Для создания контрольной точки непрерывное преобразование:

Этот процесс создания контрольных точек включает как операции поиска, так и индексирования в кластере. Мы стремились к контролю производительности при разработке преобразований. Мы посчитали предпочтительным, чтобы преобразование выполнялось дольше, вместо того, чтобы завершиться быстро и занимать приоритет в потреблении ресурсов. Тем не менее, кластеру всё равно необходимы достаточные ресурсы для поддержки как составного агрегационного поиска, так и индексирования его результатов.

В большинстве случаев настоятельно рекомендуется использовать метку времени поглощения исходных индексов для синхронизации преобразования. Это наиболее оптимальный способ для преобразований идентифицировать новые изменения. Если ваш источник данных соответствует стандарту ECS, у вас может быть поле event.ingested. В этом случае используйте event.ingested в качестве свойства sync.time.field вашего преобразования.

Если у вас нет поля event.ingested или оно не заполнено, вы можете установить его, используя конвейер поглощения. Создайте конвейер поглощения, используя API конвейера поглощения (например, как показано ниже) или через Kibana в разделе Управление стеком > Конвейеры поглощения. Используйте set процессор для установки поля и свяжите его со значением метки времени поглощения.

resp = client.ingest.put_pipeline(
    id="set_ingest_time",
    description="Set ingest timestamp.",
    processors=[
        {
            "set": {
                "field": "event.ingested",
                "value": "{{{_ingest.timestamp}}}"
            }
        }
    ],
)
print(resp)

response = client.ingest.put_pipeline(
  id: 'set_ingest_time',
  body: {
    description: 'Set ingest timestamp.',
    processors: [
      {
        set: {
          field: 'event.ingested',
          value: '{{{_ingest.timestamp}}}'
        }
      }
    ]
  }
)
puts response

const response = await client.ingest.putPipeline({
  id: "set_ingest_time",
  description: "Set ingest timestamp.",
  processors: [
    {
      set: {
        field: "event.ingested",
        value: "{{{_ingest.timestamp}}}",
      },
    },
  ],
});
console.log(response);

PUT _ingest/pipeline/set_ingest_time
{
  "description": "Set ingest timestamp.",
  "processors": [
    {
      "set": {
        "field": "event.ingested",
        "value": "{{{_ingest.timestamp}}}"
      }
    }
  ]
}

После создания конвейера поглощения примените его к исходным индексам вашего преобразования. Конвейер добавляет поле event.ingested ко всем документам со значением метки времени поглощения. Настройте свойство sync.time.field вашего преобразования для использования этого поля, используя API создания преобразования для новых преобразований или API обновления преобразования для существующих преобразований. Поле event.ingested используется для синхронизации преобразования.

Дополнительную информацию о использовании конвейера поглощения см. в разделах Добавление конвейера к запросу индексирования и Конвейеры поглощения.

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

В данном примере данные сгруппированы по именам хостов. Обнаружение изменений определяет, какие имена хостов изменились, например, хосты A, C и G, и обновляет только документы с этими хостами, но не обновляет документы, хранящие информацию о хостах B, D или любом другом неизменённом хосте.

Другая эвристика может применяться для временных корзин, когда используется date_histogram для группировки по временным корзинам. Обнаружение изменений определяет, какие временные корзины изменились, и обновляет только их.

Ошибки в преобразованиях, как правило, связаны с поиском или индексированием. Для повышения устойчивости преобразований позиции курсоров составного поиска и поиска изменённых сущностей отслеживаются в памяти и периодически сохраняются.

Ошибки контрольных точек можно разделить на следующие категории:

Если узел, выполняющий преобразования, выходит из строя, преобразование перезапускается с последней сохранённой позиции курсора. Этот процесс восстановления может повторить некоторые операции, которые преобразование уже выполнило, но он гарантирует согласованность данных.