Using Lambda Function with Amazon Kinesis

AWS Kinesis service is used to capture/store real time tracking data coming from website clicks, logs, social media feeds. We can trigger AWS Lambda to perform additional processing on this logs.

Requisites

The basic requirements to get started with Kinesis and AWS Lambda are as shown −

• Create role with required permissions
• Create data stream in Kinesis
• Create AWS Lambda function.
• Add code to AWS Lambda
• Add data to Kinesis data stream

Example

Let us work on an example wherein we will trigger AWS Lambda for processing the data stream from Kinesis and send mail with the data received.

A simple block diagram for explaining the process is shown below −

Create Role with Required Permissions

Go to AWS console and create a role.

Create Data Stream in Kinesis

Go to AWS console and create data stream in kinesis.

There are 4 options as shown. We will work on Create data stream in this example.

Click Create data stream. Enter the name in Kinesis stream name given below.

Enter number of shards for the data stream.

The details of Shards are as shown below −

Enter the name and click the Create Kinesis stream button at the bottom.

Note that it takes certain time for the stream to go active.

Create AWS Lambda Function

Go to AWS console and click Lambda. Create AWS Lambda function as shown −

Click Create function button at the end of the screen. Add Kinesis as the trigger to AWS Lambda.

Add configuration details to the Kinesis trigger −

Add the trigger and now add code to AWS Lambda.

Adding Code to AWS Lambda

For this purpose, we will use nodejs as the run-time. We will send mail once AWS Lambda is triggered with kinesis data stream.

const aws =  require("aws-sdk");
var ses = new aws.SES({
   region: 'us-east-1'
});
exports.handler = function(event, context, callback) {
   let payload = "";
   event.Records.forEach(function(record) {
      // Kinesis data is base64 encoded so decode here
      payload = new Buffer(record.kinesis.data, 'base64').toString('ascii');
      console.log('Decoded payload:', payload);
   });
   var eParams = {
      Destination: {
         ToAddresses: ["xxxxxxx@gmail.com"]
      },
      Message: {
         Body: {
            Text: {
               Data:payload
            }
         },
         Subject: {
            Data: "Kinesis data stream"
         }
      },
      Source: "cxxxxxxxxx@gmail.com"
   };    
   var email = ses.sendEmail(eParams, function(err, data) {
      if (err) console.log(err);
      else {
         console.log("===EMAIL SENT===");
         console.log("EMAIL CODE END");
         console.log('EMAIL: ', email);
         context.succeed(event);
         callback(null, "email is send");
      }
   });
};

The event param has the data entered in kinesis data stream. The above aws lambda code will get activated once data is entered in kinesis data stream.

Add Data to Kinesis Data Stream

Here we will use AWS CLI to add data kinesis data stream as shown below. For this purpose, we can use the following command −

aws kinesis put-record --stream-name kinesisdemo  --data "hello world" --
partition-key "789675"

Then, AWS Lambda is activated and the mail is sent.