Агрегация событий#

Шаг предназначен для агрегации нескольких событий в одно с помощью модуля declarative-mapper.

Тип агрегации «aggregation»#

Представляет собой объект с полем type со значением "aggregation" и полями:

  1. name — имя шага;

  2. environmentVariables — загрузка переменных из файла;

  3. format — описание форматов входящих и исходящих событий:

  4. key — описание вычисление ключа события;

  5. trigger — описание триггера агрегации событий:

    • объект с полем type со значениями:

      • count — триггер подсчета заданного количества событий, в отдельном поле count задается число агрегируемых событий;

      • dsl — триггер с логикой работы на DSL, содержит поля аналогичные полю dsl;

      • time — триггер срабатывания по времени, срабатывает раз в сутки в заданное время;

      • timeFromEvent — триггер срабатывания по времени из события, срабатывает раз в сутки согласно заданному времени, полученному из события при помощи DSL;

    • также содержит поле timeout со значением таймаута в миллисекундах после получения последнего события;

  6. dsl — описание расположения правил агрегации;

  7. stateFull — сохранение Environment.cache в состоянии Flink, логическое, по умолчанию false;

  8. destination — следующий шаг потока;

  9. error — шаг обработки в случае возникновения ошибки;

  10. shouldLogDebug - флаг для вывода информации о результатах выполнения шагов dsl (при триггере с логикой работы на dsl);

  11. storage - настройки хранилища агрегации, отличающиеся от заданных в reactiveConfig.

Значения полей dsl и environmentVariables представляет собой объект с полями:

  1. source — тип источника правил, принимает значения:

    • file — правила расположены в файловой системе;

    • classpath — правила расположены в classpath JVM.

  2. path — путь до файла на диске или в classpath JVM.

Значения полей input и output могут иметь значения:

  1. json — для событий в формате JSON;

  2. xml — для событий в формате XML;

  3. событие в формате csv:

    • type — со значением csv;

    • columnSeparator — разделитель полей события;

    • fieldNames — список имен полей события для использования при трансформации, также определяет порядок полей.

  4. объект с полями:

    • type — со значением avro;

    • schema — список схем, значение аналогично полю dsl, объект имеющий логическое поле default со значением true, будет считаться схемой по умолчанию.

Пример агрегации событий#

{
  source: {
    name: "testInput"
    type: "test"
    topic: "input"
    config: ${defaults.kafka}
    destination: ${aggregationStep}
  }

  deadletter: {
    name: "deadletter"
    type: "destination"
    topic: "deadletter"
    config: ${defaults.kafka}
  }

  destination: {
    name: "output"
    type: "destination"
    topic: "output"
    config: ${defaults.kafka}
  }

  aggregationStep: {
    name: "dsl-aggregation"
    type: "aggregation"
    format: {
      input: "json"
      output: "json"
    }
    key {
      type: "dsl"
      path: "key.tr"
    }
    trigger: {
      type: "dsl"
      path: "trigger.tr"
      timeout: 100
    }
    dsl: {
      path: "simple.tr"
    }
    destination: ${destination}
    error: ${deadletter}
  }

  flow: {
    name: "Event Process test"
    source: [${source}]
  }
}

Avro-схема выбирается исходя из значения заголовка "schemaName" в сообщении Apache Kafka. Если заголовок не задан, то используется схема по умолчанию или первая в списке в случае, если никакая схема не помечена полем default со значением true.

Тип агрегации «aggregationJs»#

Представляет собой объект с полем type со значением "aggregationJs" и полями:

  1. name — имя шага;

  2. environmentVariables — загрузка переменных из файла;

  3. format — описание форматов входящих и исходящих событий:

  4. key — описание вычисление ключа события;

  5. trigger — описание триггера агрегации событий:

    • объект с полем type со значениями:

      • count — триггер подсчета заданного количества событий, в отдельном поле count задается число агрегируемых событий;

      • js — триггер с логикой работы на js, содержит поля аналогичные полю dsl;

      • time — триггер срабатывания по времени, срабатывает раз в сутки в заданное время;

      • timeFromEvent — триггер срабатывания по времени из события, срабатывает раз в сутки согласно заданному времени, полученному из события при помощи js;

    • также содержит поле timeout со значением таймаута в миллисекундах после получения последнего события;

  6. js — описание расположения правил агрегации;

  7. stateFull — сохранение Environment.cache в состоянии Flink, логическое, по умолчанию false;

  8. destination — следующий шаг потока;

  9. error — шаг обработки в случае возникновения ошибки;

  10. shouldLogDebug - флаг для вывода информации о результатах выполнения шагов dsl (при триггере с логикой работы на dsl);

  11. storage - настройки хранилища агрегации, отличающиеся от заданных в reactiveConfig.

