在Amazon Kinesis中使用Lambda函数

AWS Kinesis 服务用于捕获/存储来自网站点击，日志，社交媒体供稿的实时跟踪数据.我们可以触发AWS Lambda对此日志执行其他处理.

必需品

开始使用Kinesis和AWS Lambda的基本要求如下所示 :

• 创建具有所需权限的角色

• 在Kinesis中创建数据流

• 创建AWS Lambda函数.

• 向AWS Lambda添加代码

• 向Kinesis数据流添加数据

示例

让我们来一个例子，我们将触发AWS Lambda处理来自Kinesis的数据流并发送邮件收到的数据.

用于解释该过程的简单框图如下所示 :

创建具有所需权限的角色

转到AWS控制台并创建角色.

在Kinesis中创建数据流

转到AWS控制台并在kinesis中创建数据流.

如图所示，有4个选项.我们将在此示例中处理创建数据流.

单击创建数据流.在下面给出的Kinesis流名称中输入名称.

输入数据流的分片数.

碎片的详细信息如下所示 :

输入名称并单击底部的创建Kinesis流按钮.

请注意，流需要一定时间才能生效.

创建AWS Lambda函数

Go到AWS控制台并单击Lambda.创建AWS Lambda函数，如图所示 :

单击在屏幕末尾创建功能按钮.将Kinesis添加为AWS Lambda的触发器.

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");
      }
   });
};

事件参数具有在kinesis数据流中输入的数据.一旦在kinesis数据流中输入数据，上面的aws lambda代码就会被激活.

将数据添加到Kinesis数据流

这里我们将使用AWS CLI添加数据kinesis数据流，如下所示.为此，我们可以使用以下命令 :

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

然后，AWS Lambda被激活并发送邮件.