FairCom Edge concepts

Integration tables are the heart of FairCom Edge. They are database tables that are enhanced and optimized for inserts, transformation processes, queries, and forwarding data asynchronously to output services. They make it possible to bridge data across otherwise incompatible protocols, data serialization formats, data shapes, and data types. All integration tables are created with only insert and read permissions. This is because an integration table is an immutable, audit log of events.

Input services insert data into integration tables. Output services read and deliver data from the integration tables.

Input services insert records, they are not allowed to change or delete records because that would destroy the integrity of the audit log. When an input source submits data for insertion, automatic transforms are created and stored in the record with the original data. It becomes read-only and can be queried by SQL and JSON DB APIs.

When an input service is configured, it is assigned to an existing integration table or it creates a new integration table. MQTT automatically creates an integration table for each new topic it creates. Integration processes can be refined and integration tables can be renamed to better label the data they hold. The "configureTopic" and "alterIntegrationTable" actions in the API are used to rename an integration table.

Figure 2, “The integration table” identifies the FairCom Edge components that interact with the integration table.

Figure 2. The integration table

Each integration table record contains the fields shown in Figure 3, “Integration table fields”.

Figure 3. Integration table fields

Since the MQTT engine requires additional fields to track message data and since any integration table can be connected to the MQTT engine, each integration table contains MQTT fields.

Plugins can also add fields to each integration table. This occurs automatically when such a plugin is added to FairCom Edge. It is done invisibly in the background with no performance impact because of the FairCom Hot Alter Table feature.

The SQL and JSON DB APIs should only write to the "source_payload" field and user-defined fields. They can query all fields.

FairCom Edge uses the same integration process across all its services — see Figure 4, “The integration data”.

Figure 4. The integration data

The process starts with an input service receiving or gathering data from a device, user, or application and inserting a record into an integration table. The incoming data is placed in the "source_payload" field of the integration table record. Prior to committing the insert, FairCom Edge automatically runs transformations on the source data and places the results of those transformations in fields created by the user. The Table Transform Service can optionally extract data from JSON properties in the "source_payload" and insert the data into the fields of a record in another table.

After the transform process finishes, FairCom begins to asynchronously deliver each inserted record to the non-query-based output services. Output services deliver data to external systems. Query services, such as SQL and JSON DB APIs can query data in integration tables at any time.

The delivery process

FairCom's delivery engine is available for all integration tables. In Figure 5, “The delivery process”, the right side of the integration table is the forward part of the Store and Forward delivery model used by FairCom Edge.

The delivery engine pushes data to outputs with guaranteed delivery in the order data is received. It is used to deliver data to output services such as MQTT, ThingWorx, AWS IoT Core, and other MQTT brokers. The delivery engine creates multiple processes that read records from each integration table and push the data to output services. Each delivery process delivers data in the order it is inserted into the integration table. Each delivery process runs independently (asynchronously) from all other output services, which means no output service can block another output service.

Figure 5. The delivery process

The delivery engine guarantees the delivery of data to output services. If anything in the output chain is not working properly (such as a failed network, an offline MQTT client, or a failed external system) the integration table retains the stored data. When the external system is available, the delivery engine starts where it left off and forwards the data to the external output service.

Each integration table is configured to retain data for a different period of time. It can also be configured to forward data to external systems at maximum rates to avoid saturating the network or overwhelming the external system.

When a record is inserted into an integration table, it triggers the transformation engine for that table — see Figure 6, “The transform process”.

Figure 6. The transform process

A transform process consists of one or more steps. All steps run within a single database transaction. You configure each transform step to get its data from one or more fields in the table. The "source_payload" field is the most common source for transform steps because most input processes, such as MQTT, store their data in "source_payload". Multiple transform steps may share the same input field or fields.

You configure each transform step to run a transform method with specific parameters.

You also configure each transform step to store its results in one or more output fields. Once a transform step has written to a field, the transform engine prevents any other transform step from writing to that field. Transform steps cannot be written to the same output fields because that would make it impossible to troubleshoot transform problems. Likewise, transform steps cannot be written to the "source_payload" field.

If a transform fails, error processing is performed, which logs the error in the log field and sets the error field to 1 (true). A failure in one transform does not prevent other transforms from running and does not prevent the record from being inserted.

