Multiple additions to datastore from single JSON

Edit: formatting.

I think the simplest way to achieve this would be to leverage a Lambda Activity on your Pipeline (https://docs.aws.amazon.com/iotanalytics/latest/userguide/pipeline-activities.html#aws-iot-analytics-pipeline-activities-lambda), and have it parse the single JSON payload into the desired structure. This depends somewhat on the 'raw' structure of the messages sent to the Channel.

So, for instance, we can send data to the Channel via CLI batch-put-message (https://docs.aws.amazon.com/cli/latest/reference/iotanalytics/batch-put-message.html), like so:

aws iotanalytics batch-put-message --channel-name sample_channel --messages '[{"messageId": "message1", "payload": "{\"sensors\": [{\"name\": \"office_1\", \"value\": 24.6, \"category\": \"temperature\"},{\"name\": \"office_2\", \"value\": 55, \"category\": \"temperature\"}]}"}]'

The Channel would then have a single message structured like this:

{
  "messageId": "message1",
  "payload": {
    "sensors": [
      {
        "name": "office_1",
        "value": 24.6,
        "category": "temperature"
      },
      {
        "name": "office_2",
        "value": 55,
        "category": "temperature"
      }
    ]
  }
}

If your Pipeline has a Lambda Activity, then the message(s) from the Channel will be passed to your Lambda function in the event argument.

I created a simple Lambda function (using Python 3.7) using the AWS Lambda console inline editor, and named it sample_lambda:

import json
import sys
import logging

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
streamHandler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)


def lambda_handler(event, context):
    # This can be handy to see the raw structure of the incoming event
    # will log to the matching CloudWatch log:
    # /aws/lambda/<name_of_the_lambda>
    # logger.info("raw event: {}".format(event))
    
    parsed_rows = []
    
    # Depending on the batchSize setting of the Lambda Pipeline Activity,
    # you may receive multiple messages in a single event
    for message_payload in event:
        if 'sensors' in message_payload:
            for row in message_payload['sensors']:
                parsed = {}
                for key, value in row.items():
                    parsed[key] = value
                parsed_rows.append(parsed)

    return parsed_rows

I added the proper permissions so that the IoT-Analytics could invoke the lambda function via CLI:

aws lambda add-permission --function-name sample_lambda --statement-id statm01 --principal iotanalytics.amazonaws.com --action lambda:InvokeFunction

Reprocessing the Pipeline, the parsed rows are placed in the DataStore; executing the DataSet, I get this net result:

"sensors","name","value","category","__dt"
,"office_1",24.6,"temperature","2019-07-19 00:00:00.000"
,"office_2",55,"temperature","2019-04-26 00:00:00.000"

Edited by: Frank-AWS on Jul 22, 2019 6:39 AM

Edited by: Frank-AWS on Jul 22, 2019 6:41 AM

Thanks, this worked perfectly.