Ретроспективная нормализация событий

Данная инструкция описывает процесс настройки ретроспективной нормализации событий в системе.

О ретроспективной нормализации событий

Ретроспективная нормализация событий — это процесс преобразования исходных событий в унифицированный формат после их сохранения в системе. Данный процесс позволяет привести исторические данные к единому виду, удобному для анализа.

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

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

Настройка ретроспективной нормализации поддерживается только в том случае, если СУБД ClickHouse установлена в выделенном кластере.

Подготовка к настройке

Для ретроспективной нормализации и корреляции необходим драйвер БД ClickHouse версии 0.9.8 или выше. Чтобы его настроить:

Если драйвер БД ClickHouse отсутствует в системе, добавьте его. Для этого:
1. Скачайте файл драйвера clickhouse-jdbc-<version>-all-dependencies.jar из репозитория ClickHouse, где <version> — версия драйвера, не ниже 0.9.8.
  
  Например, драйвер для версии 0.9.8 можно скачать по следующей ссылке.
2. В разделе Ресурсы → Драйверы БД добавьте скачанный драйвер в систему.
В разделе Ресурсы → Секреты создайте секрет со строкой подключения к вашей БД ClickHouse:
```
jdbc:clickhouse://<address>:<port>/default?user=reader&password=<password>&compress=0
```
Здесь:
- <address> — полное доменное имя (FQDN) или IP-адрес сервера, где развернут ClickHouse. Если вы указываете FQDN, убедитесь, что оно корректно разрешается в IP-адрес на машине, которая выполняет роль worker-узла в кластере Kubernetes, где развернута система.
- <port> — HTTP-порт, который вы задавали при развертывании и настройке ClickHouse.
- <password> — пароль от учетной записи reader, который вы задавали при развертывании и настройке ClickHouse.

Настройка ретроспективной нормализации

Для настройки ретроспективной нормализации событий выполните следующие шаги:

При необходимости создайте новое хранилище событий, куда будут записываться события после ретроспективной нормализации.
Перейдите в раздел Ресурсы → Коллекторы.
Создайте новый коллектор для выполнения задачи ретроспективной нормализации и включите его.
Добавьте новый конвейер в коллекторе.
Откройте конфигурацию конвейера.
Добавьте на конвейер элемент Точка входа со следующими параметрами:
- Название: введите название точки входа.
- Тип точки входа: выберите вариант Database.
- Драйвер базы данных: выберите необходимый драйвер БД ClickHouse.
- Адрес подключения: выберите созданную ранее строку подключения к БД ClickHouse.
- SQL-запрос: введите SQL-запрос для выборки событий из хранилища.
  Пример запроса:
  
  SELECT id, tenantId, timestamp, raw, collectorId, type FROM EventStorage_<ID>_local WHERE timestamp > toDateTime64(toUInt32(?) + 1, 3) ORDER BY timestamp ASC LIMIT 10000
  
  Здесь:
  
  <ID> — идентификатор хранилища, сырые события из которого необходимо отправить на повторную нормализацию.
  Как получить идентификатор хранилища
  Через карточку хранилища:
  
  Перейдите в раздел Ресурсы → Хранилища событий.
  
  Выберите требуемое хранилище в списке. В правой части экрана отобразится карточка с подробной информацией.
  
  Скопируйте идентификатор хранилища.
  
  Замените в идентификаторе все символы - на _.
  
  Через диспетчер запросов:
  
  Перейдите в раздел Поиск.
  
  Выполните поиск событий в требуемом хранилище.
  
  В разделе Инструменты → Диспетчер запросов перейдите на вкладку Журнал запросов.
  
  Выберите выполненный запрос в списке. В правой части экрана отобразится карточка с подробной информацией.
  
  В поле Запрос SQL скопируйте идентификатор хранилища, указанный в операторе FROM.
- Поле идентификатора: укажите ключ timestamp и задайте начальное значение времени в формате Unix. Это значение определяет момент времени, начиная с которого были собраны сырые события, требующие повторной нормализации.
  
  Для перевода даты и времени в формат Unix можно воспользоваться онлайн-конвертером.
  Пример настройки временного диапазона
  Предположим, что на повторную нормализацию отправляются события, собранные с 23 декабря 2025 года, что соответствует значению 1766498149 в формате Unix, и до текущего момента.
  
  Для этого в параметре Поле идентификатора необходимо добавить ключ timestamp со значением 1766498149.
  
  SQL-запрос будет выглядеть следующим образом:
  
  SELECT id, tenantId, timestamp, raw, collectorId, type FROM EventStorage_<ID>_local WHERE timestamp > toDateTime64(toUInt32(?) + 1, 3) ORDER BY timestamp ASC LIMIT 10000
  
  Здесь:
  
  <ID> — идентификатор хранилища, сырые события из которого необходимо отправить на повторную нормализацию.
  
  Верхняя граница временного диапазона в запросе может быть не задана, как в примере. В этом случае точка входа вычитывает события до текущего момента времени. Для ограничения выборки событий добавьте в оператор WHERE фильтр с указанием верхней границы в формате Unix, например:
  
  AND timestamp < toDateTime64(1766692377, 3)
  
  Дополнительно в операторе WHERE можно настроить фильтрацию по следующим полям: id, tenantId, timestamp, raw, collectorId, type.
- Интервал запроса: укажите частоту получения событий из базы данных ClickHouse. Рекомендуемое значение — 5 секунд или более.
- Максимальное количество строк: укажите значение, соответствующее параметру LIMIT в SQL-запросе.
Добавьте на конвейер элемент VRL-трансформация со следующим кодом:
```
. = parse_json(to_string(.raw) ?? .raw) ?? .raw
.raw = parse_json(to_string(.raw) ?? .raw) ?? .raw
.id = uuid_v4()
```
Представленный код позволяет привести поля событий из ClickHouse к формату, соответствующему исходным данным от источника.
Соедините добавленную точку входа и VRL-трансформацию.

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

Рекомендации по настройке конвейеров обработки событий для различных источников приведены в руководстве по настройке источников на справочном портале R-Vision.

При необходимости добавьте перед конечной точкой элемент Фильтр со следующим кодом:
```
.type == 1
```
Данный фильтр позволяет исключить из обработки события, которые не прошли нормализацию. Это позволяет предотвратить сохранение в хранилище событий, оставшихся в сыром виде.
Добавьте на конвейер элемент Конечная точка типа Хранилище событий. Укажите в настройках элемента хранилище, в которое будут записываться события после ретроспективной нормализации.
Установите конфигурацию конвейера.

Если при работе конвейера возникнут ошибки, обратитесь к разделу Устранение проблем при ретроспективной обработке событий.

Была ли полезна эта страница?