jq Functions

jq is a powerful command line tool and programming language designed primarily for transforming and querying data encoded as JSON.

EMQX rules allow users to define SQL-like rules to process and route messages. These rules can include jq functions to perform complex transformations on JSON payloads as they pass through the broker.

If you are new to jq function, you can refer to the Reference section to speedily get started.

TIP

jq functions can be convenient for transformations that are difficult to do with only the rule SQL language.

However, to maintain efficient message processing, you are recommended to avoid long-running computations in the rule and use the timeouts feature (with configuration item rule_engine.jq_function_default_timeout) to prevent buggy jq programs.

The jq Function

The basic format of a jq statement in the rule engine SQL is as follows:

jq('<JQ_program>', '<JSON_input>', <timeout_value>)

Where,

1. <JQ_program>: A string containing a valid jq program.
2. <JSON_input>: A JSON encoded string or object as the input for the jq program.
3. <timeout_value>: An optional integer timeout value in milliseconds, with a default value of 10 seconds.''

The jq function returns a list of objects generated by the given jq program when provided with the specified input. If the execution doesn't finish before the timeout or if the jq program encounters an exception, the function will throw an error.

Use Cases

The following are some examples of simple jq function calls and their results:

JSON Data Manipulation

This example illustrates various ways to manipulate JSON data using jq, including accessing, transforming, and calculating values within JSON objects,

Code Example:

jq('.', '{"temperature": 10}') =
[json_decode('{"temperature": 10}')]

jq('.', json_decode('{"temperature": 10}')) =
[json_decode('{"temperature": 10}')]

jq('.temperature', '{"temperature": 10}') =
[10]

jq('{temperature_C:.temperature,
     temperature_F: (.temperature * 1.8 + 32)}',
   '{"temperature": 10}') =
[json_decode('{"temperature_C": 10, "temperature_F": 50}')]

jq('.temperature,(.temperature * 1.8 + 32)', '{"temperature": 10}') =
[10, 50]

Calculate Averages by Removing Outliers

For example, the below JSON object contains a date and an array of sensors, each with a name and a set of data points, representing sensor readings on a specific date.

{
  "date": "2020-04-24",
  "sensors": [
    {
      "name": "a",
      "data": [3, 1, 2, 4, 5, 5]
    },
    {
      "name": "b",
      "data": [1, -100, 2, 3, 4, 5, 2000]
    },
    {
      "name": "c",
      "data": [3, 7, 9]
    }
  ]
}

You can combine the jq function with the FOREACH statement to divide jq's output objects into multiple messages, with each containing one field for the date and one field for the average of the sensor's data field after removing outliers.

FOREACH   jq('def rem_first:
                 if length > 2 then del(.[0]) else . end;
              def rem_last:
                 if length > 1 then del(.[-1]) else . end;
              .date as $date |
              .sensors[] |
                (.data | sort | rem_first | rem_last | add / length) as $average |
                {$average, $date}',
             payload)
FROM    "jq_demo/complex_rule/jq/#"

Then the three output messages will have the following payloads:

Message 1:

{
  "average": 3.5,
  "date": "2020-04-24"
}

Message 2:

{
  "average": 3,
  "date": "2020-04-24"
}

Message 3:

{
  "average": 7,
  "date": "2020-04-24"
}

Split One Messsage into Separate Messages

The example code processes an input message containing multiple sensor measurements and splits it into separate messages for each sensor type. This is how it works:

• The FOREACH uses the jq function to transform the input message into an array of objects containing sensor_type and value fields.
• The DO clause selects relevant fields for the output messages.
• The FROM clause applies the rule to messages with a matching topic filter, car/measurements.

FOREACH
    ## The data must be an array
    jq('
       [{
         sensor_type: "temperature",
         value: .temperature
        },
        {
         sensor_type: "humidity",
         value: .humidity
        },
        {
         sensor_type: "pressure",
         value: .pressure
        },
        {
         sensor_type: "light",
         value: .light
        },
        {
         sensor_type: "battery",
         value: .battery
        },
        {
         sensor_type: "speed",
         value: .speed
        }]',
        payload) as sensor  
DO
    payload.client_id,
    payload.timestamp,
    sensor.sensor_type,
    sensor.value
FROM "car/measurements"

Alternative Way for Splitting Messages

This example illustrates an alternative approach to splitting an input message containing multiple sensor measurements into separate messages for each sensor type.

The jq function within the FOREACH clause saves the input and all sensor types, then outputs an object for each sensor type with relevant fields.

FOREACH
    jq('
       # Save the input
       . as $payload |
       
       # All sensor types
       [ 
         "temperature",
         "humidity",
         "pressure",
         "light",
         "battery",
         "speed" 
       ] as $sensor_types |
       
       # Output an object for each sensor type
       $sensor_types[] |
       {
         client_id: $payload.client_id,
         timestamp: $payload.timestamp,
         sensor_type: .,
         value: $payload[.] 
       }
       ',
       payload) as sensor  
FROM "car/measurements"

References

If you are new to the jq function, the following materials are recommended:

• jq documentation
• Online jq programming playground
• EMQX's jq introduction video