Desktop Tips for Working with Continuous Data Streams

Junction Order

Order is also important after the initial read. If you are making copies of features for different parts of the stream, there might be a need to make sure one part of the workflow runs first. For example, ensuring features go into a transformer downstream that requires features to come in first—e.g. NeighbourFinder with Candidates First set to Yes.

To control this right-click on the connection where the features are duplicated and select “Set Creator Runtime Order”.

Process When Group Changes

Unless you are doing basic processing (filtering or event detection), the stream needs windowing into groups so you can process and summarize data. Using Group By in combination with the TimeWindower is how you achieve this. This article goes into detail on how to use the TimeWindower to break a stream into groups for processing. 

Group by Window ID

The TimeWindower assigns a different window ID to all features in each window. It is this window ID that can then be used to ensure the transformers with this set will process all features in that window as one group, and then flush when the next window arrives. When running in stream mode you need to make sure that the Group By Mode is set to "Process When Group Changes".

Grouping By Multiple Keys

Grouping by the time window ID plus another key is supported, it just can be tricky as you need to ensure the features are sorted correctly to ensure delivery is in the correct order. The Sorter transformer can help do this. Reducing data volumes is a key workflow when working with streams and the Sorter + Sampler is needed for this.

Example

To explain we can walk through an example. Fifty thousand vehicles are reporting their location every five seconds. Rather than storing all of that data, we want to use FME to remove duplication and lower the resolution to 30 seconds.

1. The TimeWindower transformer then groups the messages into 30-second windows.

2. Messages in each window are then sorted based on the vehicle ID and timestamp. This means the data is grouped by vehicle and then the vehicle's location is sorted with the first known location of the vehicle in the window first and the last known location of the vehicle last. Note we are also using the Window ID as the group by.

3. Now the data is grouped by window ID and vehicle ID and sorted by time, a Sampler can be used to get the last location for each vehicle in the window. Note we are now grouping by window ID and vehicle ID.

In this example, if you don’t use the Sorter the vehicle ID will change multiple times back and forth within a single window ID which would cause the Group By to produce undesirable results. With the right grouping, it is easy to get the first or last in a window. Because you are only storing the features in each window in memory then it is performant.
 

Writing Out File-Based Data

In most scenarios writing out to a database is likely the best option. More details on this here. There are some scenarios where you may need to write a stream out to a file, for example, writing out data such as CSV or Parquet to a data lake.

If you do need to write a stream out to a file you need to use a FeatureWriter. The FeatureWriter must be used in combination with a TimeWindower and the following need to be configured on the Feature Writer:-

• Group Processing: Set the attribute to the Window ID (and a secondary attribute if required). Make sure the Group By Mode is set to “Process When Group Changes (Advanced)”.
• Set the filename to be dynamically generated based upon the group by. For example Name_@Value(_window_id).

  • Note that a DateTimeConverter may be useful to convert a time-based window ID (which is stamped in seconds since epoch -- %s in DateTimeConvertor lingo) to a human-readable timestamp to use as part of the generated filename.  

Below is an example that writes out a CSV file. A new file is created every time a window changes and the files will be written out with the name Report_<WindowID>.csv.