## 5 Proposed Implementation for Parametric Event Transformation
### Composite Transformation: Ingestion  
The motivation for creating the Ingestion Composite Transformation is ingesting measurements of natural quantities from sensors in the natural environment. The goal is to clean, validate, enrich, and homogenize the events. The user defines a vocabulary of physical quantities as parameters with individual details for each physical quantity. In this case, the details include instructions for converting measurements to a specific unit of measure, such as Fahrenheit to Celsius. As an extra degree of parameterization, the user also defines how physical quantities not included in the predefined quantity vocabulary are treated, for example, keep the events in a topic or discard them. Each IoT device with sensing capabilities measures one or more things from the physical environment and sends them to the IoT Gateway. The IoT Gateway converts these data into an IoT Event with a specific schema that includes all measurements and valuable information, such as the IoT device ID and the measurement time. Furthermore, it promotes all instances of this type of IoT Event to a topic creating a single event stream. Therefore, a necessary abstraction is ingestion for sourcing and manipulating these coarse-grained primitive IoT Events into fine-grained IoT Events that each includes only one measurement and carries the minimum possible semantics. Moreover, transformed IoT Events are promoted to a different but defined topic, creating a single event stream. This enables further processing by other components without repeating these common transformations. In the left side of Figure 3, an event stream is depicted that has records of the type \"SensorTelemetryEvent\" (i.e., all records have the same schema). The topic of this event stream is \"sensor_telemetry_event\". In the right side of Figure 3, the desired generated event stream is shown, with records of type \"SensorTelemetryMeasurementEvent\" (i.e., all records have the same schema). The topic of the generated event stream is \"sensor_telemetry_measurement_event\". That is, the second generated event stream includes a record for every measurement from every record of the first event stream. This abstraction is parametric. The parameter is a list of the physical quantities to include in the generated event stream. The physical quantities that are not included in the list are discarded. In detail, the processing includes the following stages:  
**1. stream**: Consumption of data stream from the topic \"sensor_telemetry_event\", i.e., the IoT Events, from the Event Bus sorted by the store timestamp. Also, this step includes deserializing the message payload from bytes in the format required by Event Engine, i.e., \"SensorTelemetryEvent\", which can be physically represented as either POJO or JSON. Comment: 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. process**: Pre-processing that involves converting specific values into the format required. More specifically, if an IoT Event does not have a timestamp, the processing timestamp is used. Second, since the IoT Event has a timestamp, it is checked for validity and then converted to a numeric representation as an epoch/Unix timestamp. That covers two prominent cases. First, the case where the IoT device does not include the event timestamp. Second, the case where the event timestamp is in a different format or in a different time zone than the one Event Engine requires. Event Engine only supports Unix timestamp as a universal convention so that there is homogeneity in the representation of instantaneous points in the timeline. **3. flatMapValues**: Each \"SensorTelemetryEvent\" can contain more than one measurement.