Выполнение правил трансформации события#

Описание#

Правила трансформации событий описываются в файле *.tr. По умолчанию обработчик не производит никаких действий над потоком, все необходимые действия должны указываться явно.

При выполнении правил трансформации события используются идентификаторы для входа и выхода. Описываются зарезервированными словами:

  1. INPUT, Input, input — для адресации элемента из входящего сообщения;

  2. OUTPUT, Output, output — для адресации элемента из выходящего сообщения.

Или

  1. IN, In, in — для адресации элемента из входящего сообщения;

  2. OUT, Out, out — для адресации элемента из выходящего сообщения.

Вход и выход имеют одинаковую структуру:

  1. headers — заголовки сообщения;

  2. key — ключ сообщения Kafka;

  3. body — тело сообщения;

  4. sourceTopic — топик источника;

  5. targetTopic — топик назначения;

  6. branch — имя ветки для маршрутизации.

Так же в этих корнях могут быть массивы элементов в случае нескольких сообщений на вход или выход.

При использовании INPUT и OUTPUT после (возможно, после индекса) обязательно должен быть указан тип узла:

  • headers;

  • key;

  • body;

  • sourceTopic;

  • targetTopic;

  • branch.

Примеры использования#

  1. Пример правил трансформации:

out = in
out.branch = "second"

headers[>].TargetTopic = "output"
out[>].first = in
headers[+>].BranchName = "transformSecond"
out[+>].second = in
headers[+>].BranchName = "transformThird"
out[+>].third = in

  1. Пример правил трансформации с указанием типа узла:

output.body = in
output.body.branch = "second"

output[>].targetTopic = "output"
output[>].body.first = in
output[+>].branch = "transformSecond"
output[>].body.second = in
output[+>].branch = "transformThird"
output[>].body.third = in

OUTPUT.body.name.var = INPUT.body.text

Использование правил трансформации при обратной совместимости#

Для обратной совместимости применимы старые методы обращения к входам и выходам.

При трансформации события происходит выполнение DSL правил трансформации:

  1. Вход IN содержит входящее событие;

  2. Environment.SourceTopic содержит имя топика, откуда было прочитано сообщение;

  3. Environment.SourceKey содержит значение ключа из сообщения Apache Kafka в виде строки;

  4. Environment.kafka.headers содержит словарь заголовков из сообщения Apache Kafka;

  5. Environment.cache содержит элементы, сохраняющиеся между вызовами правил трансформации;

  6. Environment.loadedVariables содержит элементы, загруженные из JSON файла из параметра environmentVariables шага трансформации событий.

Выход OUT должен содержать исходящее событие или массив исходящих событий. Выход HEADERS может содержать следующие поля:

  1. TargetTopic имя топика назначения события;

  2. TargetKey значение ключа в сообщении Apache Kafka;

  3. kafka словарь заголовков сообщения Apache Kafka, не поддерживает вложенные словари;

  4. BranchName имя следующего шага из побочной ветки потока.

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

При работе с INPUT и OUTPUT вместо выхода HEADERS нужно использовать соответствующие значения узлов INPUT и OUTPUT.

Поддерживается вызов функции:

  • return("decline"), который прерывает функцию трансформации и фильтрует полученное событие;

  • return("passthrough"), который передает входящие сообщения на следующий шаг без изменений (имеется возможность изменять заголовки и другую информацию при помощи DSL).

Вызов функции return() с любым другим параметром прервет дальнейшее выполнение правил трансформации, при этом уже заполненные выходы OUT и HEADERS будут обработаны и преобразованы в исходящие события.

Функция агрегации событий#

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

При выполнении DSL правил трансформации:

  1. Вход IN содержит события из окна агрегации;

  2. Environment.sources содержит массив переменных окружения для каждого события, индекс элементов соответствует индексам событий из IN. Каждый элемент массива содержит поля:

    • SourceTopic — содержит имя топика, откуда было прочитано сообщение;

    • SourceKey — содержит значение ключа из сообщения Apache Kafka в виде строки;

    • kafka.headers — содержит словарь заголовков из сообщения Apache Kafka;

  3. Environment.cache — содержит элементы, сохраняющиеся между вызовами функции агрегации;

  4. Environment.loadedVariables — содержит элементы, загруженные из JSON файла из параметра environmentVariables шага агрегации;

  5. Environment.AggregationKey — содержит значение ключа агрегации;

  6. Environment.AggregationWindowTimeMs — содержит продолжительность существования окна агрегации в миллисекундах;

  7. Environment.AggregationEventCount — содержит количество событий, переданных в функцию агрегации.