Ingesting enriched IoT knowledge into Amazon S3 utilizing Amazon Kinesis Information Firehose

Introduction

When sending knowledge from Web of Issues (IoT) units to a knowledge lake, you could want to complement the system knowledge payload with extra metadata within the cloud for additional knowledge processing and visualization. There are a number of causes this knowledge won’t exist within the system payload, corresponding to minimizing the system payload in restricted bandwidth environments or modifying it with enterprise inputs within the cloud. For instance, a machine on the manufacturing facility ground could be assigned to totally different operators in the course of the day. This variable enterprise knowledge could be saved in a database. In your knowledge lake, you may want this info to be saved together with the payload.

On this weblog submit, you’ll discover ways to ingest enriched IoT knowledge to a knowledge lake in close to real-time.

Conditions

• An AWS account
• AWS Command Line Interface (AWS CLI). See AWS CLI fast setup for configuration.

Use case definition

Let’s assume that in your logistics firm, you will have containers geared up with sensor-enabled IoT units. When the container is loaded right into a ship, the container ID is related to the ship ID. You have to retailer the IoT system payload with the ship ID in your knowledge lake.

In such a use case, the sensor payload comes from the IoT system connected to the container. Nevertheless, the related ship ID is just saved within the metadata retailer. Due to this fact, the payload have to be enriched with the ship ID earlier than placing it into the information lake.

Answer structure

Within the structure diagram,

1. The IoT units stream payloads to the AWS IoT Core message dealer to a selected MQTT matter system/knowledge/DEVICE_ID. The AWS IoT Core message dealer permits units to publish and subscribe to messages through the use of supported protocols.
2. The AWS IoT rule is triggered when there’s a payload in its matter. It’s configured with an Amazon Kinesis Information Firehose motion on this use case. You need to use AWS IoT guidelines to work together with AWS providers by calling them when there’s a message in a selected MQTT matter or straight through the use of Fundamental Ingest characteristic.
3. Amazon Kinesis Information Firehose buffers the system payloads earlier than delivering them to the information retailer based mostly on the dimensions or the time, whichever occurs first. Kinesis Information Firehose delivers real-time streaming knowledge to locations for storing or processing.
4. As soon as the buffer hits the dimensions or the time threshold, Kinesis Information Firehose calls an AWS Lambda operate to complement the system payloads in batches with the metadata retrieved from an Amazon DynamoDB AWS Lambda is a serverless compute service that runs your code for any kind of software. Amazon DynamoDB is a totally managed NoSQL database that gives quick efficiency.
5. The enriched payloads are returned again to Kinesis Information Firehose to ship to the vacation spot.
6. The enriched payloads are put into an Amazon Easy Storage Service (Amazon S3) bucket as a vacation spot. Amazon S3 is an object storage service which shops any quantity of knowledge for a spread of use circumstances.

AWS CloudFormation template

Obtain the AWS Cloudformation template from the code repository.

The AWS CloudFormation template deploys all the mandatory sources to run this instance use case. Let’s have a more in-depth have a look at AWS IoT guidelines, Kinesis Information Firehose, and AWS Lambda operate sources.

AWS IoT guidelines useful resource

IoTToFirehoseRule:
  Kind: AWS::IoT::TopicRule
  Properties:
    TopicRulePayload:
      Actions:
        -
          Firehose:
            RoleArn: !GetAtt IoTFirehosePutRecordRole.Arn
            DeliveryStreamName: !Ref FirehoseDeliveryStream
            Separator: "n"
      AwsIotSqlVersion: ‘2016-03-23’
      Description: This rule logs IoT payloads to S3 Bucket by aggregating in Kinesis Firehose.
      RuleDisabled: false
      Sql: !Ref IotKinesisRuleSQL

The AWS IoT rule takes a SQL parameter which defines the IoT matter to set off the rule and knowledge to extract from the payload.

• Within the instance, the SQL parameter is about to SELECT *, matter(3) as containerId FROM ‘system/knowledge/+’ by default. SELECT * means the entire payload is taken as it’s and containerId is generated from the second merchandise within the MQTT matter and included to the payload.
• FROM ‘system/knowledge/+’ describes the IoT matter that may set off the AWS IoT rule. + is a wildcard character for MQTT matters and the IoT units will publish knowledge payloads to system/knowledge/DEVICE_ID matter to set off this rule.

