Tutorial: Usar o AWS Lambda com o Amazon DynamoDB Streams

Neste tutorial, você cria uma função do Lambda para consumir eventos do Amazon DynamoDB Streams.

Pré-requisitos

Instalar a AWS Command Line Interface

Se você ainda não instalou a AWS Command Line Interface, siga as etapas em Instalar ou atualizar a versão mais recente da AWS CLI para instalá-la.

O tutorial requer um terminal de linha de comando ou um shell para executar os comandos. No Linux e no macOS, use o gerenciador de pacotes e de shell de sua preferência.

nota

No Windows, alguns comandos da CLI do Bash que você costuma usar com o Lambda (como zip) não são compatíveis com os terminais integrados do sistema operacional. Para obter uma versão do Ubuntu com o Bash integrada no Windows, instale o Subsistema do Windows para Linux.

Instalar a AWS Command Line Interface

Se você ainda não instalou a AWS Command Line Interface, siga as etapas em Instalar ou atualizar a versão mais recente da AWS CLI para instalá-la.

O tutorial requer um terminal de linha de comando ou um shell para executar os comandos. No Linux e no macOS, use o gerenciador de pacotes e de shell de sua preferência.

nota

No Windows, alguns comandos da CLI do Bash que você costuma usar com o Lambda (como zip) não são compatíveis com os terminais integrados do sistema operacional. Para obter uma versão do Ubuntu com o Bash integrada no Windows, instale o Subsistema do Windows para Linux.

Criar a função de execução

Crie a função de execução que dá à sua função permissão para acessar recursos do AWS.

Para criar uma função de execução

1. Abra a página Roles (Funções) no console do IAM.

2. Selecione Criar perfil.

3. Crie uma função com as propriedades a seguir.

   • Entidade confiável: Lambda.

   • Permissions (Permissões): AWSLambdaDynamoDBExecutionRole.

   • Role name (Nome da função): lambda-dynamodb-role.

O AWSLambdaDynamoDBExecutionRole tem as permissões necessárias para a função ler itens do DynamoDB e gravar logs no CloudWatch Logs.

Criar a função

Crie uma função do Lambda que processe seus eventos do DynamoDB. O código da função grava alguns dos dados de eventos de entrada no CloudWatch Logs.

.NET

SDK para .NET

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Como consumir um evento do DynamoDB com o Lambda usando .NET.

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text.Json;
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.DynamoDBEvents;

// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]

namespace AWSLambda_DDB;

public class Function
{
    public void FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context)
    {
        context.Logger.LogInformation($"Beginning to process {dynamoEvent.Records.Count} records...");

        foreach (var record in dynamoEvent.Records)
        {
            context.Logger.LogInformation($"Event ID: {record.EventID}");
            context.Logger.LogInformation($"Event Name: {record.EventName}");

            context.Logger.LogInformation(JsonSerializer.Serialize(record));
        }

        context.Logger.LogInformation("Stream processing complete.");
    }
}

Go

SDK para Go V2

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Como consumir um evento do DynamoDB com o Lambda usando Go.

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main

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

func HandleRequest(ctx context.Context, event events.DynamoDBEvent) (*string, error) {
	if len(event.Records) == 0 {
		return nil, fmt.Errorf("received empty event")
	}

	for _, record := range event.Records {
	 	LogDynamoDBRecord(record)
	}

	message := fmt.Sprintf("Records processed: %d", len(event.Records))
	return &message, nil
}

func main() {
	lambda.Start(HandleRequest)
}

func LogDynamoDBRecord(record events.DynamoDBEventRecord){
	fmt.Println(record.EventID)
	fmt.Println(record.EventName)
	fmt.Printf("%+v\n", record.Change)
}

Java

SDK para Java 2.x

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Consumir um evento do DynamoDB com o Lambda usando Java.

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.DynamodbEvent;
import com.amazonaws.services.lambda.runtime.events.DynamodbEvent.DynamodbStreamRecord;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

public class example implements RequestHandler<DynamodbEvent, Void> {

    private static final Gson GSON = new GsonBuilder().setPrettyPrinting().create();

    @Override
    public Void handleRequest(DynamodbEvent event, Context context) {
        System.out.println(GSON.toJson(event));
        event.getRecords().forEach(this::logDynamoDBRecord);
        return null;
    }

    private void logDynamoDBRecord(DynamodbStreamRecord record) {
        System.out.println(record.getEventID());
        System.out.println(record.getEventName());
        System.out.println("DynamoDB Record: " + GSON.toJson(record.getDynamodb()));
    }
}

