AWS Kinesis Integration

AWS Kinesis Integration connects ThingsBoard to Amazon Kinesis Data Streams. ThingsBoard reads every shard of a stream with the Kinesis Client Library (KCL). An uplink data converter transforms each record into device or asset data, including telemetry and attributes. ThingsBoard can also write downlink records to a Kinesis stream.

Use AWS Kinesis Integration when:

• Devices, AWS IoT rules, or backend services already write data to a Kinesis data stream.
• One stream carries data for many devices: the converter resolves each device from the record.
• You need to send commands or events back to AWS through a stream.

1. Receive data. A producer writes a record to the stream. The integration reads each shard and stores its read position (checkpoint) in an Amazon DynamoDB table.
2. Decode the payload. The uplink converter receives the record data and its partition key, identifies the device or asset, and extracts attributes and telemetry.
3. Process the message. ThingsBoard processes the converted data through the Rule Engine. If Allow create devices or assets is enabled, ThingsBoard creates an entity that does not exist yet.
4. Send downlink data (optional). The Rule Engine passes a message to the integration. The downlink converter encodes it and names the target stream and partition key. The integration writes the record with the Kinesis Producer Library (KPL).

• Access to ThingsBoard Cloud with integrations enabled for your tenant.
• Permission to create integrations and data converters, and to edit rule chains for downlink.
• An AWS account on a paid plan, with permission to create Kinesis data streams, IAM users, and IAM policies. Kinesis Data Streams is not available on the AWS Free plan.
• The AWS CLI version 2, configured with aws configure for the Region you use, or AWS CloudShell in the AWS console, where the CLI is already configured for the console Region. This guide uses the CLI to create the streams and send test records.

Create two streams and an IAM user whose access key the integration uses.

This guide uses the stream my-thingsboard for uplink and tb-test-downlink for downlink. Create both in the same Region:

aws kinesis create-stream --stream-name my-thingsboard --shard-count 1
aws kinesis create-stream --stream-name tb-test-downlink --shard-count 1

On success, the commands print nothing. With --shard-count, each stream uses provisioned capacity with one shard.

Wait until both streams are active. Each command returns without output when StreamStatus is ACTIVE, usually within a minute:

aws kinesis wait stream-exists --stream-name my-thingsboard
aws kinesis wait stream-exists --stream-name tb-test-downlink

To check a stream, read its status, capacity mode, and shard count:

aws kinesis describe-stream-summary --stream-name my-thingsboard \
  --query 'StreamDescriptionSummary.[StreamStatus,StreamModeDetails.StreamMode,OpenShardCount]'

The output lists ACTIVE, PROVISIONED, and 1.

Besides Kinesis, the integration uses Amazon DynamoDB and Amazon CloudWatch. KCL creates three DynamoDB tables named after the Application name of the integration (by default, the stream name): a lease table with checkpoints, <name>-CoordinatorState, and <name>-WorkerMetricStats. KCL and KPL also publish metrics to CloudWatch.

1. Create the user. In the AWS console, open IAM ⇾ IAM users and click Create user.

   • Specify user details — enter a name, for example thingsboard-kinesis. Leave Provide user access to the AWS Management Console off: the integration needs only an access key.
   • Set permissions — select no group and no policy. The user gets only the policy from step 3.
   • Review and create — click Create user.
   Enter the user name and leave console access off. Select no group and no policy. The permissions summary is empty. Click Create user.

2. The page of the new user opens. On the Permissions tab, click Add permissions ⇾ Create inline policy.

   Click Add permissions, then Create inline policy.

3. In the Policy editor, switch to JSON and replace the content with the policy below. Replace REGION and ACCOUNT_ID with your Region and AWS account ID. Click Next, enter a Policy name, for example thingsboard-kinesis-policy, and click Create policy.

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "ReadUplinkStream",
         "Effect": "Allow",
         "Action": [
           "kinesis:DescribeStream",
           "kinesis:DescribeStreamSummary",
           "kinesis:GetRecords",
           "kinesis:GetShardIterator",
           "kinesis:ListShards"
         ],
         "Resource": "arn:aws:kinesis:REGION:ACCOUNT_ID:stream/my-thingsboard"
       },
   

   ▼Expand

   Paste the policy. The editor reports no errors or warnings. Name the policy and check that it covers CloudWatch, DynamoDB, and Kinesis. The policy is created. Next, open the Security credentials tab and click Create access key.

