Использование лямбда-функции с Amazon Kinesis

Сервис AWS Kinesis используется для сбора / хранения данных отслеживания в режиме реального времени, поступающих с кликов веб-сайтов, журналов, каналов социальных сетей. Мы можем запустить AWS Lambda для дополнительной обработки этих журналов.

реквизиты

Основные требования для начала работы с Kinesis и AWS Lambda:

• Создать роль с необходимыми разрешениями
• Создать поток данных в Kinesis
• Создайте функцию AWS Lambda.
• Добавить код в AWS Lambda
• Добавить данные в поток данных Kinesis

пример

Давайте поработаем над примером, в котором мы будем запускать AWS Lambda для обработки потока данных из Kinesis и отправлять почту с полученными данными.

Простая блок-схема для объяснения процесса показана ниже —

Создать роль с необходимыми разрешениями

Перейдите в консоль AWS и создайте роль.

Создать поток данных в Kinesis

Перейдите в консоль AWS и создайте поток данных в kinesis.

Есть 4 варианта, как показано. Мы будем работать над созданием потока данных в этом примере.

Нажмите Создать поток данных . Введите имя в название потока Kinesis, приведенное ниже.

Введите количество шардов для потока данных.

Детали Осколков как показано ниже —

Введите имя и нажмите кнопку Создать поток Kinesis внизу.

Обратите внимание, что для активации потока требуется определенное время.

Создать лямбда-функцию AWS

Перейдите в консоль AWS и нажмите «Лямбда». Создайте функцию AWS Lambda, как показано на рисунке —

Нажмите кнопку Создать функцию в конце экрана. Добавьте Kinesis в качестве триггера для AWS Lambda.

Добавить детали конфигурации в триггер Kinesis —

Добавьте триггер и добавьте код в AWS Lambda.

Добавление кода в AWS Lambda

Для этой цели мы будем использовать nodejs во время выполнения. Мы отправим почту, как только AWS Lambda сработает с потоком данных Kinesis.

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

Параметр события содержит данные, введенные в поток данных кинезиса. Вышеуказанный лямбда-код aws активируется после ввода данных в поток данных kinesis.

Добавить данные в поток данных Kinesis

Здесь мы будем использовать AWS CLI для добавления потока данных кинезиса данных, как показано ниже. Для этого мы можем использовать следующую команду —

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

Затем активируется AWS Lambda и отправляется письмо.