JavaScript

SDK para JavaScript (v3)

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Consumir um evento do DynamoDB com o Lambda usando JavaScript.

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
    console.log(JSON.stringify(event, null, 2));
    event.Records.forEach(record => {
        logDynamoDBRecord(record);
    });
};

const logDynamoDBRecord = (record) => {
    console.log(record.eventID);
    console.log(record.eventName);
    console.log(`DynamoDB Record: ${JSON.stringify(record.dynamodb)}`);
};

Consumir um evento do DynamoDB com o Lambda usando TypeScript.

export const handler = async (event, context) => {
    console.log(JSON.stringify(event, null, 2));
    event.Records.forEach(record => {
        logDynamoDBRecord(record);
    });
}
const logDynamoDBRecord = (record) => {
    console.log(record.eventID);
    console.log(record.eventName);
    console.log(`DynamoDB Record: ${JSON.stringify(record.dynamodb)}`);
};

PHP

SDK para PHP

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Consumir um evento do DynamoDB com o Lambda usando PHP.

<?php

# using bref/bref and bref/logger for simplicity

use Bref\Context\Context;
use Bref\Event\DynamoDb\DynamoDbEvent;
use Bref\Event\DynamoDb\DynamoDbHandler;
use Bref\Logger\StderrLogger;

require __DIR__ . '/vendor/autoload.php';

class Handler extends DynamoDbHandler
{
    private StderrLogger $logger;

    public function __construct(StderrLogger $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @throws JsonException
     * @throws \Bref\Event\InvalidLambdaEvent
     */
    public function handleDynamoDb(DynamoDbEvent $event, Context $context): void
    {
        $this->logger->info("Processing DynamoDb table items");
        $records = $event->getRecords();

        foreach ($records as $record) {
            $eventName = $record->getEventName();
            $keys = $record->getKeys();
            $old = $record->getOldImage();
            $new = $record->getNewImage();
            
            $this->logger->info("Event Name:".$eventName."\n");
            $this->logger->info("Keys:". json_encode($keys)."\n");
            $this->logger->info("Old Image:". json_encode($old)."\n");
            $this->logger->info("New Image:". json_encode($new));
            
            // TODO: Do interesting work based on the new data

            // Any exception thrown will be logged and the invocation will be marked as failed
        }

        $totalRecords = count($records);
        $this->logger->info("Successfully processed $totalRecords items");
    }
}

$logger = new StderrLogger();
return new Handler($logger);

Python

SDK para Python (Boto3).

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Como consumir um evento do DynamoDB com o Lambda usando Python.

import json

def lambda_handler(event, context):
    print(json.dumps(event, indent=2))

    for record in event['Records']:
        log_dynamodb_record(record)

def log_dynamodb_record(record):
    print(record['eventID'])
    print(record['eventName'])
    print(f"DynamoDB Record: {json.dumps(record['dynamodb'])}")

Ruby

SDK para Ruby

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Como consumir um evento do DynamoDB com o Lambda usando Ruby.

def lambda_handler(event:, context:)
    return 'received empty event' if event['Records'].empty?
  
    event['Records'].each do |record|
      log_dynamodb_record(record)
    end
  
    "Records processed: #{event['Records'].length}"
  end
  
  def log_dynamodb_record(record)
    puts record['eventID']
    puts record['eventName']
    puts "DynamoDB Record: #{JSON.generate(record['dynamodb'])}"
  end
  

Rust

SDK para Rust

nota

Há mais no GitHub. Encontre o exemplo completo e saiba como configurar e executar no repositório dos Exemplos sem servidor.

Como consumir um evento do DynamoDB com o Lambda usando Rust.

use lambda_runtime::{service_fn, tracing, Error, LambdaEvent};
use aws_lambda_events::{
    event::dynamodb::{Event, EventRecord},
   };


// Built with the following dependencies:
//lambda_runtime = "0.11.1"
//serde_json = "1.0"
//tokio = { version = "1", features = ["macros"] }
//tracing = { version = "0.1", features = ["log"] }
//tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }
//aws_lambda_events = "0.15.0"

async fn function_handler(event: LambdaEvent<Event>) ->Result<(), Error> {
    
    let records = &event.payload.records;
    tracing::info!("event payload: {:?}",records);
    if records.is_empty() {
        tracing::info!("No records found. Exiting.");
        return Ok(());
    }

    for record in records{
        log_dynamo_dbrecord(record);
    }

    tracing::info!("Dynamo db records processed");

    // Prepare the response
    Ok(())

}

fn log_dynamo_dbrecord(record: &EventRecord)-> Result<(), Error>{
    tracing::info!("EventId: {}", record.event_id);
    tracing::info!("EventName: {}", record.event_name);
    tracing::info!("DynamoDB Record: {:?}", record.change );
    Ok(())

}

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing_subscriber::fmt()
    .with_max_level(tracing::Level::INFO)
    .with_target(false)
    .without_time()
    .init();