4. Open the Security credentials tab and click Create access key. Select Application running outside AWS and click Next. AWS recommends temporary credentials instead; the integration needs an access key, so continue.

   Select Application running outside AWS.

5. Optionally, enter a Description tag value, for example ThingsBoard Kinesis integration, and click Create access key.

   Optionally, describe the key.

6. On Retrieve access keys, copy the Access key and the Secret access key, or click Download .csv file. AWS shows the secret only once. Then click Done.

   Copy both keys or download the .csv file. The user has one active access key.

Create the uplink converter, assign it to a new AWS Kinesis integration, and write a test record.

The uplink converter receives the data of each record as a byte array, resolves the target device, and returns its telemetry and attributes. AWS Kinesis Integration uses the Generic converter: the decoder is a TBEL or JavaScript script. For converter types and the full decoder reference, see Uplink data converter.

Kinesis does not define a record format. This guide uses the sample record that the Test decoder function dialog provides by default:

{
    "deviceName": "Thermometer",
    "deviceType": "thermostat",
    "serialNumber": "SN-12345",
    "temperature": 25.5
}

1. Go to Integrations ⇾ Data converters and click + Add data converter ⇾ Create new converter.

2. Converter type — keep Uplink.

3. Integration type — select AWS Kinesis.

4. Name — enter a name, for example AWS Kinesis Uplink Converter.

5. Main decoding configuration — the editor opens with the default decoder below. It already handles this record format, so keep it. It takes deviceName and deviceType from the record, the time from ts (milliseconds) or timestamp (ISO 8601), stores the keys in attributeKeyList (here serialNumber) as attributes, and stores all other fields as telemetry. When the record has no time field, it uses the server time. The editor opens in TBEL; use the TBEL / JS toggle to switch to JavaScript.

   • TBEL (Recommended)
   • JavaScript

   // Decode an uplink message from a buffer
   // payload - array of bytes
   // metadata - key/value object
   
   /** Decoder **/
   
   // decode payload to JSON
   var data = decodeToJson(payload);
   
   // --- Device name and type ---
   var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
   var deviceType = data.deviceType != null ? data.deviceType : 'default';
   // var customerName = 'Customer C';
   // var groupName = 'thermostat devices';
   
   

   ▼Expand

   To adapt:

   • Device name in another field — replace data.deviceName and data.deviceType, for example with data.deviceId.
   • Device name from the partition key — if producers use the device ID as the partition key, use metadata.partitionKey.
   • Other device properties — add their keys to attributeKeyList, for example firmwareVersion.
   • Assets instead of devices — uncomment assetName and assetType and use them instead of deviceName and deviceType.
   • Customer or group — uncomment customerName or groupName to assign the device to a customer or an entity group.

6. Click Test decoder function and check the result (see Test the Decoder).

7. Click Add.

Decoder Inputs

The decoder signature is function decoder(payload, metadata): object | object[].

For field renaming, nested values, and non-JSON payloads, see Common scripting patterns.

Test the Decoder

Click Test decoder function below the editor. The Payload already holds the sample record. Click Test. The Output panel shows:

{
    "deviceName": "Thermometer",
    "deviceType": "thermostat",
    "attributes": {
        "serialNumber": "SN-12345"
    },
    "telemetry": {
        "ts": 1792072084561,
        "values": {
            "temperature": 25.5
        }
    }
}

The record has no time field, so ts is the time of the test.

1. Go to Integrations ⇾ Integrations and click + Add integration.
2. Basic settings:
   • Integration type — select AWS Kinesis.
   • Name — enter a name, for example AWS Kinesis integration.
   • Enable integration and Allow create devices or assets — keep both on, so the device is created on its first record.
   • Click Next.
