Flow / Unit types / Core / Reducer
Reducer
Description
The Reducer unit is a Processor unit type.
A unit that combines the current event with the last reduced value. Can be configured to either use a global accumulator for all events or to create a new accumulator for each key. There are various possible outcomes depending on which port provides events to the Unit:
- An event comes in through the in port. If this is the first event in the sequence, the expression set in the initValue field is used for the evaluation. The result is stored into an accumulator. Each time an event is received through the in port, the predicate in the Expression field is executed and the internal accumulator is updated with the result. If there are no errors, the input event is enriched with a new field containing the accumulator value. Then, the event is output via the out port.
- When an event comes in through the get port, the input event ignores the expression and the reducer enriches the event with a new field carrying the current accumulator value of the key fields entered in the properties. In this case, the event is output via the current port.
- When the unit receives an event via the reset port, the internal accumulator is reset to the initial value (initValue) in the configuration.
If there are errors produced during the evaluation, these events are enriched with two additional fields, error and exception, and the event is output through the error port
Configuration
After dragging this unit into the Flow canvas, double-click it to access its configuration options. The following table describes the configuration options of this unit:
| Tab | Field | Description |
|---|---|---|
| General | Name | Enter a name for the unit. It must start with a letter, and cannot contain spaces. Only letters, numbers, and underscores are allowed. |
| Key field(s) | The name of input event field(s) containing key(s). | |
| Language | Specify the language you will use to write the expression, e.g. Javascript, Groovy, etc. | |
Init Value | An expression written in the specified language to specify the initial value of the accumulator. For example, 0L would be a long value starting at 0. | |
Field name | Enter a name for the field that will be added to the events with the reducer results. | |
Accumulator Type | The expected data type of the values generated by the reducer. e.g. Long. | |
Expression | Expression that will determine the behavior of the reducer. Use __acc __ to refer to the accumulator value. | |
| Overwrite | Whether to overwrite an existing input event field if one with the same name already exists. |
Input ports
| Port | Description |
|---|---|
| in | If new events enter through this port, the expression entered in the initValue field is evaluated and the result is stored. |
| get | Events that enter through this port are enriched with a new field carrying the current cumulative value. |
| reset | Events that enter through this port reset the internal accumulator to the initial value (initValue) in the configuration. |
Output ports
| Port | Description |
|---|---|
| out | Outputs events that enter through the in port and are enriched with the new cumulative value. |
| current | Outputs events that enter through the get port and are enriched with the current cumulative value. |
| error | Outputs all events that produced an error during evaluation. These events have two additional fields, error, and exception, that describe the error. |
Example
In this example, we want to send an alert via email when a user enters our Devo domain more than 1,000 times.
First, we will add a Devo Source unit to run the required query in the siem.logtrust.web.activity table and link it to a Filter unit that will filter events containing the required user mail. Then, we will add a Reducer to add a count to each of the events filtered by the previous unit. This is the configuration needed:
- Name - Enter a name for the unit (in this example, reducerCount).
- InitValue - Expression that determines the initial value of the reducer. In this example, we enter 0L.
- Field name - Enter a name for the field that will be added to the events with the reducer results. In this case, we will name it counter.
- Accumulator Type - The expected data type of the values generated by the reducer. In this case, we select Long.
Expression - Groovy expression that will determine the behavior of the reducer. In this case, we will add the following expression to get the count of occurrences:
__acc__ + 1
Finally, we will add a second Filter unit to filter events with a value of 1,000 or more in the counter column and send it to the required email using an Email Sink unit.
You can try this flow by downloading this JSON file and uploading it to your domain using the Import option.