    let func = service_fn(function_handler);
    lambda_runtime::run(func).await?;
    Ok(())
    
}

Para criar a função

1. Copie o código de amostra em um arquivo chamado example.js.

2. Crie um pacote de implantação.

   zip function.zip example.js

3. Crie uma função do Lambda com o comando create-function.

   aws lambda create-function --function-name ProcessDynamoDBRecords \
       --zip-file fileb://function.zip --handler example.handler --runtime nodejs18.x \
       --role arn:aws:iam::111122223333:role/lambda-dynamodb-role

Testar a função do Lambda

Nesta etapa, você invoca sua função do Lambda manualmente usando o comando invoke da CLI do AWS Lambda e o seguinte exemplo de evento do DynamoDB. Copie o seguinte para um arquivo chamado input.txt.

exemplo input.txt

{
   "Records":[
      {
         "eventID":"1",
         "eventName":"INSERT",
         "eventVersion":"1.0",
         "eventSource":"aws:dynamodb",
         "awsRegion":"us-east-1",
         "dynamodb":{
            "Keys":{
               "Id":{
                  "N":"101"
               }
            },
            "NewImage":{
               "Message":{
                  "S":"New item!"
               },
               "Id":{
                  "N":"101"
               }
            },
            "SequenceNumber":"111",
            "SizeBytes":26,
            "StreamViewType":"NEW_AND_OLD_IMAGES"
         },
         "eventSourceARN":"stream-ARN"
      },
      {
         "eventID":"2",
         "eventName":"MODIFY",
         "eventVersion":"1.0",
         "eventSource":"aws:dynamodb",
         "awsRegion":"us-east-1",
         "dynamodb":{
            "Keys":{
               "Id":{
                  "N":"101"
               }
            },
            "NewImage":{
               "Message":{
                  "S":"This item has changed"
               },
               "Id":{
                  "N":"101"
               }
            },
            "OldImage":{
               "Message":{
                  "S":"New item!"
               },
               "Id":{
                  "N":"101"
               }
            },
            "SequenceNumber":"222",
            "SizeBytes":59,
            "StreamViewType":"NEW_AND_OLD_IMAGES"
         },
         "eventSourceARN":"stream-ARN"
      },
      {
         "eventID":"3",
         "eventName":"REMOVE",
         "eventVersion":"1.0",
         "eventSource":"aws:dynamodb",
         "awsRegion":"us-east-1",
         "dynamodb":{
            "Keys":{
               "Id":{
                  "N":"101"
               }
            },
            "OldImage":{
               "Message":{
                  "S":"This item has changed"
               },
               "Id":{
                  "N":"101"
               }
            },
            "SequenceNumber":"333",
            "SizeBytes":38,
            "StreamViewType":"NEW_AND_OLD_IMAGES"
         },
         "eventSourceARN":"stream-ARN"
      }
   ]
}

Execute o seguinte comando invoke.

aws lambda invoke --function-name ProcessDynamoDBRecords \
    --cli-binary-format raw-in-base64-out \
    --payload file://input.txt outputfile.txt

A opção cli-binary-format será necessária se você estiver usando a AWS CLI versão 2. Para que essa seja a configuração padrão, execute aws configure set cli-binary-format raw-in-base64-out. Para obter mais informações, consulte A AWS CLI comporta opções de linha de comando globais no Guia do usuário da AWS Command Line Interface versão 2.

A função retorna a string message no corpo da resposta.

Verifique a saída no arquivo outputfile.txt.

Criar uma tabela do DynamoDB com uma transmissão habilitada

Criar uma tabela do Amazon DynamoDB com uma transmissão habilitada.

Como criar uma tabela do DynamoDB

1. Abra o console do DynamoDB.

2. Escolha Create table.

3. Crie uma tabela com as configurações a seguir.

   • Table name (Nome da tabela): lambda-dynamodb-stream

   • Primary key (Chave primária): id (string)