3. Uplink data converter:
   • Click Select existing and choose AWS Kinesis Uplink Converter.
   • Alternatively, click Create new to write a decoder here.
   • Click Next.
4. Downlink data converter:
   • Click Skip. Add a downlink converter later if you need to write records to a stream.
5. Connection:
   • Stream name — my-thingsboard.
   • Region — the code of the Region where you created the streams, for example eu-north-1.
   • Initial position in stream — select Latest.
   • Access Key Id and Secret Access Key — paste the access key of the IAM user.
   • Leave the other parameters at their defaults. See Connection Settings for details.
6. Click Add.

Advanced settings

Write a record to the stream, then confirm that ThingsBoard decoded it and created the device.

Run aws kinesis put-record. The --cli-binary-format raw-in-base64-out option makes AWS CLI version 2 send --data as plain text:

aws kinesis put-record \
  --stream-name my-thingsboard \
  --partition-key Thermometer \
  --cli-binary-format raw-in-base64-out \
  --data '{"deviceName":"Thermometer","deviceType":"thermostat","serialNumber":"SN-12345","temperature":25.5}'

The command prints the ShardId and SequenceNumber of the new record. The integration reads it within a few seconds.

put-record returns the shard and the sequence number of the record.

With Latest, the integration skips records written before it started reading. Write the record after the integration is created. If no uplink arrives, write the record again.

• Integration events — go to Integrations ⇾ Integrations, open the integration, and click the Events tab. The uplink appears with status OK. Click ⋅⋅⋅ in the Message column to see the shard ID, sequence number, partition key, and payload.
• Converter events — open AWS Kinesis Uplink Converter in Integrations ⇾ Data converters and click the Events tab. In shows the record data, Out shows the decoded result, and Metadata shows the partition key, sequence number, and integration name.
• Device — go to Devices & assets ⇾ Devices. The integration has created Thermometer with type thermostat. Open it and check temperature = 25.5 on the Latest telemetry tab and serialNumber on the Attributes tab.

ThingsBoard writes a downlink record when the Rule Engine routes a message to the integration — for example, after a shared attribute update. The target stream and partition key come from streamName and partitionKey in the downlink converter’s output metadata. The stream must be in the integration’s Region.

The downlink converter turns a Rule Engine message into a Kinesis record. For the full encoder reference, see Downlink data converter.

Add a downlink converter to the integration. The encoder returns:

1. Open the integration in Integrations ⇾ Integrations and click Toggle edit mode (pencil icon).
2. In the Downlink data converter field, click Create new. The Downlink converter type and AWS Kinesis integration type are preselected.
3. Enter a name, for example AWS Kinesis Downlink Converter.
4. In Main encoding configuration, replace the default encoder with the one below.
5. To record downlink events, click the debug icon next to the name, select All messages, and click Apply.
6. Click Add, then Apply changes.

Encoder used in this guide (TBEL and JavaScript):

/** Encoder **/

// Write the Rule Engine message data as the record data.
var result = {
    contentType: "JSON",
    data: JSON.stringify(msg),
    metadata: {
        // Target stream.
        streamName: "tb-test-downlink",
        // One partition key per device keeps its records in order.
        // The originator fields node adds deviceName to the metadata.
        partitionKey: metadata.deviceName
    }
};

return result;

To adapt:

• Other stream — change streamName. The IAM policy must allow kinesis:PutRecords on that stream.
• Selected fields — build data from the fields you need, for example JSON.stringify({ interval: msg.temperatureUploadFrequency }).
• Other message types — branch on msgType, for example RPC_CALL_FROM_SERVER_TO_DEVICE.

Attribute update messages carry no device name in their metadata. An originator fields node adds it, so the encoder can use it as the partition key.

1. Go to Data processing ⇾ Rule chains and open the Root Rule Chain.
2. From the Enrichment section, drag an originator fields node onto the canvas. Enter a Name (for example, Add device name). Under Originator fields mapping, map Name to the target key deviceName. Keep Add mapped originator fields to set to Metadata, and click Add.
3. From the Action section, drag an integration downlink node onto the canvas. Enter a Name (for example, Downlink to AWS Kinesis), select your AWS Kinesis integration, and click Add.
4. Connect the message type switch node to the originator fields node with the Attributes Updated relation.
5. Connect the originator fields node to the integration downlink node with the Success relation.
6. Click Apply changes.