The AWS IoT rule additionally defines actions. Within the instance, you’ll be able to see a Kinesis Information Firehose motion which defines the goal Kinesis Information Firehose supply stream and the AWS Identification and Entry Administration (IAM) position wanted to place information into this supply stream. A separator may be chosen to separate every file, within the given instance it’s a new line character.

Kinesis Information Firehose supply stream useful resource

FirehoseDeliveryStream:
  Kind: AWS::KinesisFirehose::DeliveryStream
  Properties:
    ExtendedS3DestinationConfiguration:
      BucketARN: !GetAtt IoTLogBucket.Arn
      BufferingHints:
        IntervalInSeconds: 60
        SizeInMBs: 1
      Prefix: device-data/
      RoleARN: !GetAtt FirehosePutS3Role.Arn
      ProcessingConfiguration:
        Enabled: true
        Processors:
          - Kind: Lambda
             Parameters:
               - ParameterName: LambdaArn
                  ParameterValue: !Sub '${FirehoseTransformLambda.Arn}:$LATEST'
               - ParameterName: RoleArn
                  ParameterValue: !GetAtt FirehoseLambdaInvokeRole.Arn

Kinesis Information Firehose supply stream should outline a vacation spot to place the stream into. It helps various kinds of locations. You will discover the accessible vacation spot sorts and their utilization on this documentation. On this instance, you’re going to use Amazon S3 because the vacation spot.

The instance Supply Stream useful resource defines the next properties:

• BucketARN: the vacation spot bucket which can retailer the aggregated knowledge. The vacation spot bucket is created by the CloudFormation stack.
• BufferingHints: the dimensions and time threshold for knowledge buffering. On this instance, they’re set to 1MB and 60 seconds respectively to see the outcomes quicker. It may be adjusted in keeping with the enterprise wants. Conserving these thresholds low will trigger the Lambda operate to be invoked extra ceaselessly. If the thresholds are excessive, the information will likely be ingested to the information retailer much less ceaselessly, subsequently, it is going to take time to see the newest knowledge within the knowledge retailer.
• Prefix: the created objects will likely be put beneath this prefix. Kinesis Information Firehose partitions the information based mostly on the timestamp by default. On this instance, the objects will likely be put beneath the device-data/YYYY/MM/dd/HH folder. Kinesis Information Firehose has superior options for knowledge partitioning corresponding to dynamic partitioning. The partitioning of the information is essential when querying the information lake. For instance, if you want to question the information per system foundation through the use of Amazon Athena, scanning solely the partition of the related system ID will considerably scale back the scan time and the price. You will discover particulars on partitioning on this documentation.
• RoleARN: that is the IAM position that provides PutObject permission to Kinesis Information Firehose to have the ability to put aggregated knowledge into the Amazon S3 bucket.
• ProcessingConfiguration: As described within the use case, a rework Lambda operate will enrich the IoT knowledge with the metadata. Processing Configuration defines the processor which is a Lambda operate within the instance. For every batch of knowledge, Kinesis Information Firehose will name this Lambda operate for the transformation of the information. You possibly can learn extra about knowledge processing on this documentation.

Transformation Lambda Operate

As you’ll be able to see within the following Python code, Kinesis Information Firehose returns a batch of information the place every file is a payload from the IoT units. First, the base64 encoded payload knowledge is decoded. Then, the corresponding ship ID comes from the DynamoDB desk based mostly on the container ID. The payload is enriched with the ship ID and encoded again to base64. Lastly, the file checklist is returned again to Kinesis Information Firehose.

As soon as Kinesis Information Firehose receives the information, it places them as an aggregated file into the Amazon S3 bucket.

import os
import boto3
import json
import base64

dynamodb = boto3.useful resource('dynamodb')
desk = dynamodb.Desk(os.environ['METADATA_TABLE'])
information = []

def function_handler(occasion, context):
  for file in occasion["records"]:
    # Get knowledge subject of the file in json format. It's a base64 encoded string.
    json_data = json.hundreds(base64.b64decode(file["data"]))
    container_id = json_data["containerId"]

    # Get corresponding shipId from the DynamoDB desk
    res = desk.get_item(Key={'containerId': container_id})
    ddb_item = res["Item"]
    ship_id = ddb_item["shipId"]

    # Append shipId to the precise file knowledge
    enriched_data = json_data
    enriched_data["shipId"] = ship_id

    # Encode the enriched file to base64
    json_string = json.dumps(enriched_data).encode("ascii")
    b64_encoded_data = base64.b64encode(json_string).decode("ascii")

    # Create a file with enriched knowledge and return again to Firehose
    rec = {'recordId': file["recordId"], 'consequence': 'Okay', 'knowledge': b64_encoded_data}
    information.append(rec)
  return {'information': information}

