3.1   Transformed telemetry tables¶

Let’s use the WeatherStation data to examplify the creation of a transformed telemetry table.

The rationale for the suggested schema is the following:

• The transformed WeatherStation telemetry table combine information from multiple raw WeatherStation telemetry topics.
• Fields that are not relevant to the Science Platform user can be excluded. In particular, most of the private_ fields added by SAL can be excluded.
• In this example, the original topics have aggregated fields like min24H, avg24H, max24H. We decided to keep only the “1 minute average” fields, which is the higher resolution available for all the WeatherStation telemetry topics.

Note: despite their names, the value and value10M fields for the windDirection, windSpeed and windDirection topics also have 1 minute average values.

• In the transformed table, we decided to prefix the fields with the source WeatherStation topic name to identify its origin.
• The timestamp field in the transformed table is derived from the private_efdStamp field. The other timestamps are discarded.

Note: the timestamps from the raw WeatherStation telemetry topics are not necessarilly aligned, see section 5.2   Joining for details.

From this example, and also after looking at a handful of other T&S subsystems, we conclude that:

• the EFD transformation service must specify a mapping between the source telemetry topics and the transformed telemetry table, and which fields within those topics to use.
• in some cases the EFD transformation service needs to apply transformations to the fields’ values, and must allow for new descriptions and units for the transformed fields.

The EFD transformation service holds the decisions necessary to create the transformed telemetry tables from the raw telemetry topics.

3.2   Advantages¶

Some advantages of this approach:

• Instead of 249 tables we might have 62, one per subsystem, or even less. By reducing the number of tables in the Consolidated Database we simplify the queries considerably and reduce the number of joins and relations in the database.
• By exposing only the relevant information to the Science Platform user, we also reduce the amount of data in the Consolidated Database making it more managable over time.
• By transforming field values, we add value and make the EFD telemetry data easier to use.
• Another task of the EFD transformation service is the aggregation of field values over time for high-frequency telemetry streams, which also reduces the amount of data in the Consolidated Database to a great extent.

In the following sections, we describe the Kafka Connect JDBC Sink connector and kafka-aggregator. We try to use the JDBC Sink connector features as much as possible, and delegate to kafka-aggregator the functionalities that cannot be performed by the connector.

4.1   Schema creation¶

The Kafka Connect JDBC Sink connector requires an explicit schema to automatically create a table in a relational database. In the EFD, we accomplish that by using Avro and storing the Avro schemas in the Confluent Schema Registry.

4.2   Schema evolution¶

The JDBC Sink connector has limited support to schema evolution, but it supports forward compatible schema changes with PostreSQL and can automatically issue an ALTER TABLE to add new columns to an existing table.

4.3   JDBC Sink transforms¶

'transforms'                          = 'flatten',
'transforms.flatten.type'             = 'org.apache.kafka.connect.transforms.Flatten$Value'

'transforms.setTimestampType.type'        = 'org.apache.kafka.connect.transforms.TimestampConverter$Value',
'transforms.setTimestampType.field'       = 'timestamp',
'transforms.setTimestampType.target.type' = 'Timestamp'

4.4   Declaring primary keys¶

The natural choice for the primary key in the transformed telemetry tables is the timestamp field. To do that, pk.mode must be set to record_value to use one or more fields as primary key.

'pk.mode'                                 = 'record_value',
'pk.fields'                               = 'timestamp',

4.5   Working with multiple tables¶

When working with multiple tables, database ingestion can be improved by adding more Kafka Connect workers. There are two ways of doing this within the Kafka Connect framework. One is to define multiple connectors, one for each table. The other is to create a single connector but increase the number of connector tasks.

With the InfluxDB Sink, MirrorMaker 2 and S3 Sink connectors, creating a single connector and increasing the number of connector tasks seems enough to handle the current data throughput in the EFD. This should work with the JDBC Sink connector too, as long as we can use the same connector configuration for all tables.

5.1   Kafka-aggregator¶

kafka-aggregator is based on Faust, a Python Stream Processing library. It implements Faust agents that subscribe to one kafka topic and produce a new aggregated topic.

In the current implementation kafka-aggregator computes summary statistics on tumbling windows (window aggregation), where the size of the window sets the frequency of the aggregated stream. However, it can aggregate only one source topic and produced one aggregrated topic at the time (the 1:1 case).

The main extensions to kafka-aggregator include the ability of joining multiple Kafka topics into one source stream, and transformations like filtering and mapping in addition to window aggregation. All of these transformations are chained together to produce the transformed teletemery streams.

5.2   Joining¶

With Faust, it is possible to subscribe to multiple source topics by listing them in the topic description. Faust also supports other join strategies.

5.3   Filtering and mapping¶

The extended kafka-aggregator requires a new mechanism to define the schema for the aggregated topics. In the new implementation, kafka-aggregator specify the source Kafka topics to an aggregated Kafka topic, by explicilty listing the source topics and fields to use.

In the same mapping configuration, we can specify functions to transform the field values (map), if needed, and enable or disable window aggregation on fields.

We propose replacing some the kafka-aggregator settings by a configuration file like the following:

---
aggregated_topic_name1:
   filter:
      source_topic_name1:
         field1:
            name: new_name
            description: "new description for the transformed field"
            units: "new units for the transformed field"
            map: func1
         field2:
            description: "new description for the transformed field"
            units: "new units for the transformed field"
            map: func2
         field3:
         ...
      source_topic_name2:
         field1:
         field2:
         field3:
         ...
      ...
aggregated_topic_name2:
   window_aggregation_size: 1s
   window_expiration_seconds: 1s
   operations:
      - min
      - median
      - max
   filter:
      source_topic_name3:
         field1:
         field2:
         field3:
         ...
      ...
...

In this configuration, we specify the aggregated topic (the new topic in Kafka where the aggregated data is produced to), the source topics in Kafka to consume from, and the fields within those topics to use.

For each field, we can specify optionally a name, a description, units and a function (map) to transform its values. If not specified, the default field name, description, and units are obtained from the source topic schema. If a map is not specified, the value from the source field is used.

5.4   Window aggregation¶

For each aggregated topic we can optionally specify the window aggregation transformation, configuring the tumbling window parameters and the summary statistics to compute (operations).

The allowed operations are min, q1, mean, median, q3, stdev and max.

The window aggregation is computed after the map transformation if any.