Значения полей js и environmentVariables представляет собой объект с полями:

  1. source — тип источника правил, принимает значения:

    • file — правила расположены в файловой системе;

    • classpath — правила расположены в classpath JVM.

  2. path — путь до файла на диске или в classpath JVM.

Значения полей input и output могут иметь значения:

  1. json — для событий в формате JSON;

  2. xml — для событий в формате XML;

  3. событие в формате csv:

    • type — со значением csv;

    • columnSeparator — разделитель полей события;

    • fieldNames — список имен полей события для использования при трансформации, также определяет порядок полей.

  4. объект с полями:

    • type — со значением avro;

    • schema — список схем, значение аналогично полю js, объект имеющий логическое поле default со значением true, будет считаться схемой по умолчанию.

Пример агрегации событий#

{
  source: {
    name: "testInput"
    type: "test"
    topic: "input"
    config: ${defaults.kafka}
    destination: ${aggregationStep}
  }

  deadletter: {
    name: "deadletter"
    type: "destination"
    topic: "deadletter"
    config: ${defaults.kafka}
  }

  destination: {
    name: "output"
    type: "destination"
    topic: "output"
    config: ${defaults.kafka}
  }

  aggregationStep: {
    name: "dsl-aggregation"
    type: "aggregationJs"
    format: {
      input: "json"
      output: "json"
    }
    key {
      type: "js"
      path: "testJs/aggregation/simpleKey.js"
    }
    trigger: {
      type: "js"
      path: "testJs/aggregation/simpleTrigger.js"
      timeout: 100
    }
    js: {
      path: "testJs/aggregation/simpleAgg.js"
    }
    destination: ${destination}
  }

  flow: {
    name: "Event Process test"
    source: [${source}]
  }
}

Avro-схема выбирается исходя из значения заголовка "schemaName" в сообщении Apache Kafka. Если заголовок не задан, то используется схема по умолчанию или первая в списке в случае, если никакая схема не помечена полем default со значением true.

Пример вычисления ключа агрегации на JS#

function transform(environment, inputs) {
    let output = createEmptyOutputItem()
    let status = "accept"
    if (typeof inputs[0].headers.testHeader !== 'undefined' && inputs[0].headers.testHeader !== null) {
        status = inputs[0].headers.testHeader
    } else {
        status = "defaultKey"
    }

    return {output,status,environment};
}

Пример вычисления триггера на JS#

function transform(environment, inputs) {
    let output = createEmptyOutputItem()
    let status = "accept"
    if (inputs[0].body.message === "second" && typeof environment.cache.first !== 'undefined' && environment.cache.first) {
        status = "FireAndPurge"
    } else if (inputs[0].body.message === "second") {
        environment.cache.second = true
        status = "continue"
    } else if (inputs[0].body.message === "first" && environment.cache !== null && typeof environment.cache !== 'undefined' && environment.cache.second !== null && typeof environment.cache.second !== 'undefined'&& environment.cache.second) {
        status = "FireAndPurge"
    } else if (inputs[0].body.message === "first") {
        environment.cache.first = true
        status = "continue"
    } else {
        status = "purge"
    }

    return {output,status,environment};
}

Пример правила агрегации на JS#

function transform(environment, inputs) {
    let output = createEmptyOutputItem()
    output.key = inputs[0].key;
    output.headers = inputs[0].headers;
    if (inputs.length > 1) {
        output.body.message = inputs[0].body.message + ";" + inputs[1].body.message
    } else {
        output.body.message = inputs[0].body.message + ";" + inputs[0].body.message
    }
    output.branch = inputs[0].branch;
    output.source = inputs[0].source;
    output.destination = inputs[0].destination;
    output.key = environment.AggregationKey.value;
    output.headers.AggregationKey = environment.AggregationKey;
    output.headers.AggregationWindowTimeMs = environment.AggregationWindowTimeMs;
    output.headers.AggregationOnTimeout = environment.AggregationOnTimeout;
    let status = "accept"
    return {output,status};
}

Логика потока агрегации событий#

Поток агрегации состоит из 3 частей:

  1. Вычисление ключа события и сортировка потока событий по ключу;

  2. Временное окно и триггер отправки событий в функцию агрегации;

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

Вычисление ключа событий#