The transform process is synchronous and completes before the record is inserted.  Once a record is inserted, the source payload and transformed field values can be queried by SQL and the JSON DB APIs, and FairCom's store-and-forward engine engine starts delivering the record to all configured outputs.

Each MQTT topic in FairCom Edge and each integration table can be configured to run its own transform process.

{
  "transformErrors": 
  [
    { 
      "errorCode": 0, 
      "errorMessage": "", 
      "errorData": {} }
  ]
}

JSON to table transform

The JSON-to-table transform extracts properties from a JSON payload in the "source_payload" field and inserts a record into another table. The destination integration table can then be used as a source for output services as shown in Figure 7, “Transform JSON to a table”.

Figure 7. Transform JSON to a table

FairCom Edge uses integration tables to connect inputs to outputs. One or more inputs can be mapped to the same integration table for inserting data — for example, MQTT, Modbus, and OPC UA can all insert data into the same integration table.

One or more outputs can be mapped to the same integration table — for example, MQTT and Thingworx can be mapped to the same integration table, and it will push each inserted record to each output.

Each integration table can be thought of as a topic — for example, each MQTT topic uses a different integration table to store its messages.

You can configure each input service to use a different integration table when each input represents a different type of data or a different MQTT topic. Conversely, you can configure multiple input services to feed data to the same integration table when each input represents the same type of data or the same MQTT topic.

Some input services automatically create and use their own integration tables. So, each time a topic is created in MQTT, FairCom Edge automatically creates a new integration table to store its messages.

Some input services, such as OPC UA and Modbus, create a table when they are configured to collect and store data.

Example 3. Connect input and output services

Each integration between a service and an integration table is configured separately. In Figure 8, “Connect inputs to outputs”, each arrow going to or coming from an integration table represents a different integration. An Input Integration connects an input service to an integration table. Likewise, an output integration connects an integration table to an output service.

Figure 8. Connect inputs to outputs

• Input integration 1

  An application configures an OPC UA service to collect a specific set of data from a device and insert it into an integration table. FairCom Edge automatically creates the integration table. Most input services automatically create integration tables to store the data they collect, but they can also be configured to use an existing integration table.

• Input integration 2

  An IT message broker, such as RabbitMQ, ActiveMQ, or IBM MQ, receives enterprise messages. It is configured to use its MQTT plugin to publish messages to a topic on FairCom Edge's MQTT broker. In FairCom Edge, a topic is mapped to one integration table. Each MQTT message inserts a record into the appropriate integration table. The FairCom delivery engine delivers each record to each output that is mapped to that integration table.

• Output integration 3

  The MQTT output service is configured to use the integration table as a source for sending messages to a topic. This associates the integration table with the named MQTT topic. The MQTT output service uses the forward engine to read each inserted record and asynchronously publish it as an MQTT message to all clients that subscribed to that topic. Thus, any input service, such as OPC UA, Modbus, SQL, JSON DB API or MQTT, can insert a record into an integration table assigned to an MQTT topic, and the MQTT output service will publish a message to all subscribers of that topic.

• Output integration 4

  The ThingWorx connector service is configured to use the integration table as a source. The ThingWorx connector uses the forward engine to asynchronously deliver each inserted record to ThingWorx.

• Output integration 5

  FairCom Edge can forward MQTT messages to any other MQTT broker. Cloud providers, such as AWS IoT Core and Azure IoT Hub, provide MQTT brokers to receive IoT data and route it to cloud services, such as machine learning, data lakes, data warehouses, etc. Conversely, FairCom Edge can subscribe to any other MQTT broker, which allows it to receive messages back from cloud providers.

FairCom Edge uses the same integration process across all its services — see Figure 9, “Configure input and outputs”.

Figure 9. Configure input and outputs

Input and output services can be modified at any time to route data where it needs to go.

The MQTT engine automatically maps MQTT topics to integration tables. Each MQTT topic can configure this mapping using the "configureTopic" action. Applications can add, modify, and delete these mappings at any time.

Use the "createInput" action to configure an input service. This causes FairCom Edge to create an integration table to store the input from that service. If you plan on attaching multiple inputs to the same integration table, you can first use the "createIntegrationTable" action to create an integration table, and then use the "createInput" action to connect the integration table to one or more input services.

