Código de exemplo da função do

Para processar eventos do Amazon Kinesis, percorra os registros incluídos no objeto de evento e decodifique os dados codificados em base64 incluídos em cada um.

nota

O código nesta página não oferece suporte a registros agregados. Você pode desativar a agregação na configuração da Biblioteca do Produtor ou usar a biblioteca de agregação de registros do Kinesis para desagregar registros.

O código de amostra está disponível para as seguintes linguagens.

Node.js 12.x

O código de amostra a seguir recebe uma entrada de evento do Kinesis e processa as mensagens que ela contém. Por exemplo, o código grava alguns dos dados de eventos recebidos em CloudWatch Logs.

exemplo index.js

console.log('Loading function');

exports.handler = function(event, context) {
    //console.log(JSON.stringify(event, null, 2));
    event.Records.forEach(function(record) {
        // Kinesis data is base64 encoded so decode here
        var payload = Buffer.from(record.kinesis.data, 'base64').toString('ascii');
        console.log('Decoded payload:', payload);
    });
};

Cria um zip do código do exemplo para criar um pacote de implantação. Para obter instruções, consulte Implantar funções do Lambda em Node.js com arquivos .zip.

Java 11

Veja a seguir um exemplo de código Java que recebe dados de registro de evento do Kinesis como uma entrada e a processa. Por exemplo, o código grava alguns dos dados de eventos recebidos em CloudWatch Logs.

No código, recordHandler é o handler. O handler usa a classe predefinida KinesisEvent definida na biblioteca aws-lambda-java-events.

exemplo ProcessKinesisEvents.java

package example;

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent.KinesisEventRecord;

public class ProcessKinesisRecords implements RequestHandler<KinesisEvent, Void>{
  @Override
  public Void handleRequest(KinesisEvent event, Context context)
  {
    for(KinesisEventRecord rec : event.getRecords()) {
      System.out.println(new String(rec.getKinesis().getData().array()));
  }
  return null;
  }
}

Se o manipulador obtém um retorno normal sem exceções, o Lambda considera que os lotes de entrada de registros foram processados com êxito e começa a ler novos registros no fluxo. Se o manipulador gera uma exceção, o Lambda considera que os lotes de entrada de registros não foram processados e invoca novamente a função com o mesmo lote de registros.

Dependências

• aws-lambda-java-core

• aws-lambda-java-events

• aws-java-sdk

Crie o código com as dependências da biblioteca do Lambda para criar um pacote de implantação. Para obter instruções, consulte Implantar funções do Lambda em Java com arquivos .zip ou JAR.

C#

Veja a seguir um exemplo de código C# que recebe dados de registro de evento do Kinesis como uma entrada e a processa. Por exemplo, o código grava alguns dos dados de eventos recebidos em CloudWatch Logs.

No código, HandleKinesisRecord é o handler. O handler usa a classe predefinida KinesisEvent definida na biblioteca Amazon.Lambda.KinesisEvents.

exemplo ProcessingKinesisEvents.cs

using System;
using System.IO;
using System.Text;

using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;

namespace KinesisStreams
{
    public class KinesisSample
    {
    	[LambdaSerializer(typeof(JsonSerializer))]
        public void HandleKinesisRecord(KinesisEvent kinesisEvent)
        {
            Console.WriteLine($"Beginning to process {kinesisEvent.Records.Count} records...");

            foreach (var record in kinesisEvent.Records)
            {
                Console.WriteLine($"Event ID: {record.EventId}");
                Console.WriteLine($"Event Name: {record.EventName}");

                string recordData = GetRecordContents(record.Kinesis);
                Console.WriteLine($"Record Data:");
                Console.WriteLine(recordData);
            }
            Console.WriteLine("Stream processing complete.");
        }

        private string GetRecordContents(KinesisEvent.Record streamRecord)
        {
            using (var reader = new StreamReader(streamRecord.Data, Encoding.ASCII))
            {
                return reader.ReadToEnd();
            }
        }
    }
}

Substitua o Program.cs em um projeto do .NET Core pelo exemplo acima. Para obter instruções, consulte Criar e implantar funções do Lambda em C# com arquivos .zip.

Python 3

Veja a seguir um exemplo de código Python que recebe dados de registro de eventos do Kinesis como uma entrada e a processa. Por exemplo, o código grava alguns dos dados de eventos recebidos no CloudWatch Logs.

exemplo ProcessKinesisRecords.py

from __future__ import print_function
#import json
import base64
def lambda_handler(event, context):
    for record in event['Records']:
       #Kinesis data is base64 encoded so decode here
       payload=base64.b64decode(record["kinesis"]["data"])
       print("Decoded payload: " + str(payload))

Cria um zip do código do exemplo para criar um pacote de implantação. Para obter instruções, consulte Trabalhar com arquivos .zip para funções do Lambda em Python.

Go

O seguinte é um exemplo de código Go que recebe dados de registros de eventos do Kinesis como uma entrada e os processa. Por exemplo, o código grava alguns dos dados de eventos recebidos no CloudWatch Logs.

exemplo ProcessKinesisRecords.go

import (
    "strings"
    "github.com/aws/aws-lambda-go/events"
)

func handler(ctx context.Context, kinesisEvent events.KinesisEvent) {
    for _, record := range kinesisEvent.Records {
        kinesisRecord := record.Kinesis
        dataBytes := kinesisRecord.Data
        dataText := string(dataBytes)

        fmt.Printf("%s Data = %s \n", record.EventName, dataText)
    }
}

Crie o executável com o go build e crie um pacote de implantação. Para obter instruções, consulte Implantar funções do Lambda em Go com arquivos .zip.