Логика вычисления ключа для события задается параметром key, объект с полем type которое принимает следующие значения:

  1. key — вычисление ключа по значению ключа из сообщения Apache Kafka, представляет собой объект с полем type равным key;

  2. headers — вычисление ключа по определенным значениям заголовков, представляет собой объект с полем type равным headers и в поле keys содержит массив заголовков, из которых должен формироваться ключ. Значение ключа является объединением значений заголовков с разделителем |.

  3. hash — вычисление ключа хеш функцией из тела сообщения. Представляет собой объект с полем type равным hash, опционально в поле algorithm содержит имя хеш алгоритма, по умолчанию SHA-256.

  4. dsl — вычисление ключа с помощью правил трансформации, объект содержит поля:

    • type со значением "dsl";

    • source — источник правил трансформации classpath или file;

    • path — путь до правил трансформации.

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

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

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

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

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

При вычислении ключа с типом dsl правила трансформации должны завершаться вызовом функции return(<key>), где <key> — значение ключа. Выходы OUT и HEADERS скрипта трансформации игнорируются.

Окно агрегации событий#

После вычисления ключа событие попадает в окно агрегации. В универсальном обработчике используется глобальное окно агрегации.

Session Window

В нашем случае параметр timeout у триггера задает session gap, то есть максимальное время между получением событий с одним ключом. В момент возникновения таймаута все собранные события в окне передаются в функцию агрегации, а окно агрегации очищается.

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

Окна до объединения#

Before merge

Окна после объединения#

After merge

В процессе объединения окон происходит объединение состояний, после чего вызывается триггер.

Триггер отправки событий в функцию агрегации#

Предусмотрено 4 вида триггера:

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

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

  3. Триггер срабатывания по времени, срабатывает раз в сутки в заданное время.

  4. Триггер срабатывания по времени из события, срабатывает раз в сутки согласно заданному времени, полученному из события при помощи DSL.

Настройка триггера подсчета количества событий#

Задается объектом в поле trigger шага агрегации с полями:

  1. type — со значением "count";

  2. count — целое число, равное количеству ожидаемых событий. Триггер срабатывает в момент достижения количества событий этого значения;

  3. timeout — время в миллисекундах после получения последнего события в окне агрегации и до срабатывания триггера по таймауту;

  4. shouldFireAtProcessingTime — должен ли срабатывать триггер по таймауту относительно получения первого события, значение по умолчанию true.

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

trigger: {
    type: "count"
    count: 2
    timeout: 100
}

Настройка триггера с логикой на DSL#

Задается объектом в поле trigger шага агрегации с полями:

  1. type — со значением "dsl";

  2. source — источник правил трансформации, может принимать значения "classpath" и "file", по умолчанию "classpath";

  3. path — путь до файла с правилами;

  4. timeout — время в миллисекундах после получения последнего события в окне агрегации и до срабатывания триггера по таймауту;

  5. shouldFireAtProcessingTime — должен ли срабатывать триггер по таймауту относительно получения первого события, значение по умолчанию true.

trigger: {
    type: "dsl"
    source: "file"
    path: "script.tr"
    timeout: 100
}

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

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

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

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

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

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

  5. Environment.TriggerFireMs содержит время вызова триггера в миллисекундах в формате Unix;

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

DSL скрипт правил трансформации должен вернуть функцией return() один из следующих статусов:

  1. return("continue") — продолжение накопления событий в окне агрегации;

  2. return("fire") — передача накопленных событий в функцию агрегации без очистки и закрытия окна;

  3. return("purge") — очистка и закрытие окна без передачи накопленных событий в функцию агрегации;

  4. return("fireAndPurge") — передача накопленных событий в функцию агрегации с очисткой и закрытием окна.

Выходы OUT и HEADERS в триггере игнорируются.

Если функция return() не была вызвана или вызвана с другим значением, то накопленные события будут переданы в функцию агрегации, а окно очищено и закрыто.

Настройка триггера срабатывания по времени#

Задается объектом в поле trigger шага агрегации с полями:

  1. type со значением "time";

  2. time — время, равное ожидаемому времени срабатывания, триггер срабатывает в момент достижения текущего времени этого значения;

  3. timePattern — формат времени;

  4. maxEmptyIntervals — количество пустых интервалов до сброса ключей агрегации.

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

trigger: {
  type: "time"
  time: "11:05:00+03:00"
  timePattern: "HH:mm:ssXXX"
  maxEmptyIntervals: 3
}

Настройка триггера срабатывания по времени из события#

Задается объектом в поле trigger шага агрегации с полями:

  1. type со значением "timeFromEvent";

  2. source — источник правил трансформации, может принимать значения "classpath" и "file", по умолчанию "classpath";

  3. path — путь до файла правил трансформации с логикой срабатывания триггера по времени из события на DSL;

  4. timePattern — формат времени;

  5. maxEmptyIntervals — количество пустых интервалов до сброса ключей агрегации.

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

trigger: {
  type: "timeFromEvent"
  path: "time.tr"
  timePattern: "HH:mm:ss"
  maxEmptyIntervals: 3
}