Some input and output services do not need configuration — see Use SQL and JSON DB APIs.

Output services, such as the REST connector, read data from a table. They cannot insert, update, or delete data.

Input services collect, transform, and insert data into integration tables, which are historical logs of collected data. When FairCom Edge transforms data, it stores the original source data along with the result of each transformation. Transform services do not modify a field that was written to by another transform service. Thus, the history of all data inserts and transforms is always available for auditing and troubleshooting.

A connector can be used as an input service, output service, or both.

FairCom Edge separates MQTT into two parts, the MQTT Pub and MQTT Sub services. The MQTT Pub service is an input service that receives published MQTT messages and the MQTT Sub service is an output service that delivers MQTT messages to subscribers. These services are decoupled as shown in Figure 10, “Configure MQTT”.

Figure 10. Configure MQTT

In Figure 10, “Configure MQTT” the arrow between the MQTT Pub service (the input service) and the integration table represents the input integration. The arrow between the integration table and the MQTT Sub service (the output service) represents the output integration. All input and output services are connected through integration tables.

Each MQTT topic is bound to a dedicated integration table that stores messages sent to a topic and delivers messages to subscribers of the topic. FairCom Edge automatically creates this table when a message is first published to a topic or when the "configureTopic" action is used to create the topic.

Each time the MQTT input service receives a message, it inserts the message into the "source_payload" field of a new record in the topic’s integration table. Transforms optionally transform this data and put results in the user-defined transform fields.

Each MQTT topic is configured using the "configureTopic" action which configures the topic, the integration table, the MQTT input service, and the MQTT output service.  If the topic being configured has not yet been published to FairCom Edge, the "configureTopic" action automatically creates the integration table, the topic, and its configuration.

Optionally, when "configureTopic" creates a new MQTT topic, it can assign an existing integration table to the topic. Thus, an integration table may be created using a protocol such as Modbus, and later bound to an MQTT topic. Similarly, an integration table may be created by the "createIntegrationTable" action and later bound to an MQTT topic.

The MQTT output service uses FairCom’s delivery engine to asynchronously read records that are in the integration table. The "configureTopic" action reads the MQTT payload from a field in the table. By default, it reads from the "source_payload" field, but it can be configured to read from one of the user-defined transform fields. The MQTT output service then delivers the message to MQTT clients that are subscribed to the topic. FairCom’s delivery engine guarantees the delivery of messages to subscribers in the order they are received.

The SQL and JSON DB APIs directly use integration tables without prior configuration. Thus, applications and users can use these APIs to insert, update, delete, and read from any table their account has permission to access. Customers typically limit accounts to insert records into an integration table to ensure it is a tamper-resistant log of collected data. An administrator account can update, delete, insert, and read records, and it can grant permission to other accounts to do the same.

Each arrow in Figure 11, “Use SQL and JSON DB APIs” represents an account logged into these APIs with the appropriate permissions to insert data into and query the integration table. Each integration table may grant different accounts different permissions.

Figure 11. Use SQL and JSON DB APIs

FairCom Edge’s simple integration process can be repeated to form complex connections from any input to any output through integration tables.

Example 4. Complex integration example

Figure 12, “Integration engine process” shows the MQTT Pub service receiving messages sent to a topic and placing them into integration table A. The MQTT Subscriber service automatically retrieves these messages and delivers them to subscribers.

Figure 12. Integration engine process

integration table A has been configured to run a payload transform that extracts a nested array of JSON objects from the payload of each record inserted into integration table A. It transforms these nested records into tabular data and inserts these records into integration table D. A machine learning application uses SQL Queries to read that data to provide feedback to factory processes.

OPC UA is configured to poll a device every minute for specific telemetry data. It inserts the collected data into integration table B. It runs a transform to convert the data into JSON. The MQTT Subscriber service is configured to use integration table B as the source of an MQTT topic. It sends each record inserted into integration table B to subscribers of that topic. Additionally, the ThingWorx connector is configured to use integration table B to get inserted records from this table and deliver them to a digital twin in ThingWorx.

