Star
Try it now
Solutions
Customers
Use Cases
Partners
About Us
Pricing Docs Guides

ThingsBoard Documentation

Documentation for using ThingsBoard IoT Platform.

AWS Kinesis Integration


ThingsBoard PE Feature


Only ThingsBoard Professional Edition supports Platform Integrations feature.

See ThingsBoard PE Installation Options to install ThingsBoard PE.

Overview

AWS Kinesis provides easily collect, process, and analyze video and data streams in real time, so you can get timely insights and react quickly to new information. After integrating AWS Kinesis with the Thingsboard, you can process and visualize data from AWS Kinesis streams in the Thingsboard IoT platform.

Please make sure that you know AWS Kinesis basics and what are the AWS Kinesis streams in general before continue with this topic.

AWS Kinesis setup

Install and configure AWS CLI

The first step is to obtain AWS Access Keys for your AWS account. Access key to your AWS account must be able to create AWS Kinesis stream, put data to the stream and get data from the stream. Please go to Managing Access Keys for Your AWS Account Root User to create your AWS access key.

Additionally, please make sure that access key for your account has access to AWS Kinesis, AWS DynamoDB and AWS CloudWatch services.

Once the Access Key is created, please note down AWS Access Key ID, AWS Secret Access Key:

Additionally, please make sure that you know your default AWS region name:

We we’ll refer to them later as AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION accordingly during the configuration of AWS Kinesis Thingsboard integration.

The second step is to install and configure AWS CLI to be able to create streams, put records into the streams and get records from the streams from the command line. Go to AWS CLI install and configuration and install AWS CLI onto your machine.

Once completed, please make sure that you are able to see help for command line:

aws kinesis help

and receive the following output:

KINESIS()                                                            KINESIS()

NAME
       kinesis -

DESCRIPTION
       Amazon  Kinesis  Data  Streams is a managed service that scales elasti-
       cally for real-time processing of streaming big data.

AVAILABLE COMMANDS
       o add-tags-to-stream
       o create-stream
       o decrease-stream-retention-period
       o delete-stream
       o deregister-stream-consumer
       o describe-limits
       o describe-stream
       o describe-stream-consumer
       o describe-stream-summary
       o disable-enhanced-monitoring
....

Please click q to close help.

Kinesis stream data format

Kinesis uses Base64 format for data inside streams. In mean time Thingsboard AWS Kinesis integration will automatically convert Base64 encoding into character payload.

If your application sends data as CSV, you will receive the same CSV payload on the Thingsboard converter. If your application sends data as JSON string, Thingsboard converter will receive JSON string in the text payload.

In this tutorial, we will use JSON string to put records into Kinesis data stream. As well we will send data back to Kinesis streams from the Thingsboard in the JSON string. In real life scenario, it is up to you what data format to use to decode/encode data.

AWS Kinesis demo streams

In this demo we will use two AWS Kinesis streams:

Let’s create uplink stream with the help of AWS CLI:

aws kinesis create-stream --stream-name tb-test-uplink --shard-count 1

Please verify that stream was created successfully:

aws kinesis describe-stream --stream-name tb-test-uplink

The output should be similar to this:

{
    "StreamDescription": {
        "KeyId": null, 
        "EncryptionType": "NONE", 
        "StreamStatus": "ACTIVE", 
        "StreamName": "tb-test-uplink", 
        "Shards": [
            {
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": {
                    "EndingHashKey": "340282366920938463463374607431768211455", 
                    "StartingHashKey": "0"
                }, 
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49599874927422395224927130964668873176443726009932447746"
                }
            }
        ], 
        "StreamARN": "arn:aws:kinesis:eu-west-1:XXXXXXXXXX:stream/tb-test-uplink", 
        "EnhancedMonitoring": [
            {
                "ShardLevelMetrics": []
            }
        ], 
        "StreamCreationTimestamp": 1569503664.0, 
        "RetentionPeriodHours": 24
    }
}

Next step is to create downlink stream:

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

Please verify that stream was created successfully:

aws kinesis describe-stream --stream-name tb-test-downlink

The output should be similar to this:

{
    "StreamDescription": {
        "KeyId": null, 
        "EncryptionType": "NONE", 
        "StreamStatus": "ACTIVE", 
        "StreamName": "tb-test-downlink", 
        "Shards": [
            {
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": {
                    "EndingHashKey": "340282366920938463463374607431768211455", 
                    "StartingHashKey": "0"
                }, 
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49599874927422395224927130964668873176443726009932447746"
                }
            }
        ], 
        "StreamARN": "arn:aws:kinesis:eu-west-1:XXXXXXXXXX:stream/tb-test-downlink", 
        "EnhancedMonitoring": [
            {
                "ShardLevelMetrics": []
            }
        ], 
        "StreamCreationTimestamp": 1569503664.0, 
        "RetentionPeriodHours": 24
    }
}

Integration with the Thingsboard

We have done all necessary steps on the AWS Kinesis side. Now we can start configuring the Thingsboard.

First, we need to create Uplink Data converter that will be used for converting messages received from the AWS Kinesis. The converter should transform incoming payload into the required message format. Message must contains deviceName and deviceType. Those fields are used for submitting data to the correct device. If a device was not found then new device will be created. Here is how demo payload from the AWS Kinesis will look like:

{
    "devName": "kitchen_thermostat", 
    "devType": "thermostat",
    "temperature": 22
}

We will take devName and map it to the deviceName and devType map to the deviceType. But you can use another mapping in your specific use cases. Also, we will take the value of the temperature field and use it as a device telemetry.