Deployment

Run the next command in a terminal to deploy the stack.

aws cloudformation deploy --stack-name IoTKinesisDataPath --template-file IoTKinesisDataPath.yml --parameter-overrides IotKinesisRuleSQL="SELECT *, matter(3) as containerId FROM 'system/knowledge/+'" --capabilities CAPABILITY_NAMED_IAM

After the deployment is full, run the next command in a terminal to see the output of the deployment.

aws cloudformation describe-stacks --stack-name IoTKinesisDataPath

Word the IoTLogS3BucketName, MetadataTableName output parameters.

Testing

After the deployment is full, very first thing you want to do is to create a metadata merchandise for knowledge enrichment. Run the next command to create an merchandise within the DynamoDB desk. It is going to create an merchandise with cont1 as containerId and ship1 as shipId. Substitute IoTKinesisDataPath-MetadataTable-SAMPLE parameter with the DynamoDB desk output parameter from the CloudFormation stack deployment.

aws dynamodb put-item --table-name IoTKinesisDataPath-MetadataTable-SAMPLE --item '{"containerId":{"S":"cont1"},"shipId":{"S":"ship1"}}'

In a real-life situation, the units publish the payloads to a selected MQTT matter. On this instance, as a substitute of making IoT units, you’ll use AWS CLI to publish payloads to MQTT matters. Run the next command in a terminal to publish a pattern knowledge payload AWS IoT Core. Take note of the payload subject of the command, the one knowledge supplied by the system is the dynamic knowledge.

aws iot-data publish --topic "system/knowledge/cont1" --payload '{"temperature":20,"humidity":80,"latitude":0,"longitude":0}' --cli-binary-format raw-in-base64-out

Now, navigate to Amazon S3 from the AWS Administration Console and choose the bucket that has been created with the CloudFormation stack. You must see the device-data folder on this bucket. It could take as much as 1 minute for the information to look because of the buffering configuration that’s set for the Firehose supply stream. Should you navigate into the device-data/YYYY/MM/dd/HH folder, you will notice an object has been created. Go forward and open this file. You will notice the content material of the file is the information payload with enriched shipId subject.

{“temperature”: 20, “humidity”: 80, “latitude”: 0, “longitude”: 0, “containerId”: “cont1”, “shipId”: “ship1”}

Troubleshooting

In case of failure within the system, the next sources may be helpful for analyzing the supply of the issue.

To observe AWS IoT Core Guidelines Engine, you want to allow AWS IoT Core logging. This can give detailed details about the occasions taking place in AWS IoT Core.

AWS Lambda may be monitored through the use of Amazon CloudWatch. The instance CloudFormation template has needed permissions to create a log group for the Lambda operate logging.

In case of failure, Kinesis Information Firehose will create a processing-failed folder beneath the device-data prefix within the AWS IoT Guidelines Engine motion, rework Lambda operate or Amazon S3 bucket. The small print of the failure may be learn as json objects. You will discover extra info on this documentation.

Clear up

To scrub up the sources which were created, first empty the Amazon S3 bucket. Run the next command by altering the bucket-name parameter with the identify of the bucket deployed by the CloudFormation stack. Essential: this command will delete all the information contained in the bucket irreversibly.

aws s3 rm s3://bucket-name --recursive

Then, you’ll be able to delete the CloudFormation stack by working the next command in a terminal.

aws cloudformation delete-stack --stack-name IoTKinesisDataPath

Conclusion

On this weblog, you will have realized a typical sample of enriching IoT payloads with metadata and storing cheaply in an information lake in close to real-time through the use of AWS IoT Guidelines Engine and Amazon Kinesis Information Firehose supply stream. The proposed resolution and the CloudFormation template can be utilized as a baseline for a scalable IoT knowledge ingestion structure.

You possibly can learn additional about AWS IoT Core Guidelines Engine and Amazon Kinesis Information Firehose. Greatest practices for utilizing MQTT matters within the AWS IoT Guidelines Engine will information you to outline your matter buildings.