A table transform process has been configured to use the delivery engine to get inserted records asynchronously from integration table A. It transforms these records into tabular data and inserts them into integration table C. An application uses the JSON DB API to gather data into a manufacturing execution system. In addition, the ThingWorx connector is configured to use the delivery engine to get inserted records asynchronously from this table and guarantee their delivery to a digital twin in ThingWorx.

FairCom Edge has standardized on JSON as the primary data interchange format. JSON is the preferred data format in FairCom Edge because it is easy to convert into other formats. It is also human-readable, which makes it good for troubleshooting. Standardizing on JSON simplifies the development and troubleshooting of input and output services. For these reasons, JSON is the most common data format used in IoT and MQTT payloads.

FairCom Edge supports any type of data, such as XML, binary images, Siemens binary user-defined types, SparkplugB, and so forth. FairCom Edge supports transform services that can convert binary data to JSON and convert from JSON to binary data. FairCom Edge can also convert JSON to tabular data by extracting values from JSON properties into the fields of a record in a table.

Each output and transform service is designed to read, process, and transform a specific type of data. When you bind an output or a transform service to an integration table, you must designate one of its fields as its data source. This field is typically the "source_payload" field, but it may be a user-defined transformation field. The maximum field size is 2GB which is also the maximum record size. Each of these fields can contain any type of binary data, but the data type must match the data type expected by the output or transform Services that are bound to it.

Because output and transform services expect their source data to be in a specific format, data often needs to be transformed. It is typical to use one transform to convert from a proprietary binary format into JSON and use another transform to convert that JSON into a different proprietary format. This leverages JSON as a universal intermediary.

FairCom JSON APIs often have a property called "propertyPath", which allows you to specify the location of a property you want to create or a property value you want to retrieve. For example, FairCom Edge device connectors collect data from IoT devices and put it in a JSON document to deliver to other protocols. The "propertyPath" property is what enables you to specify the location where the connector will put the data in the JSON document.

FairCom uses a simple version of JSON Path to specify the location of properties in a JSON document. It uses the period "." character to define the nesting of properties in objects, such as "personNames.givenName". It uses straight brackets to define the position of a value in an array, such as "favoriteFoods[0]". It supports the last keyword to refer to the last item in the array, such as "favoriteFoods[last]". It also supports referencing elements before the last item in the array, such as "favoriteFoods[last-1]" and "favoriteFoods[last-2]".

FairCom JSON Path does not use "$." to refer to the root object. If you use an online JSON Path evaluator to create JSON Path expressions, such as jsonpath.com, you must add or remove "$." to make them work, such as $.personNames.legalName[2].

{
  "comment": "JSON Paths for each property in this document.",
  "jsonPaths": [
    "comment",
    "jsonPaths",
    "id",
    "type",
    "birthdate",
    "favoriteFoods",
    "favoriteFoods[0]",
    "favoriteFoods[1]",
    "favoriteFoods[2]",
    "favoriteFoods[last]",
    "favoriteFoods[last-1]",
    "favoriteFoods[last-2]",
    "personNames",
    "personNames.givenName",
    "personNames.surname",
    "personNames.nickname",
    "personNames.legalName",
    "personNames.legalName[0]",
    "personNames.legalName[1]",
    "personNames.legalName[2]",
    "emailList",
    "emailList[0]",
    "emailList[0].emailPurpose",
    "emailList[0].emailAddress",
    "emailList[1]",
    "emailList[1].emailPurpose",
    "emailList[1].emailAddress"
  ],
  
  "id": 1,
  "type": "person",
  "birthdate": "2000-01-01",
  "favoriteFoods": [ "pizza", "ice cream", "steak" ],

  "personNames": {
    "givenName": "Michael",
    "surname": "Bowers",
    "nickname": "Mike",
    "legalName": [ "Michael", "Thomas", "Bowers" ]
  },
  
  "emailList": [
    {
      "emailPurpose": "home",
      "emailAddress": "myhome@gmail.com"
    },
    {
      "emailPurpose": "work",
      "emailAddress": "mywork@gmail.com"
    }    
  ]
}

{
  "temperature": 20.1
}    

{
  "humidity":
  {
    "temperature": 20.1
  }
}

{
  "temperature": [20.1 ]
}