Go to Data Converters and create new uplink Converter with this function:

var data = decodeToJson(payload);
var deviceName = data.devName;
var deviceType = data.devType;

var result = {
    deviceName: deviceName,
    deviceType: deviceType,
    telemetry: {
         temperature: data.temperature
    }
};

function decodeToString(payload) {
    return String.fromCharCode.apply(String, payload);
}

function decodeToJson(payload) {
    var str = decodeToString(payload);
    var data = JSON.parse(str);
    return data;
}

return result;

image

For sending Downlink messages from the Thingsboard to the Kinesis stream, we need to define downlink Converter. In general, output from Downlink converter should have the following structure:

{
    "contentType": "JSON",
    "data": "{\"devName\":\"kitchen_thermostat\",\"version\":\"0.11\"}",
    "metadata": {
        "streamName": "tb-test-downlink",
        "partitionKey": "1234"
    }
}

Go to Data Converters and create new downlink Converter with this function:

var data = {
    devName: "kitchen_thermostat",
    version: msg.version
};

var result = {
    contentType: "JSON",
    data: JSON.stringify(data),
    metadata: {
        streamName: "tb-test-downlink",
        partitionKey: "123",
    }

};

return result;

This converter will take version field from the incoming message and add it as a payload field in the outbound message. Additionally, we will put device name in the outbound message. In this way next applications in our business flow, that will consume messages from the donwlink stream, will be able to identify, that this payload belongs to kitchen_thermostat device.

image

AWS Kinesis Integration

Next we will create Integration with AWS Kinesis inside the Thingsboard. Open Integrations section and add new Integration with type AWS Kinesis

image

image

Validation

Lets verify our integration. First, lets put message into uplink stream, so Thingsboard will fetch this message. Type in the console:

aws kinesis put-record --stream-name tb-test-uplink --partition-key 123 --data '{"devName": "kitchen_thermostat", "devType": "thermostat", "temperature": 22}'

Go to Device Group -> All -> kitchen_thermostat - you can see that

image

For testing Downlink Messages, we will update our Root Rule Chain to send downlink message when device attribute is changed. Open and edit Root Rule Chain. Add Integration Downlink Action node and connect it with the Message Type Switch Node using relation Attributes Updated

image

image

Save Changes.

Go to Device Group -> All -> kitchen_thermostat -> attributes section. We will add Shared attribute with name version and value v.0.11

image

By making this step, we triggered downlink message to the downlink stream tb-test-downlink and this message should contains device name and version field value. Let’s use command line AWS CLI tool to validate, that we have received the massage in the downlink stream. First, we need to get shard iterator for the downlink stream, to get records from the stream:

aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name tb-test-downlink

In the output you will receive ID of the iterator:

{
    "ShardIterator": "AAAAAAAAAAHHpat8cWmSZ1LSNg2oOra2yaRN/75ZhdwXFNZjnznXU1BQVjo9B+IVi3Pk6ltKN3V8Vhihi74MtXvG9rKcL9VRP1xRebo/NVjaMWd5FmAcp1rvs/DOahY+0FVm6UC1fKFwUXmgnsCzxkU9V8G9cFyGU/4Vvpf2PcrGsHG/JtNouCYT4wwdoXZ6FIFTuWdW7/Qs48PAZCkSjluwA7K1pld4"
}

Now, let’s use the ShardIterator from the above JSON in the get-records command:

aws kinesis get-records --shard-iterator **ShardIterator**

In our example:

aws kinesis get-records --shard-iterator AAAAAAAAAAFQtL3oAo74irn+ccC3vghADqqmh2MH+HKI9qYTi1NP957vDe8KyV6VdQ+I4shIP0HIRRVYyTZs0W9v6jaai9LevlJayMw6TgdPkVIGmV5SYZF8sGWgtd0wJuRqB+6QwCAUHQ52dgT4m+lypNSzJJw4Mo6h+9Wdk5fpwQxu/GlM8J+Uqblnq4EEr17FkWLahikaSZXktfLq5dh23+LEIc22

In the output you should receive records from the donwlink stream in the JSON format:

{
    "Records": [
        {
            "Data": "eyJkZXZOYW1lIjoia2l0Y2hlbl90aGVybW9zdGF0IiwidmVyc2lvbiI6InYuMC4xMSJ9", 
            "PartitionKey": "1234", 
            "ApproximateArrivalTimestamp": 1569609612.27, 
            "SequenceNumber": "49599912710236940383450082324919185009278025474345271298"
        }
    ], 
    "NextShardIterator": "AAAAAAAAAAFQlgSyxBdpKxlRrocJCYT9YDrCi/vxl0sstJgg4CM+pttVsK4AjjQwJ/QJsags5vdpQdopaqk9aKefAUOWobgwHVaZvhI4tdkmHBr45uO0Hq9AxUlKDxfiYbM0qgN33+5SvGxU8gJBUihYFY4ydPWOWdVTf2lOxp0a9X6DFrjsUqwMXR9skLw8/lQkBmHVFBlFURPy+z/AMuYHga5mDch/", 
    "MillisBehindLatest": 0
}

As Kinesis uses Base64 for data presentation, we need to use some online Base64 converter tool to encode from the Base64 to JSON representative. As a result we will get next JSON payload:

{
    "devName":"kitchen_thermostat",
    "version":"v.0.11"
}

Now other application is able to listen to this downlink stream and react accordingly to your business logic.

See also

With this integration you can also configure Downlink converters and trigger required actions using Rule Engine nodes.

Next steps