Introduction
FME Flow Automations allow the creation of triggers - a way to receive content from clients over a supported protocol. Examples of triggers may be incoming emails, directory watching, webhooks, or SQS. These actions contain the logic to trigger a new event.
If you have a client or dataset that isn’t supported as an action in FME, you can create a workspace and schedule it to run on FME Flow at desired intervals. The workspace can then be used to trigger an action (this could be another workspace to do processing).
Workspace Requirements
To build a workspace that can check the state of a system or dataset, check that FME Workbench supports the data format or communication method.
If you are checking a dataset for changes or new features, does it have any fields that indicate when the data was changed, for example, a timestamp?
If the dataset has no way to identify changes, do you want to keep a copy of the data for comparison (caching), or just process the entire dataset?
Caching data
With data caching, data size and storage may be a consideration. If you have a large dataset, reading two datasets into a workspace for change detection may increase workspace runtime and memory usage. You will also need sufficient space to store a copy of the data.
Some workflows may need an up-to-date set of data (eg a database table of current data), in this case you will already have a dataset to compare against, and won’t need a duplicate for change detection. If your workflow reads a dataset and publishes a trigger (for example, an email notification or SMS) based on new data or changes, and does not store data, this may require data caching.
The ChangeDetector transformer can be used to compare two datasets.
If the data is cached or stored in a format that supports SQL queries, this can save time by avoiding the need for FME to read the entire dataset, only reading features where two values differ.
Scenario
In this scenario, a CARTO air quality dataset needs to be kept up to date. The current air quality values are retrieved from a JSON data feed and written to CARTO for visualization on a map.
There are several methods of doing this:
ChangeDetector
In this workspace, the existing CARTO data is being read in, along with the JSON air quality data feed.
Both datasets are passed through the ChangeDetector, matching data by UUID. Any added features will be sent to the CARTO writer to update the dataset.
FeatureReader
This workspace produces the same output as the ChangeDetector when using a FeatureReader. The FeatureReader is connected to the CARTO dataset, but will only read in requested features. In this case, this is specified by a WHERE clause:
@Value(uuid) = "uuid" AND @Value(aqi) != "aqi"The FeatureReader is looking for features where the UUID matches, but the air quality index is different. When these features are read in, there will be a conflict between air quality index values, so the FeatureReader is set to ‘Use Initiator’, which takes the most current air quality index value and sends it to the CARTO writer to update the dataset.
SQLExecutor
One step further is to use the SQLExecutor to identify features in the CARTO dataset where the air quality index value differs from the JSON data feed, and then perform the update via SQL.
UPDATE aqdata SET aqi='@Value(aqi)' WHERE uuid=@Value(uuid) AND @Value(aqi) != "aqi"Comparing Timestamps
If your data has a timestamp, you can use it to determine whether the data is newer or has been updated since the workspace last ran.
If you have regular schedules, you can use the DateTimeCalculator to determine if a timestamp is newer based on a time interval.
If there’s a chance that your job may be delayed in running, resulting in inconsistent time intervals, you can record the workspace last run time. This can be stored on FME Flow, to be read in and updated every time the workspace runs.
In this example, the workspace is split into two sections. The first section reads the last runtime of the workspace and passes that timestamp to the other half of the workspace using a VariableSetter and VariableRetriever.
The input data is an ATOM feed of roadworks in Vancouver that includes an updated timestamp. The difference between the workspace runtime and the data update is calculated. Any interval values greater than 0 indicate newly scheduled roadworks or updated existing ones, and these can be filtered for processing or notification.
FME Flow Schedules
Once a workspace has been published to FME Flow, it can be set to run on a schedule or as part of an automated schedule. For more information about triggering an automation on a schedule, see the article Schedule a Workspace to Run with FME Flow Automations.
The Schedules tab displays a list of existing schedules, including each schedule's information and whether it is enabled.
Creating a new schedule lets you enter a name, category, and description. The recurrence can be set to run at an interval, once, or according to a CRON expression for more complex scheduling.
After the correct workspace is selected, additional settings can be applied. To notify a topic on job success or failure, expand the Notifications view.
Some additional properties can be set on schedules, such as Run Until Canceled, Queued Job Expiry Time, and Running Job Expiry Time.