Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
# Utilizzo AWS Lambda con Amazon DynamoDB
**Nota**
[Se desideri inviare dati a una destinazione diversa da una funzione Lambda o arricchire i dati prima di inviarli, consulta Amazon Pipes. EventBridge ](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html)
Puoi utilizzare una AWS Lambda funzione per elaborare i record in un flusso [Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html). Con DynamoDB Streams, è possibile attivare una funzione Lambda per eseguire lavoro aggiuntivo ogni volta che una tabella DynamoDB viene aggiornata.
Durante l'elaborazione di flussi DynamoDB, è necessario implementare una logica di risposta batch parziale per evitare che i record elaborati correttamente vengano ritentati quando alcuni record di un batch falliscono. [L'utilità Batch Processor](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) di Powertools for AWS Lambda è disponibile in Python TypeScript, .NET e Java e semplifica questa implementazione gestendo automaticamente la logica di risposta batch parziale, riducendo i tempi di sviluppo e migliorando l'affidabilità.
**Topics**
+ [
## Flussi di polling e batching
](#dynamodb-polling-and-batching)
+ [
## Posizioni di partenza di polling e flussi
](#dyanmo-db-stream-poll)
+ [
## Lettori simultanei di uno shard in DynamoDB Streams
](#events-dynamodb-simultaneous-readers)
+ [
## Esempio di evento
](#events-sample-dynamodb)
+ [
# Elaborare i record DynamoDB con Lambda
](services-dynamodb-eventsourcemapping.md)
+ [
# Configurazione della risposta batch parziale con DynamoDB e Lambda
](services-ddb-batchfailurereporting.md)
+ [
# Conservare i record scartati per un'origine eventi DynamoDB in Lambda
](services-dynamodb-errors.md)
+ [
# Implementazione dell'elaborazione del flusso DynamoDB stateful in Lambda
](services-ddb-windows.md)
+ [
# Parametri Lambda per gli strumenti di mappatura dell'origine degli eventi di Amazon DynamoDB
](services-ddb-params.md)
+ [
# Utilizzo del filtro eventi con un'origine eventi DynamoDB
](with-ddb-filtering.md)
+ [
# Tutorial: Utilizzo AWS Lambda con flussi Amazon DynamoDB
](with-ddb-example.md)
## Flussi di polling e batching
Lambda esegue il polling delle partizioni presenti nel proprio flusso DynamoDB ricercando i record a una velocità di base di 4 volte al secondo. Quando sono disponibili dei record, Lambda invoca la funzione e attende il risultato. Se l'elaborazione ha esito positivo, Lambda riprende il polling fino a quando non riceve più record.
Per impostazione predefinita, Lambda richiama la funzione non appena i record sono disponibili. Se il batch che Lambda legge dall'origine eventi contiene un solo record, Lambda invia solo un record alla funzione. Per evitare di richiamare la funzione con pochi record è possibile, configurando un *periodo di batch*, chiedere all'origine eventi di memorizzare nel buffer i registri per un massimo di 5 minuti. Prima di richiamare la funzione, Lambda continua a leggere i registri dall'origine eventi fino a quando non ha raccolto un batch completo, fino alla scadenza del periodo di batch o fino a quando il batch non ha raggiunto il limite del payload di 6 MB. Per ulteriori informazioni, consulta [Comportamento di batching](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching).
**avvertimento**
Gli strumenti di mappatura dell'origine degli eventi elaborano ogni evento almeno una volta e può verificarsi un'elaborazione duplicata dei record. Per evitare potenziali problemi legati agli eventi duplicati, ti consigliamo vivamente di rendere idempotente il codice della funzione. Per ulteriori informazioni, consulta [Come posso rendere idempotente la mia funzione Lambda](https://repost.aws/knowledge-center/lambda-function-idempotent) nel Knowledge Center. AWS
Lambda non attende il completamento di alcuna [estensione](lambda-extensions.md) prima di inviare il batch successivo per l'elaborazione. In altre parole, le estensioni possono continuare a funzionare mentre Lambda elabora il successivo batch di record. Ciò può causare problemi di limitazione in caso di violazione delle impostazioni o dei limiti di [simultaneità](lambda-concurrency.md). Per rilevare se si tratta di un potenziale problema, monitora le tue funzioni e verifica se i [parametri di simultaneità](monitoring-concurrency.md#general-concurrency-metrics) per lo strumento di mappatura dell'origine degli eventi sono superiori al previsto. A causa degli intervalli ridotti tra le invocazioni, Lambda potrebbe segnalare brevemente un utilizzo della simultaneità maggiore rispetto al numero di partizioni. Ciò può essere vero anche per le funzioni Lambda senza estensioni.
Configura l'[ ParallelizationFactor](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-ParallelizationFactor)impostazione per elaborare uno shard di un flusso DynamoDB con più di una chiamata Lambda contemporaneamente. È possibile specificare il numero di batch simultanei di cui Lambda esegue il polling da uno shard da un fattore di parallelizzazione compreso tra da 1 (predefinito) e 10. Ad esempio, impostando `ParallelizationFactor` su 2, puoi disporre al massimo di 200 invocazioni Lambda simultanee per elaborare 100 shard di flusso DynamoDB (anche se nella pratica potresti vedere valori differenti per il parametro `ConcurrentExecutions`). Ciò permette di dimensionare verso l'alto il throughput di elaborazione quando il volume dei dati è volatile e l'[IteratorAge](monitoring-metrics-types.md#performance-metrics) è alta. Se si aumenta il numero di batch simultanei per shard, Lambda garantisce comunque l'ordine di elaborazione a livello di elemento (partizione e chiave di ordinamento).
## Posizioni di partenza di polling e flussi
Tieni presente che il polling dei flussi durante la creazione e gli aggiornamenti dello strumento di mappatura dell’origine degli eventi alla fine è coerente.
+ Durante la creazione dello strumento di mappatura dell'origine degli eventi, potrebbero essere necessari alcuni minuti per l'avvio degli eventi di polling dal flusso.
+ Durante gli aggiornamenti dello strumento di mappatura dell'origine degli eventi, potrebbero essere necessari alcuni minuti per l'avvio degli eventi di polling dal flusso.
Questo comportamento implica che se specifichi `LATEST` come posizione iniziale del flusso, lo strumento di mappatura dell’origine degli eventi potrebbe perdere eventi durante la creazione o gli aggiornamenti. Per garantire che nessun evento venga perso, specifica la posizione di inizio del flusso come `TRIM_HORIZON`.
## Lettori simultanei di uno shard in DynamoDB Streams
Per le tabelle di una regione singola che non sono tabelle globali, è possibile progettare contemporaneamente fino a due funzioni Lambda per leggere dallo stesso shard di DynamoDB Streams nello stesso momento. Il superamento di questo limite comporta una limitazione delle richieste. Per le tabelle globali consigliamo di limitare il numero di lettori simultanei a uno per evitare richieste di limitazione della larghezza di banda della rete.
## Esempio di evento
**Example**
```
{
"Records": [
{
"eventID": "1",
"eventVersion": "1.0",
"dynamodb": {
"Keys": {
"Id": {
"N": "101"
}
},
"NewImage": {
"Message": {
"S": "New item!"
},
"Id": {
"N": "101"
}
},
"StreamViewType": "NEW_AND_OLD_IMAGES",
"SequenceNumber": "111",
"SizeBytes": 26
},
"awsRegion": "us-west-2",
"eventName": "INSERT",
"eventSourceARN": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525",
"eventSource": "aws:dynamodb"
},
{
"eventID": "2",
"eventVersion": "1.0",
"dynamodb": {
"OldImage": {
"Message": {
"S": "New item!"
},
"Id": {
"N": "101"
}
},
"SequenceNumber": "222",
"Keys": {
"Id": {
"N": "101"
}
},
"SizeBytes": 59,
"NewImage": {
"Message": {
"S": "This item has changed"
},
"Id": {
"N": "101"
}
},
"StreamViewType": "NEW_AND_OLD_IMAGES"
},
"awsRegion": "us-west-2",
"eventName": "MODIFY",
"eventSourceARN": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525",
"eventSource": "aws:dynamodb"
}
]}
```
# Elaborare i record DynamoDB con Lambda
Crea una mappatura dell'origine eventi per indicare a Lambda di inviare i record dal proprio flusso a una funzione Lambda. È possibile creare più mappature dell'origine eventi per elaborare gli stessi dati con più funzioni Lambda o per elaborare elementi da più flussi con una singola funzione.
È possibile configurare gli strumenti di mappatura dell'origine degli eventi per elaborare i record da un flusso in un altro Account AWS. Per ulteriori informazioni, consulta [Creazione di uno strumento di mappatura dell'origine degli eventi multi-account](#services-dynamodb-eventsourcemapping-cross-account).
**Per configurare la tua funzione per la lettura da DynamoDB Streams, collega la policy gestita di [DBExecutionDynamo Role AWSLambda al](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaDynamoDBExecutionRole.html) tuo AWS ruolo di esecuzione e quindi crea un trigger DynamoDB.**
**Per aggiungere le autorizzazioni e creare un trigger**
1. Aprire la pagina [Functions](https://console.aws.amazon.com/lambda/home#/functions) (Funzioni) della console Lambda.
1. Scegliere il nome della funzione.
1. Quindi, seleziona la scheda **Configuration** (Configurazione) e poi **Permissions** (Autorizzazioni).
1. In **Nome del ruolo**, scegli il link al tuo ruolo di esecuzione. Questo ruolo si apre nella console IAM.
![\[\]](http://docs.aws.amazon.com/it_it/lambda/latest/dg/images/execution-role.png)
1. Seleziona **Aggiungi autorizzazioni**, quindi seleziona **Collega policy**.
![\[\]](http://docs.aws.amazon.com/it_it/lambda/latest/dg/images/attach-policies.png)
1. Inserisci `AWSLambdaDynamoDBExecutionRole` nel campo di ricerca. Aggiungi questa policy al tuo ruolo di esecuzione. Si tratta di una policy AWS gestita che contiene le autorizzazioni che la funzione deve leggere dal flusso DynamoDB. *Per ulteriori informazioni su questa politica, consulta [AWSLambdaDynamo DBExecution Role](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaDynamoDBExecutionRole.html) nel Managed Policy Reference.AWS *
1. Torna alla funzione nella console Lambda. In **Panoramica delle funzioni**, scegliere **Aggiungi trigger**.
![\[\]](http://docs.aws.amazon.com/it_it/lambda/latest/dg/images/add-trigger.png)
1. Scegliere un tipo di trigger.
1. Configurare le opzioni richieste, quindi scegliere **Add (Aggiungi)**.
Lambda supporta le seguenti opzioni per le origini eventi DynamoDB:
**Opzioni di origine eventi**
+ **DynamoDB table (Tabella DynamoDB):** la tabella DynamoDB da cui leggere i record.
+ **Batch size (Dimensione batch):** il numero di record da inviare alla funzione in ogni batch, fino a 10.000. Lambda passa tutti i record del batch alla funzione in una singola chiamata, purché la dimensione totale degli eventi non superi il [limite di payload](gettingstarted-limits.md) per l'invocazione sincrona (6 MB).
+ **Batch window (Periodo di batch):** specifica il tempo massimo in secondi per la raccolta dei record prima di richiamare la funzione.
+ **Starting position (Posizione iniziale):** elabora solo i nuovi record o tutti i record esistenti.
+ **Latest (Ultimi)**: consente di elaborare i nuovi record aggiunti al flusso.
+ **Trim Horizon (Orizzonte di taglio):** elabora tutti i record contenuti nel flusso.
Dopo l'elaborazione di qualsiasi record esistente, la funzione è aggiornata e continua a elaborare nuovi record.
+ **Destinazione in caso di errore**: una coda SQS standard o un argomento SNS standard per i record che non è possibile elaborare. Quando Lambda scarta un batch di record perché è troppo datato o ha esaurito tutti i tentativi, invia i dettagli sul batch alla coda o all'argomento.
+ **Retry attempts (Nuovi tentativi):** il numero massimo di tentativi che Lambda effettua quando la funzione restituisce un errore. Non si applica agli errori di servizio o alle limitazioni in cui il batch non ha raggiunto la funzione.
+ **Maximum age of record (Età massima del record):** l'età massima di un record inviato da Lambda alla funzione.
+ **Split batch on error (Dividi batch in caso di errore):** quando la funzione restituisce un errore, divide il batch in due prima di un nuovo tentativo. L'impostazione originale delle dimensioni del batch rimane invariata.
+ **Batch simultanei per shard**: elabora più batch dallo stesso shard simultaneamente.
+ **Enabled (Abilitato):** impostare su true per abilitare la mappatura dell'origine eventi. Impostare su false per interrompere l'elaborazione dei record. Lambda registra l'ultimo record elaborato e riprende l'elaborazione da quel punto quando la mappatura viene riabilitata.
**Nota**
Non ti vengono addebitati costi per le chiamate GetRecords API richiamate da Lambda come parte dei trigger di DynamoDB.
Per gestire la configurazione dell'origine eventi in un momento successivo, scegliere il trigger nel designer.
## Creazione di uno strumento di mappatura dell'origine degli eventi multi-account
[Amazon DynamoDB ora supporta policy basate sulle risorse.](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/access-control-resource-based.html) Con questa funzionalità, puoi elaborare i dati da un flusso DynamoDB in uno Account AWS con una funzione Lambda in un altro account.
Per creare una mappatura delle sorgenti degli eventi per la tua funzione Lambda utilizzando un flusso DynamoDB in un Account AWS altro, devi configurare il flusso utilizzando una policy basata sulle risorse per concedere alla funzione Lambda il permesso di leggere i record. *Per informazioni su come configurare lo stream per l'accesso tra account diversi, consulta [Condividi l'accesso con le funzioni Lambda tra](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/rbac-cross-account-access.html#rbac-analyze-cross-account-lambda-access) account nella Amazon DynamoDB Developer Guide.*
Dopo aver configurato lo stream con una policy basata sulle risorse che fornisce alla funzione Lambda le autorizzazioni richieste, crea la mappatura delle sorgenti degli eventi con il tuo ARN del flusso tra account. Puoi trovare l'ARN dello stream nella scheda **Esportazioni e flussi** della tabella nella console DynamoDB per più account.
Quando usi la console Lambda, incolla l'ARN dello stream direttamente nel campo di input della tabella DynamoDB nella pagina di creazione della mappatura delle sorgenti degli eventi.
**Nota: i trigger interregionali** non sono supportati.
# Configurazione della risposta batch parziale con DynamoDB e Lambda
Quando si consumano ed elaborano i dati di streaming da un'origine eventi, per impostazione predefinita i Lambda imposta i checkpoint al numero di sequenza più alto di un batch solo quando il batch è riuscito completamente. Lambda tratta tutti gli altri risultati come un fallimento completo e riprova a elaborare il batch fino al limite di tentativi. Per consentire i successi parziali durante l'elaborazione di batch da un flusso, attivare `ReportBatchItemFailures`. Consentire successi parziali può contribuire a ridurre il numero di tentativi su un record, anche se non impedisce del tutto la possibilità di tentativi in un record riuscito.
Per attivare `ReportBatchItemFailures`, includere il valore enum**ReportBatchItemFailures** nell'[FunctionResponseTypes](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-FunctionResponseTypes)elenco. Questo elenco indica quali tipi di risposta sono abilitati per la funzione. È possibile configurare questo elenco quando si [crea](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) o si [aggiorna](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) uno strumento di mappatura dell'origine degli eventi.
**Nota**
Anche quando il codice della funzione restituisce risposte parziali di errore in batch, queste risposte non verranno elaborate da Lambda a meno che la `ReportBatchItemFailures` funzionalità non sia esplicitamente attivata per la mappatura delle sorgenti degli eventi.
## Sintassi di report
Quando si configura la creazione di report sugli errori degli elementi batch, la `StreamsEventResponse` classe viene restituita con un elenco di errori degli articoli batch. È possibile utilizzare un `StreamsEventResponse` oggetto per restituire il numero di sequenza del primo record non riuscito nel batch. È inoltre possibile creare la propria classe personalizzata utilizzando la sintassi di risposta corretta. La seguente struttura JSON mostra la sintassi di risposta richiesta:
```
{
"batchItemFailures": [
{
"itemIdentifier": ""
}
]
}
```
**Nota**
Se l'array `batchItemFailures` contiene più elementi, Lambda utilizza il record con il numero di sequenza più basso come checkpoint. Lambda quindi riprova tutti i record a partire da quel checkpoint.
## Condizioni di successo e di errore
Lambda considera un batch come un successo completo se si restituisce uno degli elementi seguenti:
+ Una `batchItemFailure` lista vuota
+ Un `batchItemFailure` elenco nullo
+ Un vuoto `EventResponse`
+ Un valore nullo `EventResponse`
Lambda considera un batch come un fallimento completo se si restituisce uno degli elementi seguenti:
+ Una stringa vuota `itemIdentifier`
+ Un valore nullo `itemIdentifier`
+ Un `itemIdentifier` con un nome chiave errato
Lambda esegue nuovi tentativi in seguito ai fallimenti secondo la strategia di tentativi impostata.
## Bisezione un batch
Se il richiamo fallisce ed `BisectBatchOnFunctionError` è attivato, il batch viene bisecato a prescindere dalle `ReportBatchItemFailures` impostazioni.
Quando si riceve una risposta di successo parziale del batch ed entrambi `BisectBatchOnFunctionError` e `ReportBatchItemFailures` sono attivati, il batch viene bisecato in corrispondenza del numero di sequenza restituito e Lambda ritenta solo i record rimanenti.
Per semplificare l'implementazione della logica di risposta batch parziale, prendi in considerazione l'utilizzo dell'[utilità Batch Processor](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) di Powertools for AWS Lambda, che gestisce automaticamente queste complessità per te.
Ecco alcuni esempi di codice di funzione che restituiscono l'elenco dei messaggi non riusciti IDs nel batch:
------
#### [ .NET ]
**SDK per .NET**
C'è altro su GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite .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 StreamsEventResponse FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context)
{
context.Logger.LogInformation($"Beginning to process {dynamoEvent.Records.Count} records...");
List batchItemFailures = new List();
StreamsEventResponse streamsEventResponse = new StreamsEventResponse();
foreach (var record in dynamoEvent.Records)
{
try
{
var sequenceNumber = record.Dynamodb.SequenceNumber;
context.Logger.LogInformation(sequenceNumber);
}
catch (Exception ex)
{
context.Logger.LogError(ex.Message);
batchItemFailures.Add(new StreamsEventResponse.BatchItemFailure() { ItemIdentifier = record.Dynamodb.SequenceNumber });
}
}
if (batchItemFailures.Count > 0)
{
streamsEventResponse.BatchItemFailures = batchItemFailures;
}
context.Logger.LogInformation("Stream processing complete.");
return streamsEventResponse;
}
}
```
------
#### [ Go ]
**SDK per Go V2**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite 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/events"
"github.com/aws/aws-lambda-go/lambda"
)
type BatchItemFailure struct {
ItemIdentifier string `json:"ItemIdentifier"`
}
type BatchResult struct {
BatchItemFailures []BatchItemFailure `json:"BatchItemFailures"`
}
func HandleRequest(ctx context.Context, event events.DynamoDBEvent) (*BatchResult, error) {
var batchItemFailures []BatchItemFailure
curRecordSequenceNumber := ""
for _, record := range event.Records {
// Process your record
curRecordSequenceNumber = record.Change.SequenceNumber
}
if curRecordSequenceNumber != "" {
batchItemFailures = append(batchItemFailures, BatchItemFailure{ItemIdentifier: curRecordSequenceNumber})
}
batchResult := BatchResult{
BatchItemFailures: batchItemFailures,
}
return &batchResult, nil
}
func main() {
lambda.Start(HandleRequest)
}
```
------
#### [ Java ]
**SDK per Java 2.x**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite Java.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
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.StreamsEventResponse;
import com.amazonaws.services.lambda.runtime.events.models.dynamodb.StreamRecord;
import java.util.ArrayList;
import java.util.List;
public class ProcessDynamodbRecords implements RequestHandler {
@Override
public StreamsEventResponse handleRequest(DynamodbEvent input, Context context) {
List batchItemFailures = new ArrayList<>();
String curRecordSequenceNumber = "";
for (DynamodbEvent.DynamodbStreamRecord dynamodbStreamRecord : input.getRecords()) {
try {
//Process your record
StreamRecord dynamodbRecord = dynamodbStreamRecord.getDynamodb();
curRecordSequenceNumber = dynamodbRecord.getSequenceNumber();
} catch (Exception e) {
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber));
return new StreamsEventResponse(batchItemFailures);
}
}
return new StreamsEventResponse();
}
}
```
------
#### [ JavaScript ]
**SDK per JavaScript (v3)**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione degli errori degli elementi batch di DynamoDB utilizzando Lambda. JavaScript
```
export const handler = async (event) => {
const records = event.Records;
let curRecordSequenceNumber = "";
for (const record of records) {
try {
// Process your record
curRecordSequenceNumber = record.dynamodb.SequenceNumber;
} catch (e) {
// Return failed record's sequence number
return { batchItemFailures: [{ itemIdentifier: curRecordSequenceNumber }] };
}
}
return { batchItemFailures: [] };
};
```
Segnalazione degli errori degli elementi batch di DynamoDB utilizzando Lambda. TypeScript
```
import {
DynamoDBBatchResponse,
DynamoDBBatchItemFailure,
DynamoDBStreamEvent,
} from "aws-lambda";
export const handler = async (
event: DynamoDBStreamEvent
): Promise => {
const batchItemFailures: DynamoDBBatchItemFailure[] = [];
let curRecordSequenceNumber;
for (const record of event.Records) {
curRecordSequenceNumber = record.dynamodb?.SequenceNumber;
if (curRecordSequenceNumber) {
batchItemFailures.push({
itemIdentifier: curRecordSequenceNumber,
});
}
}
return { batchItemFailures: batchItemFailures };
};
```
------
#### [ PHP ]
**SDK per PHP**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite PHP.
```
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): array
{
$dynamoDbEvent = new DynamoDbEvent($event);
$this->logger->info("Processing records");
$records = $dynamoDbEvent->getRecords();
$failedRecords = [];
foreach ($records as $record) {
try {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$failedRecords[] = $record->getSequenceNumber();
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
// change format for the response
$failures = array_map(
fn(string $sequenceNumber) => ['itemIdentifier' => $sequenceNumber],
$failedRecords
);
return [
'batchItemFailures' => $failures
];
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**SDK per Python (Boto3)**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite Python.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def handler(event, context):
records = event.get("Records")
curRecordSequenceNumber = ""
for record in records:
try:
# Process your record
curRecordSequenceNumber = record["dynamodb"]["SequenceNumber"]
except Exception as e:
# Return failed record's sequence number
return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]}
return {"batchItemFailures":[]}
```
------
#### [ Ruby ]
**SDK per Ruby**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite Ruby.
```
def lambda_handler(event:, context:)
records = event["Records"]
cur_record_sequence_number = ""
records.each do |record|
begin
# Process your record
cur_record_sequence_number = record["dynamodb"]["SequenceNumber"]
rescue StandardError => e
# Return failed record's sequence number
return {"batchItemFailures" => [{"itemIdentifier" => cur_record_sequence_number}]}
end
end
{"batchItemFailures" => []}
end
```
------
#### [ Rust ]
**SDK per Rust**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di DynamoDB con Lambda tramite Rust.
```
use aws_lambda_events::{
event::dynamodb::{Event, EventRecord, StreamRecord},
streams::{DynamoDbBatchItemFailure, DynamoDbEventResponse},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
/// Process the stream record
fn process_record(record: &EventRecord) -> Result<(), Error> {
let stream_record: &StreamRecord = &record.change;
// process your stream record here...
tracing::info!("Data: {:?}", stream_record);
Ok(())
}
/// Main Lambda handler here...
async fn function_handler(event: LambdaEvent) -> Result {
let mut response = DynamoDbEventResponse {
batch_item_failures: vec![],
};
let records = &event.payload.records;
if records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(response);
}
for record in records {
tracing::info!("EventId: {}", record.event_id);
// Couldn't find a sequence number
if record.change.sequence_number.is_none() {
response.batch_item_failures.push(DynamoDbBatchItemFailure {
item_identifier: Some("".to_string()),
});
return Ok(response);
}
// Process your record here...
if process_record(record).is_err() {
response.batch_item_failures.push(DynamoDbBatchItemFailure {
item_identifier: record.change.sequence_number.clone(),
});
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return Ok(response);
}
}
tracing::info!("Successfully processed {} record(s)", records.len());
Ok(response)
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
## Utilizzo di Powertools per il processore AWS Lambda batch
L'utilità per il processore batch di Powertools gestisce AWS Lambda automaticamente la logica di risposta parziale in batch, riducendo la complessità dell'implementazione della segnalazione degli errori dei batch. Ecco alcuni esempi di utilizzo del processore batch:
**Python**
Per esempi completi e istruzioni di configurazione, consultate la [documentazione del processore batch](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/).
Elaborazione dei record AWS Lambda di flusso DynamoDB con processore batch.
```
import json
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, process_partial_response
from aws_lambda_powertools.utilities.data_classes import DynamoDBStreamEvent
from aws_lambda_powertools.utilities.typing import LambdaContext
processor = BatchProcessor(event_type=EventType.DynamoDBStreams)
logger = Logger()
def record_handler(record):
logger.info(record)
# Your business logic here
# Raise an exception to mark this record as failed
def lambda_handler(event, context: LambdaContext):
return process_partial_response(
event=event,
record_handler=record_handler,
processor=processor,
context=context
)
```
**TypeScript**
Per esempi completi e istruzioni di configurazione, consultate la documentazione del [processore batch](https://docs.aws.amazon.com/powertools/typescript/latest/features/batch/).
Elaborazione dei record AWS Lambda di flusso DynamoDB con processore batch.
```
import { BatchProcessor, EventType, processPartialResponse } from '@aws-lambda-powertools/batch';
import { Logger } from '@aws-lambda-powertools/logger';
import type { DynamoDBStreamEvent, Context } from 'aws-lambda';
const processor = new BatchProcessor(EventType.DynamoDBStreams);
const logger = new Logger();
const recordHandler = async (record: any): Promise => {
logger.info('Processing record', { record });
// Your business logic here
// Throw an error to mark this record as failed
};
export const handler = async (event: DynamoDBStreamEvent, context: Context) => {
return processPartialResponse(event, recordHandler, processor, {
context,
});
};
```
**Java**
Per esempi completi e istruzioni di configurazione, consultate la documentazione del [processore batch](https://docs.powertools.aws.dev/lambda/java/latest/utilities/batch/).
Elaborazione dei record AWS Lambda di flusso DynamoDB con processore batch.
```
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.StreamsEventResponse;
import software.amazon.lambda.powertools.batch.BatchMessageHandlerBuilder;
import software.amazon.lambda.powertools.batch.handler.BatchMessageHandler;
public class DynamoDBStreamBatchHandler implements RequestHandler {
private final BatchMessageHandler handler;
public DynamoDBStreamBatchHandler() {
handler = new BatchMessageHandlerBuilder()
.withDynamoDbBatchHandler()
.buildWithRawMessageHandler(this::processMessage);
}
@Override
public StreamsEventResponse handleRequest(DynamodbEvent ddbEvent, Context context) {
return handler.processBatch(ddbEvent, context);
}
private void processMessage(DynamodbEvent.DynamodbStreamRecord dynamodbStreamRecord, Context context) {
// Process the change record
}
}
```
**.NET**
Per esempi completi e istruzioni di configurazione, consultate la documentazione del [processore batch](https://docs.aws.amazon.com/powertools/dotnet/utilities/batch-processing/).
Elaborazione dei record AWS Lambda di flusso DynamoDB con processore batch.
```
using System;
using System.Threading;
using System.Threading.Tasks;
using Amazon.Lambda.Core;
using Amazon.Lambda.DynamoDBEvents;
using Amazon.Lambda.Serialization.SystemTextJson;
using AWS.Lambda.Powertools.BatchProcessing;
[assembly: LambdaSerializer(typeof(DefaultLambdaJsonSerializer))]
namespace HelloWorld;
public class Customer
{
public string? CustomerId { get; set; }
public string? Name { get; set; }
public string? Email { get; set; }
public DateTime CreatedAt { get; set; }
}
internal class TypedDynamoDbRecordHandler : ITypedRecordHandler
{
public async Task HandleAsync(Customer customer, CancellationToken cancellationToken)
{
if (string.IsNullOrEmpty(customer.Email))
{
throw new ArgumentException("Customer email is required");
}
return await Task.FromResult(RecordHandlerResult.None);
}
}
public class Function
{
[BatchProcessor(TypedRecordHandler = typeof(TypedDynamoDbRecordHandler))]
public BatchItemFailuresResponse HandlerUsingTypedAttribute(DynamoDBEvent _)
{
return TypedDynamoDbStreamBatchProcessor.Result.BatchItemFailuresResponse;
}
}
```
# Conservare i record scartati per un'origine eventi DynamoDB in Lambda
La gestione degli errori per gli strumenti di mappatura dell'origine degli eventi DynamoDB dipende dal fatto che l'errore si verifichi prima che la funzione venga richiamata o durante la chiamata della funzione:
+ **Prima della chiamata:** se una mappatura dell'origine degli eventi Lambda non è in grado di richiamare la funzione a causa di limitazioni o altri problemi, riprova finché i record non scadono o superano l'età massima configurata nella mappatura dell'origine dell'evento (). [MaximumRecordAgeInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRecordAgeInSeconds)
+ **Durante la chiamata:** se la funzione viene richiamata ma restituisce un errore, Lambda riprova fino alla scadenza dei record, al superamento dell'età massima () o al raggiungimento della quota di tentativi configurata ([MaximumRecordAgeInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRecordAgeInSeconds)). [MaximumRetryAttempts](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRetryAttempts) Per gli errori di funzione, puoi anche configurare, che divide un batch non riuscito in due batch più piccoli [BisectBatchOnFunctionError](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-response-BisectBatchOnFunctionError), isolando i record non validi ed evitando i timeout. La divisione dei batch non consuma la quota di tentativi.
Se le misure di gestione degli errori non riescono, Lambda elimina i record e continua l'elaborazione dei batch dal flusso. Con le impostazioni predefinite, ciò significa che un record errato può bloccare l'elaborazione sullo shard interessata per un massimo di un giorno. Per evitare questa situazione, configura la mappatura dell'origine eventi della funzione con un numero ragionevole di tentativi e un'età massima dei record che sia adatta al caso d'uso.
## Configurazione delle destinazioni per le chiamate non riuscite
Per mantenere i record delle chiamate non riuscite allo strumento di mappatura dell'origine degli eventi, aggiungi una destinazione allo strumento di mappatura dell'origine degli eventi della funzione. Ogni record inviato alla destinazione è un documento JSON contenente i metadati relativi alla chiamata non riuscita. Per le destinazioni Amazon S3, Lambda invia insieme ai metadati anche l'intero record di invocazione. Puoi configurare qualsiasi argomento Amazon SNS, coda Amazon SQS, bucket Amazon S3 o Kafka come destinazione.
Con le destinazioni Amazon S3, puoi utilizzare la funzionalità [Notifiche eventi Amazon S3](https://docs.aws.amazon.com/) per ricevere notifiche quando gli oggetti vengono caricati nel bucket S3 di destinazione. Puoi anche configurare Notifiche eventi S3 per richiamare un'altra funzione Lambda per eseguire l'elaborazione automatica su batch non riusciti.
Il tuo ruolo di esecuzione deve avere le autorizzazioni per la destinazione:
+ **[Per una destinazione SQS: sqs: SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html)**
+ **[Per una destinazione SNS: sns:publish](https://docs.aws.amazon.com/sns/latest/api/API_Publish.html)**
+ **[Per una destinazione S3: s3: e [s3](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html): PutObject ListBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/ListObjectsV2.html)**
+ **[Per una destinazione Kafka: kafka-cluster: WriteData](https://docs.aws.amazon.com/msk/latest/developerguide/kafka-actions.html)**
È possibile configurare un argomento di Kafka come destinazione in caso di errore per le mappature delle sorgenti degli eventi Kafka. Quando Lambda non è in grado di elaborare i record dopo estenuanti tentativi o quando i record superano l'età massima, Lambda invia i record non riusciti all'argomento Kafka specificato per un'elaborazione successiva. Fai riferimento a [Usare un argomento di Kafka come destinazione in caso di errore](kafka-on-failure-destination.md).
[Se hai abilitato la crittografia con la tua chiave KMS per una destinazione S3, anche il ruolo di esecuzione della funzione deve avere l'autorizzazione a chiamare kms:. GenerateDataKey](https://docs.aws.amazon.com/kms/latest/APIReference/API_GenerateDataKey.html) Se la chiave KMS e la destinazione del bucket S3 si trovano in un account diverso dalla funzione Lambda e dal ruolo di esecuzione, configura la chiave KMS in modo che consideri attendibile il ruolo di esecuzione da consentire. kms: GenerateDataKey
Per configurare una destinazione in caso di errore tramite la console, completa i seguenti passaggi:
1. Aprire la pagina [Funzioni](https://console.aws.amazon.com/lambda/home#/functions) della console Lambda.
1. Scegliere una funzione.
1. In **Function overview (Panoramica delle funzioni)**, scegliere **Add destination (Aggiungi destinazione)**.
1. Per **Origine**, scegli **Chiamata allo strumento di mappatura dell'origine degli eventi**.
1. Per **Strumento di mappatura dell'origine degli eventi**, scegli un'origine dell'evento configurata per questa funzione.
1. Per **Condizione**, seleziona **In caso di errore**. Per le chiamate allo strumento di mappatura dell'origine degli eventi, questa è l'unica condizione accettata.
1. Per **Tipo di destinazione**, scegli il tipo di destinazione a cui Lambda deve inviare i record di chiamata.
1. Per **Destination (Destinazione)**, scegliere una risorsa.
1. Scegli **Save** (Salva).
Puoi anche configurare una destinazione in caso di errore utilizzando (). AWS Command Line Interface AWS CLI Ad esempio, il [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html)comando seguente aggiunge una mappatura dell'origine degli eventi con una destinazione SQS in caso di errore a: `MyFunction`
```
aws lambda create-event-source-mapping \
--function-name "MyFunction" \
--event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525 \
--destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:us-east-1:123456789012:dest-queue"}}'
```
Il [update-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html)comando seguente aggiorna una mappatura dell'origine degli eventi per inviare i record di chiamata non riuscita a una destinazione SNS dopo due tentativi o se i record risalgono a più di un'ora.
```
aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--maximum-retry-attempts 2 \
--maximum-record-age-in-seconds 3600 \
--destination-config '{"OnFailure": {"Destination": "arn:aws:sns:us-east-1:123456789012:dest-topic"}}'
```
Le impostazioni aggiornate sono applicate in modo asincrono e non sono riflesse nell'output fino al completamento del processo. Utilizza il comando [get-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html) per visualizzare lo stato corrente.
Per rimuovere una destinazione, fornisci una stringa vuota come argomento del parametro `destination-config`:
```
aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--destination-config '{"OnFailure": {"Destination": ""}}'
```
### Best practice di sicurezza per destinazioni Amazon S3
L'eliminazione di un bucket S3 configurato come destinazione senza rimuovere la destinazione dalla configurazione della funzione può creare un rischio per la sicurezza. Se un altro utente conosce il nome del bucket di destinazione, può ricreare il bucket nel proprio Account AWS. I record delle invocazioni non riuscite verranno inviati al relativo bucket, esponendo potenzialmente i dati della tua funzione.
**avvertimento**
Per garantire che i record di invocazione della tua funzione non possano essere inviati a un bucket S3 in un altro Account AWS, aggiungi una condizione al ruolo di esecuzione della funzione che limiti le autorizzazioni ai bucket del tuo account. `s3:PutObject`
Di seguito viene illustrato un esempio di policy IAM che limita le autorizzazioni `s3:PutObject` della funzione ai bucket presenti nell'account. Questa policy fornisce inoltre a Lambda l'autorizzazione `s3:ListBucket` necessaria per utilizzare un bucket S3 come destinazione.
```
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "S3BucketResourceAccountWrite",
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:ListBucket"
],
"Resource": [
"arn:aws:s3:::*/*",
"arn:aws:s3:::*"
],
"Condition": {
"StringEquals": {
"s3:ResourceAccount": "111122223333"
}
}
}
]
}
```
Per aggiungere una politica di autorizzazioni al ruolo di esecuzione della tua funzione utilizzando Console di gestione AWS o AWS CLI, consulta le istruzioni nelle seguenti procedure:
------
#### [ Console ]
**Per aggiungere una policy di autorizzazioni al ruolo di esecuzione di una funzione (console)**
1. Aprire la pagina [Funzioni](https://console.aws.amazon.com/lambda/home#/functions) della console Lambda.
1. Seleziona la funzione Lambda di cui si desidera modificare il ruolo di esecuzione.
1. Nella scheda **Configurazione** scegli **Autorizzazioni**.
1. Nella scheda **Ruolo di esecuzione**, seleziona il **nome del ruolo** della funzione per aprire la pagina della console IAM del ruolo.
1. Aggiungi una policy di autorizzazioni di base al ruolo completando le seguenti operazioni:
1. Nel riquadro **Policy di autorizzazioni**, scegli **Aggiungi autorizzazioni**, poi **Crea policy in linea**.
1. Nell'**editor delle policy**, seleziona **JSON**.
1. Incolla la policy che desideri aggiungere nell'editor (sostituendo il codice JSON esistente), quindi scegli **Avanti**.
1. In **Dettagli della policy**, specifica un **nome per la policy**.
1. Scegli **Crea policy**.
------
#### [ AWS CLI ]
**Per aggiungere una policy di autorizzazioni al ruolo di esecuzione di una funzione (CLI)**
1. Crea un documento di policy JSON con le autorizzazioni richieste e salvalo in una directory locale.
1. Utilizza il comando della CLI `put-role-policy` di IAM per aggiungere le autorizzazioni per il ruolo di esecuzione di una funzione. Esegui il comando seguente dalla directory in cui hai salvato il documento di policy JSON e sostituisci il nome del ruolo, il nome della policy e il documento di policy con i tuoi valori.
```
aws iam put-role-policy \
--role-name my_lambda_role \
--policy-name LambdaS3DestinationPolicy \
--policy-document file://my_policy.json
```
------
### Record di invocazione Amazon SNS e Amazon SQS di esempio
L'esempio seguente mostra un record di chiamata che Lambda invia a una destinazione SQS o SNS per un flusso DynamoDB.
```
{
"requestContext": {
"requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
"functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction",
"condition": "RetryAttemptsExhausted",
"approximateInvokeCount": 1
},
"responseContext": {
"statusCode": 200,
"executedVersion": "$LATEST",
"functionError": "Unhandled"
},
"version": "1.0",
"timestamp": "2019-11-14T00:13:49.717Z",
"DDBStreamBatchInfo": {
"shardId": "shardId-00000001573689847184-864758bb",
"startSequenceNumber": "800000000003126276362",
"endSequenceNumber": "800000000003126276362",
"approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z",
"approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z",
"batchSize": 1,
"streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388"
}
}
```
È possibile utilizzare queste informazioni per recuperare i record interessati dal flusso per la risoluzione dei problemi. I record effettivi non sono inclusi, pertanto è necessario elaborare questo record e recuperarli dal flusso prima che scadano e vadano persi.
### Record di invocazione Amazon S3 di esempio
L'esempio seguente mostra un record di invocazione che Lambda invia a un bucket S3 per un flusso DynamoDB. Oltre a tutti i campi dell'esempio precedente per le destinazioni SQS e SNS, il campo `payload` contiene il record di chiamata originale come stringa JSON con escape.
```
{
"requestContext": {
"requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81",
"functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction",
"condition": "RetryAttemptsExhausted",
"approximateInvokeCount": 1
},
"responseContext": {
"statusCode": 200,
"executedVersion": "$LATEST",
"functionError": "Unhandled"
},
"version": "1.0",
"timestamp": "2019-11-14T00:13:49.717Z",
"DDBStreamBatchInfo": {
"shardId": "shardId-00000001573689847184-864758bb",
"startSequenceNumber": "800000000003126276362",
"endSequenceNumber": "800000000003126276362",
"approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z",
"approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z",
"batchSize": 1,
"streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388"
},
"payload": "" // Only available in S3
}
```
L'oggetto S3 contenente il record di invocazione utilizza la seguente convenzione di denominazione:
```
aws/lambda///YYYY/MM/DD/YYYY-MM-DDTHH.MM.SS-
```
# Implementazione dell'elaborazione del flusso DynamoDB stateful in Lambda
Le funzioni Lambda possono eseguire applicazioni di elaborazione di flussi continui. Un flusso rappresenta dati illimitati che scorrono continuamente attraverso l'applicazione. Per analizzare le informazioni da questo input di aggiornamento continuo, è possibile associare i record inclusi utilizzando una finestra definita in termini di tempo.
Le finestre a cascata sono finestre temporali distinte che si aprono e si chiudono a intervalli regolari. Per impostazione predefinita, le invocazioni Lambda sono senza stato: non è possibile utilizzarle per l'elaborazione dei dati tra più invocazioni continue senza un database esterno. Tuttavia, con la finestra a cascata, è possibile mantenere il proprio stato tra le invocazioni. Questo stato contiene il risultato aggregato dei messaggi precedentemente elaborati per la finestra corrente. Lo stato può essere un massimo di 1 MB per shard. Se supera quella dimensione, Lambda termina la finestra in anticipo.
Ogni record in un flusso appartiene a una finestra specifica. Lambda elaborerà ogni record almeno una volta, ma non garantisce che ogni record venga elaborato una sola volta. In rari casi, come nel caso della gestione degli errori, alcuni record potrebbero essere elaborati più di una volta. La prima volta i record vengono sempre elaborati in ordine. Se i record vengono elaborati più di una volta, possono essere elaborati fuori ordine.
## Aggregazione ed elaborazione
La funzione gestita dall'utente viene richiamata sia per l'aggregazione che per l'elaborazione dei risultati finali di tale aggregazione. Lambda aggrega tutti i record ricevuti nella finestra. È possibile ricevere questi record in più batch, ciascuno come richiamo separato. Ogni richiamo riceve uno stato. Pertanto, quando si utilizzano finestre a cascata, la risposta della funzione Lambda deve contenere una proprietà di `state`. Se la risposta non contiene una proprietà di `state`, Lambda la considera un'invocazione non riuscita. Per soddisfare questa condizione, la funzione può restituire un oggetto `TimeWindowEventResponse`, che presenta la seguente forma JSON:
**Example `TimeWindowEventResponse` valori**
```
{
"state": {
"1": 282,
"2": 715
},
"batchItemFailures": []
}
```
**Nota**
Per le funzioni Java, si consiglia di utilizzare una `Map` per rappresentare lo stato.
Alla fine della finestra, il flag `isFinalInvokeForWindow` è impostato `true` per indicare che questo è lo stato finale e che è pronto per l'elaborazione. Dopo l'elaborazione, la finestra viene completata e il richiamo finale viene completato e quindi lo stato viene eliminato.
Al termine della finestra, Lambda utilizza l'elaborazione finale per le operazioni sui risultati dell'aggregazione. L'elaborazione finale viene richiamata in modo sincrono. Dopo il richiamo riuscito, la funzione controlla il numero di sequenza e l'elaborazione del flusso continua. Se il richiamo non ha esito positivo, la funzione Lambda sospende l'ulteriore elaborazione fino a quando non viene eseguito correttamente il richiamo.
**Example DynamodbTimeWindowEvent**
```
{
"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"
}
],
"window": {
"start": "2020-07-30T17:00:00Z",
"end": "2020-07-30T17:05:00Z"
},
"state": {
"1": "state1"
},
"shardId": "shard123456789",
"eventSourceARN": "stream-ARN",
"isFinalInvokeForWindow": false,
"isWindowTerminatedEarly": false
}
```
## Configurazione
È possibile configurare le finestre a cascata quando si crea o si aggiorna un mapping di origini di eventi. Per configurare una finestra rotante, specificate la finestra in secondi (). [TumblingWindowInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-TumblingWindowInSeconds) Il seguente comando di esempio AWS Command Line Interface (AWS CLI) crea una mappatura della sorgente degli eventi in streaming con una finestra di rotazione di 120 secondi. La funzione Lambda definita per l'aggregazione e l'elaborazione è denominata `tumbling-window-example-function`.
```
aws lambda create-event-source-mapping \
--event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525 \
--function-name tumbling-window-example-function \
--starting-position TRIM_HORIZON \
--tumbling-window-in-seconds 120
```
Lambda determina i limiti delle finestre a cascata in base al momento in cui i record sono stati inseriti nel flusso. Tutti i record dispongono di un timestamp approssimativo che Lambda utilizza nelle determinazioni dei limiti.
Le aggregazioni delle finestre a cascata non supportano il risharding. Quando lo shard termina, Lambda considera la finestra chiusa e gli shard secondari iniziano la propria finestra in uno stato nuovo.
Le finestre a cascata supportano completamente le policy di ripetizione dei tentativi esistenti `maxRetryAttempts` e `maxRecordAge`.
**Example Handler.py: aggregazione ed elaborazione**
La seguente funzione Python mostra come aggregare e quindi elaborare lo stato finale:
```
def lambda_handler(event, context):
print('Incoming event: ', event)
print('Incoming state: ', event['state'])
#Check if this is the end of the window to either aggregate or process.
if event['isFinalInvokeForWindow']:
# logic to handle final state of the window
print('Destination invoke')
else:
print('Aggregate invoke')
#Check for early terminations
if event['isWindowTerminatedEarly']:
print('Window terminated early')
#Aggregation logic
state = event['state']
for record in event['Records']:
state[record['dynamodb']['NewImage']['Id']] = state.get(record['dynamodb']['NewImage']['Id'], 0) + 1
print('Returning state: ', state)
return {'state': state}
```
# Parametri Lambda per gli strumenti di mappatura dell'origine degli eventi di Amazon DynamoDB
Tutti i tipi di sorgenti di eventi Lambda condividono le stesse operazioni [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html)e quelle dell'[UpdateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html)API. Tuttavia, solo alcuni dei parametri si applicano a DynamoDB Streams.
| Parametro | Obbligatorio | Predefinito | Note |
| --- | --- | --- | --- |
| BatchSize | N | 100 | Massimo: 10.000. |
| BisectBatchOnFunctionError | N | false | nessuno |
| DestinationConfig | N | N/D | Una destinazione della coda standard di Amazon SQS o di un argomento standard di Amazon SNS per i record scartati. |
| Abilitato | N | true | nessuno |
| EventSourceArn | Y | N/D | L'ARN del flusso dei dati o di un consumer di flusso. |
| FilterCriteria | N | N/D | [Controllare gli eventi che Lambda invia alla funzione](invocation-eventfiltering.md) |
| FunctionName | Y | N/D | nessuno |
| FunctionResponseTypes | N | N/D | Per consentire alla funzione di segnalare errori specifici in un batch, includi il valore `ReportBatchItemFailures` in `FunctionResponseTypes`. Per ulteriori informazioni, consulta [Configurazione della risposta batch parziale con DynamoDB e Lambda](services-ddb-batchfailurereporting.md). |
| MaximumBatchingWindowInSeconds | N | 0 | nessuno |
| MaximumRecordAgeInSeconds | N | -1 | -1 sta per infinito: i record non riusciti vengono ripetuti fino alla scadenza del record. Il [limite di conservazione dei dati per i flussi DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.DataRetention) è di 24 ore. Minimo: -1 Massimo: 604.800 |
| MaximumRetryAttempts | N | -1 | -1 sta per infinito: i record non riusciti vengono ripetuti fino alla scadenza del record Minimo: 0 Massimo: 10.000. |
| ParallelizationFactor | N | 1 | Maximum: 10 |
| StartingPosition | Y | N/D | TRIM\$1HORIZON o LATEST |
| TumblingWindowInSeconds | N | N/D | Minimo: 0 Massimo: 900 |
# Utilizzo del filtro eventi con un'origine eventi DynamoDB
Puoi utilizzare il filtraggio degli eventi per controllare quali record di un flusso o di una coda Lambda invia alla funzione. Per informazioni generali sul funzionamento del filtraggio eventi, consulta [Controllare gli eventi che Lambda invia alla funzione](invocation-eventfiltering.md).
In questa sezione viene descritto il filtraggio degli eventi per le origini eventi DynamoDB.
**Nota**
Le mappature delle sorgenti degli eventi DynamoDB supportano solo il filtraggio sulla chiave. `dynamodb`
**Topics**
+ [
## Evento DynamoDB
](#filtering-ddb)
+ [
## Filtraggio con attributi di tabella
](#filtering-ddb-attributes)
+ [
## Filtraggio con espressioni booleane
](#filtering-ddb-boolean)
+ [
## Utilizzo dell'operatore Exists
](#filtering-ddb-exists)
+ [
## Formato JSON per il filtro DynamoDB
](#filtering-ddb-JSON-format)
## Evento DynamoDB
Supponiamo di avere una tabella DynamoDB con la chiave primaria `CustomerName` e gli attributi `AccountManager` e `PaymentTerms`. Di seguito è riportato un esempio di record dal flusso della tabella DynamoDB.
```
{
"eventID": "1",
"eventVersion": "1.0",
"dynamodb": {
"ApproximateCreationDateTime": "1678831218.0",
"Keys": {
"CustomerName": {
"S": "AnyCompany Industries"
}
},
"NewImage": {
"AccountManager": {
"S": "Pat Candella"
},
"PaymentTerms": {
"S": "60 days"
},
"CustomerName": {
"S": "AnyCompany Industries"
}
},
"SequenceNumber": "111",
"SizeBytes": 26,
"StreamViewType": "NEW_IMAGE"
}
}
```
Per filtrare in base ai valori della chiave e degli attributi nella tabella DynamoDB, utilizza la chiave `dynamodb` nel record. Le seguenti sezioni forniscono esempi per diversi tipi di filtri.
### Filtraggio con chiavi di tabella
Supponiamo che tu voglia che la tua funzione elabori solo i record in cui la chiave primaria `CustomerName` è "AnyCompany Industries". L'oggetto `FilterCriteria` dovrebbe avere la struttura seguente.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"
}
]
}
```
Per una maggiore chiarezza, ecco il valore del `Pattern` del filtro espanso in JSON semplice.
```
{
"dynamodb": {
"Keys": {
"CustomerName": {
"S": [ "AnyCompany Industries" ]
}
}
}
}
```
Puoi aggiungere il filtro utilizzando la console, la AWS CLI o un modello AWS SAM.
------
#### [ Console ]
Per aggiungere questo filtro utilizzando la console, segui le istruzioni riportate in [Collegamento dei criteri di filtro a una mappatura dell'origine evento (console)](invocation-eventfiltering.md#filtering-console) e inserisci la seguente stringa per i **criteri di filtraggio**.
```
{ "dynamodb" : { "Keys" : { "CustomerName" : { "S" : [ "AnyCompany Industries" ] } } } }
```
------
#### [ AWS CLI ]
Per creare una nuova mappatura dell'origine degli eventi con questi criteri di filtraggio utilizzando l'AWS Command Line Interface (AWS CLI), esegui il comando seguente.
```
aws lambda create-event-source-mapping \
--function-name my-function \
--event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"}]}'
```
Per aggiungere questi criteri di filtraggio a una mappatura dell'origine degli eventi esistente, esegui il comando seguente.
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"}]}'
```
------
#### [ AWS SAM ]
Per aggiungere questo filtro utilizzando AWS SAM, aggiungi il seguente frammento al modello YAML dell'origine degli eventi.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "Keys" : { "CustomerName" : { "S" : [ "AnyCompany Industries" ] } } } }'
```
------
## Filtraggio con attributi di tabella
Con DynamoDB, puoi anche utilizzare le chiavi `NewImage` e `OldImage` per filtrare i valori degli attributi. Supponiamo di voler filtrare i record in cui l'attributo `AccountManager` nell'ultima immagine della tabella è "Pat Candella" o "Shirley Rodriguez". L'oggetto `FilterCriteria` dovrebbe avere la struttura seguente.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\", \"Shirley Rodriguez\" ] } } } }"
}
]
}
```
Per una maggiore chiarezza, ecco il valore del `Pattern` del filtro espanso in JSON semplice.
```
{
"dynamodb": {
"NewImage": {
"AccountManager": {
"S": [ "Pat Candella", "Shirley Rodriguez" ]
}
}
}
}
```
Puoi aggiungere il filtro utilizzando la console, la AWS CLI o un modello AWS SAM.
------
#### [ Console ]
Per aggiungere questo filtro utilizzando la console, segui le istruzioni riportate in [Collegamento dei criteri di filtro a una mappatura dell'origine evento (console)](invocation-eventfiltering.md#filtering-console) e inserisci la seguente stringa per i **criteri di filtraggio**.
```
{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella", "Shirley Rodriguez" ] } } } }
```
------
#### [ AWS CLI ]
Per creare una nuova mappatura dell'origine degli eventi con questi criteri di filtraggio utilizzando l'AWS Command Line Interface (AWS CLI), esegui il comando seguente.
```
aws lambda create-event-source-mapping \
--function-name my-function \
--event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\", \"Shirley Rodriguez\" ] } } } }"}]}'
```
Per aggiungere questi criteri di filtraggio a una mappatura dell'origine degli eventi esistente, esegui il comando seguente.
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\", \"Shirley Rodriguez\" ] } } } }"}]}'
```
------
#### [ AWS SAM ]
Per aggiungere questo filtro utilizzando AWS SAM, aggiungi il seguente frammento al modello YAML dell'origine degli eventi.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella", "Shirley Rodriguez" ] } } } }'
```
------
## Filtraggio con espressioni booleane
È inoltre possibile creare filtri utilizzando espressioni booleane AND. Queste espressioni possono includere sia i parametri chiave sia quelli degli attributi della tabella. Supponiamo che tu voglia filtrare i record in cui il valore di `NewImage` di `AccountManager` è "Pat Candella" e il valore di `OldImage` è "Terry Whitlock". L'oggetto `FilterCriteria` dovrebbe avere la struttura seguente.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\" ] } } } , \"dynamodb\" : { \"OldImage\" : { \"AccountManager\" : { \"S\" : [ \"Terry Whitlock\" ] } } } }"
}
]
}
```
Per una maggiore chiarezza, ecco il valore del `Pattern` del filtro espanso in JSON semplice.
```
{
"dynamodb" : {
"NewImage" : {
"AccountManager" : {
"S" : [
"Pat Candella"
]
}
}
},
"dynamodb": {
"OldImage": {
"AccountManager": {
"S": [
"Terry Whitlock"
]
}
}
}
}
```
Puoi aggiungere il filtro utilizzando la console, la AWS CLI o un modello AWS SAM.
------
#### [ Console ]
Per aggiungere questo filtro utilizzando la console, segui le istruzioni riportate in [Collegamento dei criteri di filtro a una mappatura dell'origine evento (console)](invocation-eventfiltering.md#filtering-console) e inserisci la seguente stringa per i **criteri di filtraggio**.
```
{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella" ] } } } , "dynamodb" : { "OldImage" : { "AccountManager" : { "S" : [ "Terry Whitlock" ] } } } }
```
------
#### [ AWS CLI ]
Per creare una nuova mappatura dell'origine degli eventi con questi criteri di filtraggio utilizzando l'AWS Command Line Interface (AWS CLI), esegui il comando seguente.
```
aws lambda create-event-source-mapping \
--function-name my-function \
--event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\" ] } } } , \"dynamodb\" : { \"OldImage\" : { \"AccountManager\" : { \"S\" : [ \"Terry Whitlock\" ] } } } } "}]}'
```
Per aggiungere questi criteri di filtraggio a una mappatura dell'origine degli eventi esistente, esegui il comando seguente.
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\" ] } } } , \"dynamodb\" : { \"OldImage\" : { \"AccountManager\" : { \"S\" : [ \"Terry Whitlock\" ] } } } } "}]}'
```
------
#### [ AWS SAM ]
Per aggiungere questo filtro utilizzando AWS SAM, aggiungi il seguente frammento al modello YAML dell'origine degli eventi.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella" ] } } } , "dynamodb" : { "OldImage" : { "AccountManager" : { "S" : [ "Terry Whitlock" ] } } } }'
```
------
**Nota**
Il filtraggio degli eventi DynamoDB non supporta l'uso di operatori numerici (uguali numerici e intervallo numerico). Anche se gli elementi della tabella sono memorizzati come numeri, questi parametri vengono convertiti in stringhe nell'oggetto record JSON.
## Utilizzo dell'operatore Exists
A causa del modo in cui sono strutturati gli oggetti evento JSON di DynamoDB, l'utilizzo dell'operatore Exists richiede particolare attenzione. L'operatore Exists funziona solo sui nodi foglia nel JSON dell'evento, quindi se il modello di filtro utilizza Exists per testare un nodo intermedio, non funzionerà. Considera il seguente elemento della tabella DynamoDB:
```
{
"UserID": {"S": "12345"},
"Name": {"S": "John Doe"},
"Organizations": {"L": [
{"S":"Sales"},
{"S":"Marketing"},
{"S":"Support"}
]
}
}
```
Potresti voler creare uno schema di filtro come il seguente per verificare la presenza di eventi contenenti `"Organizations"`:
```
{ "dynamodb" : { "NewImage" : { "Organizations" : [ { "exists": true } ] } } }
```
Tuttavia, questo modello di filtro non restituirebbe mai una corrispondenza perché `"Organizations"` non è un nodo foglia. L'esempio seguente mostra come utilizzare correttamente l'operatore Exists per creare lo schema di filtro desiderato:
```
{ "dynamodb" : { "NewImage" : {"Organizations": {"L": {"S": [ {"exists": true } ] } } } } }
```
## Formato JSON per il filtro DynamoDB
Per filtrare correttamente gli eventi da origini DynamoDB, sia il campo dati sia i criteri di filtraggio per il campo dati (`dynamodb`) devono essere in un formato JSON valido. Se uno dei due campi non è in un formato JSON valido, Lambda rilascia il messaggio o genera un'eccezione. La tabella seguente riepiloga il comportamento specifico:
| Formato dei dati in entrata | Formato del modello di filtro per le proprietà di dati | Operazione risultante |
| --- | --- | --- |
| JSON valido | JSON valido | Filtri Lambda in base ai criteri di filtro. |
| JSON valido | Nessun modello di filtro per le proprietà dei dati | Filtri Lambda (solo sulle altre proprietà dei metadati) in base ai criteri di filtro. |
| JSON valido | Non-JSON | Lambda genera un'eccezione al momento della creazione o dell'aggiornamento della mappatura dell'origine evento. Il modello di filtro per le proprietà dei dati deve essere in un formato JSON valido. |
| Non-JSON | JSON valido | Lambda rilascia il registro. |
| Non-JSON | Nessun modello di filtro per le proprietà dei dati | Filtri Lambda (solo sulle altre proprietà dei metadati) in base ai criteri di filtro. |
| Non-JSON | Non-JSON | Lambda genera un'eccezione al momento della creazione o dell'aggiornamento della mappatura dell'origine evento. Il modello di filtro per le proprietà dei dati deve essere in un formato JSON valido. |
# Tutorial: Utilizzo AWS Lambda con flussi Amazon DynamoDB
In questo tutorial creerai una funzione Lambda per consumare eventi da un flusso Amazon DynamoDB.
## Prerequisiti
### Installa il AWS Command Line Interface
Se non l'hai ancora installato AWS Command Line Interface, segui i passaggi indicati in [Installazione o aggiornamento della versione più recente di AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) per installarlo.
Per eseguire i comandi nel tutorial, sono necessari un terminale a riga di comando o una shell (interprete di comandi). In Linux e macOS, utilizza la shell (interprete di comandi) e il gestore pacchetti preferiti.
**Nota**
Su Windows, alcuni comandi della CLI Bash utilizzati comunemente con Lambda (ad esempio, `zip`) non sono supportati dai terminali integrati del sistema operativo. Per ottenere una versione integrata su Windows di Ubuntu e Bash, [installa il sottosistema Windows per Linux](https://docs.microsoft.com/en-us/windows/wsl/install-win10).
## Creazione del ruolo di esecuzione
Crea il [ruolo di esecuzione](lambda-intro-execution-role.md) che autorizza la funzione ad accedere alle AWS risorse.
**Per creare un ruolo di esecuzione**
1. Apri la pagina [Ruoli](https://console.aws.amazon.com/iam/home#/roles) nella console IAM.
1. Scegliere **Crea ruolo**.
1. Creare un ruolo con le seguenti proprietà.
+ **Entità attendibile** – Lambda.
+ **Autorizzazioni** — **AWSLambdaDBExecutionDynamo** Role.
+ **Nome ruolo** – **lambda-dynamodb-role**.
Il **DBExecutionruolo AWSLambda Dynamo** dispone delle autorizzazioni necessarie alla funzione per leggere gli elementi da DynamoDB e scrivere i log nei log. CloudWatch
## Creazione della funzione
Crea una funzione Lambda che elabora gli eventi DynamoDB. Il codice della funzione scrive alcuni dei dati degli eventi in entrata in Logs. CloudWatch
------
#### [ .NET ]
**SDK per .NET**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite .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 per Go V2**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite 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 per Java 2.x**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite 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 {
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 per JavaScript (v3)**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo di un evento DynamoDB con Lambda utilizzando. 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)}`);
};
```
Consumo di un evento DynamoDB con Lambda utilizzando. 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 per PHP**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite PHP.
```
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 per Python (Boto3)**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite 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 per Ruby**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite 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 per Rust**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Utilizzo di un evento DynamoDB con Lambda tramite 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) ->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(())
}
```
------
**Per creare la funzione**
1. Copiare il codice di esempio in un file denominato `example.js`.
1. Crea un pacchetto di implementazione.
```
zip function.zip example.js
```
1. Creare una funzione Lambda con il comando `create-function`.
```
aws lambda create-function --function-name ProcessDynamoDBRecords \
--zip-file fileb://function.zip --handler example.handler --runtime nodejs24.x \
--role arn:aws:iam::111122223333:role/lambda-dynamodb-role
```
## Test della funzione Lambda
In questo passaggio, si richiama manualmente la funzione Lambda utilizzando il comando `invoke` AWS Lambda CLI e il seguente evento DynamoDB di esempio. Copia quanto riportato di seguito in un file denominato `input.txt`.
**Example 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"
}
]
}
```
Eseguire il seguente comando `invoke`.
```
aws lambda invoke --function-name ProcessDynamoDBRecords \
--cli-binary-format raw-in-base64-out \
--payload file://input.txt outputfile.txt
```
L'**cli-binary-format**opzione è obbligatoria se si utilizza la versione 2. AWS CLI Per rendere questa impostazione come predefinita, esegui `aws configure set cli-binary-format raw-in-base64-out`. Per ulteriori informazioni, consulta la pagina [AWS CLI supported global command line options](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-options.html#cli-configure-options-list) nella *Guida per l'utente di AWS Command Line Interface versione 2*.
La funzione restituisce la stringa `message` nel corpo della risposta.
Verificare l'output nel file `outputfile.txt`.
## Creazione di una tabella DynamoDB con un flusso abilitato
Crea una tabella Amazon DynamoDB con un flusso abilitato.
**Per creare una tabella DynamoDB**
1. Aprire la [console DynamoDB](https://console.aws.amazon.com/dynamodb).
1. Scegliere **Create table (Crea tabella)**.
1. Creare una tabella con le impostazioni seguenti.
+ **Table name (Nome tabella)** – **lambda-dynamodb-stream**
+ **Primary key (Chiave primaria)** – **id** (string)
1. Scegliere **Create (Crea)**.
**Per abilitare i flussi**
1. Aprire la [console DynamoDB](https://console.aws.amazon.com/dynamodb).
1. Scegliere **Tables (Tabelle)**.
1. Seleziona la tabella **lambda-dynamodb-stream**.
1. In **Exports and streams (Esportazioni e flussi)**, scegliere **DynamDB stream details (Dettagli del flusso di Dynamo DB)**.
1. Scegliere **Turn On (Attiva)**.
1. Per **Tipo di visualizzazione**, scegli **Solo attributi chiave**.
1. Scegli **Attiva il flusso**.
Prendi nota dell'ARN del flusso. Questa operazione è necessaria nella fase successiva quando si associa il flusso alla funzione Lambda. Per ulteriori informazioni sull'attivazione dei flussi, consultare [Acquisizione dell'attività sulla tabella tramite DynamoDB Streams](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html).
## Aggiungi una fonte di eventi in AWS Lambda
Creare una mappatura dell'origine eventi in AWS Lambda. Questa mappatura dell'origine eventi associa il flusso DynamoDB alla funzione Lambda. Dopo aver creato questa mappatura delle sorgenti degli eventi, AWS Lambda inizia il polling dello stream.
Eseguire il seguente comando AWS CLI `create-event-source-mapping`. Dopo l'esecuzione del comando, annotare l'UUID. Questo UUID è necessario per fare riferimento alla mappatura delle origini eventi nei comandi, ad esempio quando si elimina tale mappatura.
```
aws lambda create-event-source-mapping --function-name ProcessDynamoDBRecords \
--batch-size 100 --starting-position LATEST --event-source DynamoDB-stream-arn
```
Questa procedura crea una mappatura tra il flusso DynamoDB specificato e la funzione Lambda. È possibile associare un flusso DynamoDB a più funzioni Lambda e associare la stessa funzione Lambda a più flussi. Tuttavia, le funzioni Lambda condivideranno il throughput di lettura per il flusso che condividono.
Mediante l'esecuzione del comando riportato di seguito è possibile ottenere l'elenco di mappature delle origini eventi.
```
aws lambda list-event-source-mappings
```
L'elenco restituisce tutte le mappature delle origini eventi create e per ciascuna mappatura mostra, tra l'altro, il valore `LastProcessingResult`. Questo campo è utilizzato per fornire un messaggio informativo nel caso in cui vengano riscontrati problemi. Valori come `No records processed` (indica che non AWS Lambda è stato avviato il polling o che non ci sono record nello stream) e `OK` (indica che i record dallo stream sono AWS Lambda stati letti correttamente e che è stata richiamata la funzione Lambda) indicano che non ci sono problemi. Se vengono riscontrati problemi, si riceve un messaggio di errore.
Se disponi di una grande quantità di mappature delle origini eventi, utilizza il parametro funzione-nome per restringere i risultati.
```
aws lambda list-event-source-mappings --function-name ProcessDynamoDBRecords
```
## Eseguire il test della configurazione
Metti alla prova l'esperienza. end-to-end Quando vengono eseguiti gli aggiornamenti delle tabelle, DynamoDB scrive record di eventi nel flusso. Quando AWS Lambda esegue il polling del flusso, rileva nuovi record nel flusso e richiama la funzione Lambda per proprio conto passando eventi alla funzione.
1. Nella console di DynamoDB, aggiungi, aggiorna ed elimina elementi dalla tabella. DynamoDB scrive record di queste operazioni nel flusso.
1. AWS Lambda esegue il polling dello stream e quando rileva aggiornamenti allo stream richiama la funzione Lambda passando i dati degli eventi che trova nello stream.
1. La tua funzione viene eseguita e crea registri in Amazon CloudWatch. Puoi verificare i log riportati nella CloudWatch console Amazon.
## Fasi successive
Questo tutorial ti ha mostrato le basi dell'elaborazione degli eventi di flusso DynamoDB con Lambda. Per i carichi di lavoro di produzione, prendi in considerazione l'implementazione della logica di risposta parziale in batch per gestire gli errori dei singoli record in modo più efficiente. [L'utilità di elaborazione batch](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) di Powertools for AWS Lambda è disponibile in Python TypeScript, .NET e Java e fornisce una soluzione robusta per questo, gestendo automaticamente la complessità delle risposte parziali in batch e riducendo il numero di tentativi per i record elaborati correttamente.
## Pulizia delle risorse
Ora è possibile eliminare le risorse create per questo tutorial, a meno che non si voglia conservarle. Eliminando AWS le risorse che non utilizzi più, eviti addebiti inutili ai tuoi. Account AWS
**Per eliminare la funzione Lambda**
1. Aprire la pagina [Functions (Funzioni)](https://console.aws.amazon.com/lambda/home#/functions) della console Lambda.
1. Selezionare la funzione creata.
1. Scegliere **Operazioni**, **Elimina**.
1. Inserisci **confirm** nel campo di immissione del testo, quindi scegli **Elimina**.
**Come eliminare il ruolo di esecuzione**
1. Aprire la pagina [Ruoli](https://console.aws.amazon.com/iam/home#/roles) della console IAM.
1. Selezionare il ruolo di esecuzione creato.
1. Scegliere **Elimina**.
1. Inserisci il nome del ruolo nel campo di immissione testo e seleziona **Delete** (Elimina).
**Per eliminare la tabella DynamoDB**
1. Aprire la pagina [Tables (Tabelle)](https://console.aws.amazon.com//dynamodb/home#tables:) della console DynamoDB.
1. Selezionare la tabella creata.
1. Scegliere **Delete (Elimina)**.
1. Immettere **delete** nella casella di testo.
1. Seleziona **Delete Table (Elimina tabella)**.