Usando a função Lambda com Amazon Kinesis

AWS Kinesisserviço é usado para capturar / armazenar dados de rastreamento em tempo real provenientes de cliques em sites, logs, feeds de mídia social. Podemos acionar o AWS Lambda para realizar processamento adicional nesses logs.

Requisitos

Os requisitos básicos para começar a usar Kinesis e AWS Lambda são os mostrados -

  • Criar função com as permissões necessárias
  • Crie fluxo de dados no Kinesis
  • Crie a função AWS Lambda.
  • Adicionar código ao AWS Lambda
  • Adicionar dados ao fluxo de dados Kinesis

Exemplo

Vamos trabalhar em um exemplo em que acionaremos o AWS Lambda para processar o fluxo de dados do Kinesis e enviaremos e-mail com os dados recebidos.

Um diagrama de blocos simples para explicar o processo é mostrado abaixo -

Criar função com as permissões necessárias

Vá para o console da AWS e crie uma função.

Criar fluxo de dados no Kinesis

Vá para o console da AWS e crie o fluxo de dados no kinesis.

Existem 4 opções conforme mostrado. Vamos trabalhar em Criar fluxo de dados neste exemplo.

Clique Create data stream. Insira o nome no stream Kinesis fornecido abaixo.

Insira o número de fragmentos para o fluxo de dados.

Os detalhes dos fragmentos são mostrados abaixo -

Insira o nome e clique no Create Kinesis stream botão na parte inferior.

Observe que leva certo tempo para que o stream fique ativo.

Criar função AWS Lambda

Vá para o console da AWS e clique em Lambda. Crie a função AWS Lambda conforme mostrado -

Clique Create functionbotão no final da tela. Adicione Kinesis como o gatilho para AWS Lambda.

Adicione detalhes de configuração ao gatilho Kinesis -

Adicione o gatilho e agora adicione o código ao AWS Lambda.

Adicionando código ao AWS Lambda

Para este propósito, usaremos nodejs como o tempo de execução. Enviaremos e-mail assim que o AWS Lambda for acionado com o fluxo de dados 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: ["[email protected]"]
      },
      Message: {
         Body: {
            Text: {
               Data:payload
            }
         },
         Subject: {
            Data: "Kinesis data stream"
         }
      },
      Source: "[email protected]"
   };    
   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");
      }
   });
};

O parâmetro do evento possui os dados inseridos no fluxo de dados do kinesis. O código aws lambda acima será ativado assim que os dados forem inseridos no fluxo de dados do kinesis.

Adicionar dados ao fluxo de dados Kinesis

Aqui, usaremos o AWS CLI para adicionar fluxo de dados de cinesia de dados conforme mostrado abaixo. Para este propósito, podemos usar o seguinte comando -

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

Em seguida, o AWS Lambda é ativado e o e-mail é enviado.