{
  "temperature": [ 20.1, 19.3 ]
}

{
  "humidity": 
  [
    {
      "temperature": 20.1
    }
  ]
}

<MTConnectStreams>
  <Streams>
    <DeviceStream 
      name="X1" 
      uuid="X1_373f-4ab9-9c7a-173edd23e4f3">
      <ComponentStream 
        component="Device" 
        name="Stamper1" 
        componentId="X1_Stamper1">
        <Samples>
          <REAL 
            dataItemId="X1_Stamper1_Temperature"
            name="Temperature"
            sequence="5" 
            timestamp="2010-04-06T06:19:35.153141"
            >20.1</REAL>
        </Samples>
      </ComponentStream>
    </DeviceStream>
  </Streams>
</MTConnectStreams>

{
  "api": "hub",
  "action": "createInput",
  "params": {
    "inputName": "mymtconnect",
    "serviceName": "mtconnect",
    "settings": {
      "host": "195.167.1.5:5000",
      "propertyMapList": [
        {
          "propertyPath": "temperature.value",
          "dataType": "string",
          "mtconnectDeviceUuid": "X1_373f-4ab9-9c7a-173edd23e4f3",
          "mtconnectDataItemId": "X1_Stamper1_Temperature",

          "mtconnectCategoryPropertyPath": "temperature.category",
          "mtconnectComponentNamePropertyPath": "temperature.component",
          "mtconnectDataItemIdPropertyPath": "temperature.dataItemId",
          "mtconnectDataNamePropertyPath": "temperature.dataName",
          "mtconnectDataTypePropertyPath": "temperature.dataType",
          "mtconnectDeviceNamePropertyPath": "temperature.device",
          "mtconnectDeviceUuidPropertyPath": "temperature.deviceUuid",
          "mtconnectSequencePropertyPath": "temperature.sequence",
          "mtconnectTimestampPropertyPath": "temperature.timestamp"
        }
      ]
    },
    "tableName": "mtconnectTable"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

{
  "temperature": 
  {
    "value": 20.1,
    "category": "SAMPLE",
    "component": "Stamper1",
    "dataItemId": "X1_Stamper1_Temperature",
    "dataName": "Temperature",
    "dataType": "REAL",
    "device": "Device1",
    "deviceUuid": "X1_373f-4ab9-9c7a-173edd23e4f3",
    "sequence": 5,
    "timestamp": "2010-04-06T06:19:35.153141"
  }
}

{
  "api": "hub",
  "action": "createInput",
  "params": {
    "inputName": "mymtconnect",
    "serviceName": "mtconnect",
    "settings": {
      "host": "195.167.1.5:5000",
      "propertyMapList": [
        {
          "propertyPath": "temperatureValue",
          "dataType": "string",
          "mtconnectDeviceUuid": "X1_373f-4ab9-9c7a-173edd23e4f3",
          "mtconnectDataItemId": "X1_Stamper1_Temperature",

          "mtconnectCategoryPropertyPath": "temperatureCategory",
          "mtconnectComponentNamePropertyPath": "temperatureComponent",
          "mtconnectDataItemIdPropertyPath": "temperatureDataItemId",
          "mtconnectDataNamePropertyPath": "temperatureDataName",
          "mtconnectDataTypePropertyPath": "temperatureDataType",
          "mtconnectDeviceNamePropertyPath": "temperatureDevice",
          "mtconnectDeviceUuidPropertyPath": "temperatureDeviceUuid",
          "mtconnectSequencePropertyPath": "temperatureSequence",
          "mtconnectTimestampPropertyPath": "temperatureTimestamp"
        }
      ]
    },
    "tableName": "mtconnectTable"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

{
  "temperatureValue": 20.1,
  "temperatureCategory": "SAMPLE",
  "temperatureComponent": "Stamper1",
  "temperatureDataItemId": "X1_Stamper1_Temperature",
  "temperatureDataName": "Temperature",
  "temperatureDataType": "REAL",
  "temperatureDevice": "Device1",
  "temperatureDeviceUuid": "X1_373f-4ab9-9c7a-173edd23e4f3",
  "temperatureSequence": 5,
  "temperatureTimestamp": "2010-04-06T06:19:35.153141"
}