4. Escolha Criar.

Como habilitar fluxos

1. Abra o console do DynamoDB.

2. Escolha Tables (Tabelas).

3. Escolha a tabela lambda-dynamodb-stream.

4. Em Exports and streams (Exportações e transmissões), escolha DynamoDB stream details (Detalhes da tranmissão do DynamoDB).

5. Selecione Ativar.

6. Em Tipo de exibição, escolha Somente atributos principais.

7. Escolha Ativar fluxo.

Anote o ARN do fluxo. Você precisará dele na próxima etapa quando for associar a transmissão à função do Lambda. Para obter mais informações sobre como habilitar fluxos, consulte Capturing table activity with DynamoDB Streams (Capturar atividades de tabelas com o DynamoDB Streams).

Adicionar uma fonte de eventos no AWS Lambda

Crie um mapeamento da fonte do evento no AWS Lambda. Este mapeamento de fontes de eventos associa a transmissão do DynamoDB à função do Lambda. Depois de criar esse mapeamento da fonte do evento, o AWS Lambda começa a sondar o fluxo.

Execute o comando AWS CLI a seguir da create-event-source-mapping. Depois que o comando for executado, anote o UUID. Você precisará deste UUID para se referir ao mapeamento da fonte do evento em todos os comandos, por exemplo, ao excluir o mapeamento da fonte do evento.

aws lambda create-event-source-mapping --function-name ProcessDynamoDBRecords \
    --batch-size 100 --starting-position LATEST --event-source DynamoDB-stream-arn

Isso cria um mapeamento entre a transmissão do DynamoDB especificado e a função do Lambda. Você pode associar uma transmissão do DynamoDB a várias funções do Lambda e associar a mesma função do Lambda a várias transmissões. No entanto, as funções do Lambda compartilharão o throughput de leitura para os fluxos dos quais compartilham.

Você pode obter a lista de mapeamentos de fontes de eventos executando o comando a seguir.

aws lambda list-event-source-mappings

Esta lista retorna todos os mapeamentos de fontes de eventos que você criou, e para cada mapeamento mostra o LastProcessingResult, entre outras coisas. Este campo será usado para fornecer uma mensagem informativa, se houver problemas. Valores como No records processed (indica que o AWS Lambda não iniciou a sondagem ou de que não há registros no fluxo) e OK (indica que o AWS Lambda foi bem-sucedido na leitura dos registros do fluxo e invocou a função do Lambda) indicam que não há problemas. Se houver problemas, você receberá uma mensagem de erro.

Se você tiver muitos mapeamentos da fonte do evento, use o parâmetro de nome da função para refinar os resultados.

aws lambda list-event-source-mappings --function-name ProcessDynamoDBRecords

Testar a configuração

Teste a experiência completa. À medida que você executa atualizações nas tabelas, o DynamoDB grava registros de eventos na transmissão. Quando o AWS Lambda sonda o stream, ele detecta novos registros no stream e executa a função do Lambda em seu nome, passando os eventos para a função.

1. No console do DynamoDB, adicione, atualize ou exclua itens da tabela. O DynamoDB grava registros dessas ações na transmissão.

2. O AWS Lambda sonda a transmissão e, ao detectar atualizações nela, invoca a função do Lambda passando os dados do evento que ele encontrar na transmissão.

3. A função é executada e cria logs no Amazon CloudWatch. Você pode verificar os logs relatados no console do Amazon CloudWatch.

Limpe os recursos

Agora você pode excluir os recursos criados para este tutorial, a menos que queira mantê-los. Excluindo os recursos da AWS que você não está mais usando, você evita cobranças desnecessárias em sua Conta da AWS.

Como excluir a função do Lambda

1. Abra a página Functions (Funções) no console do Lambda.

2. Selecione a função que você criou.

3. Selecione Ações, Excluir.

4. Digite confirm no campo de entrada de texto e escolha Delete (Excluir).

Para excluir a função de execução

1. Abra a página Roles (Funções) no console do IAM.

2. Selecione a função de execução que você criou.

3. Escolha Excluir.

4. Insira o nome do perfil no campo de entrada de texto e escolha Delete (Excluir).

Para excluir uma tabela do DynamoDB

1. Abra a página Tables (Tabelas) no console do DynamoDB.

2. Selecione a tabela que você criou.

3. Escolha Excluir.

4. Digite delete na caixa de texto.

5. Selecione Delete table (Excluir tabela).