1. Go to Devices & assets ⇾ Devices, open Thermometer, and select the Attributes tab.
2. Select Shared attributes and click +.
3. Set Key to temperatureUploadFrequency, Type to Integer, Value to 60, and click Add. ThingsBoard writes the downlink record.

• Converter events — open the downlink converter’s Events tab. In shows the Rule Engine message, and Out shows the encoded data with streamName and partitionKey = Thermometer in metadata.

• Integration events — the integration’s Events tab shows a Downlink event with the stream name, partition key, and payload.

• AWS — read the downlink stream from its oldest record. The command gets a shard iterator and passes it to get-records:

  ITER=$(aws kinesis get-shard-iterator \
    --stream-name tb-test-downlink \
    --shard-id shardId-000000000000 \
    --shard-iterator-type TRIM_HORIZON \
    --query ShardIterator --output text)
  aws kinesis get-records --shard-iterator "$ITER"

  A shard iterator expires after 5 minutes. If Records is empty, run aws kinesis get-records --shard-iterator with the returned NextShardIterator until the record appears. The record has PartitionKey = Thermometer and Base64-encoded Data:

  {
    "Records": [
      {
        "SequenceNumber": "49679318713549456377638275380422332354075930242739863554",
        "ApproximateArrivalTimestamp": "2026-10-16T14:28:39.979000+00:00",
        "Data": "eyJ0ZW1wZXJhdHVyZVVwbG9hZEZyZXF1ZW5jeSI6NjB9",
        "PartitionKey": "Thermometer"
      }
    ],
    "MillisBehindLatest": 0
  }

  Decode Data to get the payload {"temperatureUploadFrequency":60}:

  echo 'eyJ0ZW1wZXJhdHVyZVVwbG9hZEZyZXF1ZW5jeSI6NjB9' | base64 --decode

Delete the test resources when you finish, because AWS bills the streams for every shard-hour.

1. In ThingsBoard, turn off Enable integration for the integration, or delete it. A running integration creates the DynamoDB tables again.

2. Delete the streams and the DynamoDB tables. Use your Application name instead of my-thingsboard in the table names if you set one.

   aws kinesis delete-stream --stream-name my-thingsboard
   aws kinesis delete-stream --stream-name tb-test-downlink
   aws dynamodb delete-table --table-name my-thingsboard
   aws dynamodb delete-table --table-name my-thingsboard-CoordinatorState
   aws dynamodb delete-table --table-name my-thingsboard-WorkerMetricStats

3. Delete the IAM user. Its access keys and inline policy go first:

   for k in $(aws iam list-access-keys --user-name thingsboard-kinesis \
       --query 'AccessKeyMetadata[].AccessKeyId' --output text); do
     aws iam delete-access-key --user-name thingsboard-kinesis --access-key-id "$k"
   done
   aws iam delete-user-policy --user-name thingsboard-kinesis --policy-name thingsboard-kinesis-policy
   aws iam delete-user --user-name thingsboard-kinesis

4. Check that nothing is left. list-streams and list-tables print [], and get-user fails with NoSuchEntity:

   aws kinesis list-streams --query 'StreamNames'
   aws dynamodb list-tables --query 'TableNames'
   aws iam get-user --user-name thingsboard-kinesis

To read debug events, open the integration or converter, click Events, and select an event: In is the received data, Out is the converter result, and Error holds the error text.

• Integrations overview — how integrations, converters, uplinks, and downlinks fit together.
• Uplink data converter — decoder inputs, output format, and scripting patterns.
• Downlink data converter — encoder reference.
• AWS SQS Integration — read messages from an Amazon SQS queue.
• Remote Integration — run the integration outside the ThingsBoard cluster.
• Integration downlink node — forward Rule Engine messages to an integration.
• TBEL — built-in functions and syntax for converter scripts.
• Kinesis Client Library — leases, checkpoints, and DynamoDB tables.