## 5 Proposed Implementation for Parametric Event Transformation
### Abstraction and Reusable Implementation: Time Windowed Moving Average Calculation  
Given a single homogeneous event stream with fine-grained events that represent different measurements of the same thing (or, rather, the same physical quantity), an abstraction is proposed to calculate the average of the last measurements of the specific physical quantity of each IoT device for a predetermined time window. Furthermore, the abstraction must take into account the minimum number of IoT devices required to calculate the average. This abstraction is parametric. The parameters are the physical quantity, the duration of the time window, and the minimum number of IoT devices that must contribute to the calculation. Therefore, for each parameter set, a different instantiation of the specific abstraction results. The result, commonly the desired output, is a meaningful Event for each time window with the average value, details about the time window, the contributing IoT devices, and more. In detail, the processing is depicted in Figure 5 and includes the following stages:  
**1 stream**: Consumption of data stream from the topic \"sensor_telemetry_[NAME]_mea surement_event\" where \"[NAME]\" is the dynamic portion of the topic name defined by the parameters and, more specifically, by the name of the physical quantity. The records of the specific event stream are of type \"SensorTelemetryMeasurementEvent\", and all concern the same physical quantity. It is noted that the event stream is homogeneous. It includes values in the same format, and all records are correct and valid. Any record that deviates is ignored from processing and managed via Dead Letter Queue. In addition, this specific step includes deserializing the message payload from bytes in the format required by Event Engine, i.e., \"SensorTelemetryMeasurementEvent\", which can be physically represented as either POJO or JSON. To determine the timestamp of the Kafka message, we use the custom implementation \"TimestampExtractor\" to ensure that each record has a correct and valid timestamp that will enable specific functions such as ordered consumption, dealing with late events, and windowing. **2. groupByKey**: The stage to follow groups the records by the existing key. Since all records have the same key, only one sub-group will be generated. This technique is also known as global aggregation. Since we have a single sub-stream by assigning the same key to all records, later processing nodes will be executed by a single thread. This is very convenient in our prototype implementation, but it could scale better. Thus, we recommend this only when small workloads are expected. Grouping is also a prerequisite for windowing and/or aggregating a stream and ensures that data are adequately partitioned (\"keyed\") for subsequent operations. **3. windowedBy**: The next step controls how the single sub-stream will be grouped into time windows for windowed aggregation, which is a stateful operation. We use hopping time windows which are windowed based on time intervals. They model fixed-sized overlapping windows, defined by two properties: the windows' duration and their advance interval. Since hopping windows can overlap, an event record may belong to  
Figure 4: Event processor topology of Splitting Composite Transformation. more than one such window. According to [PERSON], hopping windows are sometimes called \"sliding windows\" in other stream processing tools. Kafka Streams follows the terminology in the academic literature, where the semantics of sliding windows differ from those of the hopping windows. **4. aggregate**: At this point, records have been grouped into a single sub-stream and then regrouped into multiple windows according to the hopping time window definition. So, the stage to follow is the stateful processing of each windowed event stream. Each time a window is created, a data structure is initialized to hold calculations and other data throughout the time window.