Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Utilisation AWS Lambda avec Amazon DynamoDB
**Note**
Si vous souhaitez envoyer des données à une cible autre qu'une fonction Lambda ou enrichir les données avant de les envoyer, consultez [Amazon EventBridge ](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) Pipes.
Vous pouvez utiliser une AWS Lambda fonction pour traiter les enregistrements d'un flux [Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html). DynamoDB Streams vous permet de déclencher une fonction Lambda pour effectuer un travail supplémentaire chaque fois qu’une table DynamoDB est mise à jour.
Lorsque vous traitez des flux DynamoDB, vous devez implémenter une logique de réponse partielle de lot afin d’éviter le retraitement d’enregistrements traités avec succès si d’autres enregistrements du lot échouent. L'[utilitaire Batch Processor](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) de Powertools for AWS Lambda est disponible en Python TypeScript, .NET et Java et simplifie cette implémentation en gérant automatiquement la logique de réponse partielle par lots, en réduisant le temps de développement et en améliorant la fiabilité.
**Topics**
+ [Flux d’interrogation et de mise en lots](#dynamodb-polling-and-batching)
+ [Positions de départ des interrogations et des flux](#dyanmo-db-stream-poll)
+ [Lecteurs simultanés d’une partition dans DynamoDB Streams](#events-dynamodb-simultaneous-readers)
+ [Exemple d’évènement](#events-sample-dynamodb)
+ [Traiter les enregistrements DynamoDB avec Lambda](services-dynamodb-eventsourcemapping.md)
+ [Configuration d’une réponse par lots partielle avec DynamoDB et Lambda](services-ddb-batchfailurereporting.md)
+ [Retenir les enregistrements ignorés pour une source d’événement DynamoDB dans Lambda](services-dynamodb-errors.md)
+ [Implémentation du traitement des flux DynamoDB avec état dans Lambda](services-ddb-windows.md)
+ [Paramètres Lambda pour les mappages des sources d’événement Amazon DynamoDB](services-ddb-params.md)
+ [Utilisation du filtrage des événements avec une source d’événement DynamoDB](with-ddb-filtering.md)
+ [Tutoriel : Utilisation AWS Lambda avec les flux Amazon DynamoDB](with-ddb-example.md)
## Flux d’interrogation et de mise en lots
Lambda interroge les partitions de votre flux DynamoDB en quête d’enregistrements à une fréquence de base de quatre fois par seconde. Lorsque des enregistrements sont disponibles, Lambda invoque votre fonction et attend le résultat. Si le traitement réussit, Lambda reprend l’interrogation jusqu’à recevoir plus d’enregistrements.
Par défaut, Lambda invoque votre fonction dès que des enregistrements sont disponibles. Si le lot que Lambda lit à partir de la source d’événements ne comprend qu’un seul enregistrement, Lambda envoie un seul registre à la fonction. Pour éviter d’invoquer la fonction avec un petit nombre de registres, vous pouvez indiquer à la source d’événement de les mettre en mémoire tampon pendant 5 minutes en configurant une *fenêtre de traitement par lots*. Avant d’invoquer la fonction, Lambda continue de lire les registres de la source d’événements jusqu’à ce qu’il ait rassemblé un lot complet, que la fenêtre de traitement par lot expire ou que le lot atteigne la limite de charge utile de 6 Mo. Pour de plus amples informations, veuillez consulter [Comportement de traitement par lots](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching).
**Avertissement**
Les mappages des sources d’événements Lambda traitent chaque événement au moins une fois, et le traitement des enregistrements peut être dupliqué. Pour éviter les problèmes potentiels liés à des événements dupliqués, nous vous recommandons vivement de rendre votre code de fonction idempotent. Pour en savoir plus, consultez [Comment rendre ma fonction Lambda idempotente](https://repost.aws/knowledge-center/lambda-function-idempotent) dans le Knowledge Center. AWS
Lambda envoi le prochain lot pour traitement sans attendre que les [extensions](lambda-extensions.md) configurées soient terminées. En d’autres termes, vos extensions peuvent continuer à s’exécuter pendant que Lambda traite le prochain lot d’enregistrements. Cela peut causer des problèmes de limitation si vous enfreignez l’un des paramètres ou l’une des limites de [simultanéité](lambda-concurrency.md) de votre compte. Pour détecter s’il s’agit d’un problème éventuel, surveillez vos fonctions et vérifiez si vous observez des [métriques de simultanéité](monitoring-concurrency.md#general-concurrency-metrics) plus élevées que prévu pour votre mappage des sources d’événements. En raison de la brièveté des intervalles entre les invocations, Lambda peut brièvement signaler une utilisation simultanée supérieure au nombre de partitions. Cela peut être vrai même pour les fonctions Lambda sans extensions.
Configurez le [ ParallelizationFactor](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-ParallelizationFactor)paramètre pour traiter une partition d'un flux DynamoDB avec plusieurs invocations Lambda simultanément. Vous pouvez spécifier le nombre de lots simultanés que Lambda interroge à partir d’une partition via un facteur de parallélisation de 1 (par défaut) à 10. Par exemple, quand vous définissez `ParallelizationFactor` sur 2, vous pouvez avoir jusqu’à 200 invocations Lambda simultanés pour traiter 100 partitions de données DynamoDB (bien que, dans la réalité, la métrique `ConcurrentExecutions` puisse indiquer une valeur différente). Cela permet d’augmenter le débit de traitement quand le volume de données est volatil et que la valeur du paramètre [IteratorAge](monitoring-metrics-types.md#performance-metrics) est élevée. Lorsque vous augmentez le nombre de lots simultanés par partition, Lambda assure toujours un traitement dans l’ordre au niveau de l’élément (partition et clé de tri).
## Positions de départ des interrogations et des flux
Sachez que l’interrogation des flux lors des mises à jour et de la création du mappage des sources d’événements est finalement cohérente.
+ Lors de la création du mappage des sources d’événements, le démarrage de l’interrogation des événements depuis le flux peut prendre plusieurs minutes.
+ Lors des mises à jour du mappage des sources d’événements, l’arrêt et le redémarrage de l’interrogation des événements depuis le flux peuvent prendre plusieurs minutes.
Ce comportement signifie que si vous spécifiez `LATEST` comme position de départ du flux, le mappage des sources d’événements peut manquer des événements lors de la création ou des mises à jour. Pour vous assurer de ne manquer aucun événement, spécifiez la position de départ du flux comme `TRIM_HORIZON`.
## Lecteurs simultanés d’une partition dans DynamoDB Streams
Pour les tables à région unique qui ne sont pas des tables globales, vous pouvez concevoir jusqu’à deux fonctions Lambda pour lire la même partition DynamoDB Streams simultanément. Le dépassement de cette limite peut se traduire par une limitation de la demande. Pour les tables globales, nous vous recommandons de limiter le nombre de fonctions simultanées à une seule pour éviter la limitation des demandes.
## Exemple d’évènement
**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"
}
]}
```
# Traiter les enregistrements DynamoDB avec Lambda
Créez un mappage de source d’événement pour indiquer à Lambda d’envoyer des enregistrements de votre flux à une fonction Lambda. Vous pouvez créer plusieurs mappages de source d’événement pour traiter les mêmes données avec plusieurs fonctions Lambda, ou traiter des éléments de plusieurs flux avec une seule fonction.
Vous pouvez configurer des mappages de sources d’événements pour traiter les enregistrements d’un flux dans un autre Compte AWS. Pour en savoir plus, veuillez consulter la section [Création d’un mappage des sources d’événements entre comptes](#services-dynamodb-eventsourcemapping-cross-account).
**Pour configurer votre fonction de manière à lire depuis DynamoDB Streams, associez la politique AWS gérée [AWSLambdaDynamo Role à votre DBExecution rôle](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaDynamoDBExecutionRole.html) d'exécution, puis créez un déclencheur DynamoDB.**
**Pour ajouter des autorisations et créer un déclencheur**
1. Ouvrez la [page Functions](https://console.aws.amazon.com/lambda/home#/functions) (Fonctions) de la console Lambda.
1. Choisissez le nom d’une fonction.
1. Choisissez l’onglet **Configuration**, puis **Permissions** (Autorisations).
1. Sous **Nom du rôle**, cliquez sur le lien vers votre rôle d’exécution. Ce lien ouvre le rôle dans la console IAM.
![\[\]](http://docs.aws.amazon.com/fr_fr/lambda/latest/dg/images/execution-role.png)
1. Choisissez **Ajouter des autorisations**, puis **Attacher des politiques**.
![\[\]](http://docs.aws.amazon.com/fr_fr/lambda/latest/dg/images/attach-policies.png)
1. Dans le champ de recherche, entrez `AWSLambdaDynamoDBExecutionRole`. Ajoutez cette politique à votre rôle d’exécution. Il s'agit d'une politique AWS gérée qui contient les autorisations dont votre fonction a besoin pour lire le flux DynamoDB. Pour plus d'informations sur cette politique, consultez [AWSLambdaDynamo DBExecution Role](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaDynamoDBExecutionRole.html) dans le manuel *AWS Managed Policy Reference*.
1. Revenez à votre fonction dans la console Lambda. Sous **Function overview (Vue d’ensemble de la fonction)**, choisissez **Add trigger (Ajouter un déclencheur)**.
![\[\]](http://docs.aws.amazon.com/fr_fr/lambda/latest/dg/images/add-trigger.png)
1. Choisissez un type de déclencheur.
1. Configurez les options requises, puis choisissez **Add (Ajouter)**.
Lambda prend en charge les options suivantes pour les sources d’événement DynamoDB :
**Options de source d’événement**
+ **DynamoDB table (Table DynamoDB)** – Table DynamoDB à partir de laquelle lire les enregistrements.
+ **Batch size (Taille de lot)** – Nombre d’enregistrements par lot à envoyer à la fonction, jusqu’à 10 000. Lambda transmet tous les enregistrements du lot à la fonction en une seule invocation, tant que la taille totale des événements ne dépasse pas la [limite de charge utile](gettingstarted-limits.md) pour une invocation synchrone (6 Mo).
+ **Batch window (Fenêtre de traitement par lots)** – Intervalle de temps maximum (en secondes) pour collecter des enregistrements avant d’invoquer la fonction.
+ **Starting position (Position de départ)** – Traiter uniquement les nouveaux enregistrements, ou tous enregistrement existants.
+ **Latest (Derniers)** – Traiter les nouveaux enregistrements ajoutés au flux.
+ **Trim horizon (Supprimer l’horizon)** – Traiter tous les enregistrements figurant dans le flux.
Après le traitement de tous les enregistrements existants, la fonction est à jour et continue à traiter les nouveaux enregistrements.
+ **Destination en cas d’échec** : une file d’attente SQS standard ou rubrique SNS standard pour les enregistrements qui ne peuvent pas être traités. Quand Lambda écarte un lot d’enregistrements qui est trop ancien ou qui a épuisé toutes les tentatives, il envoie les détails du lot à la file d’attente ou à la rubrique.
+ **Retry attempts (Nouvelles tentatives)** – Nombre maximum de nouvelles tentatives que Lambda effectue quand la fonction renvoie une erreur. Cela ne s’applique pas aux erreurs ou limitations de service où le lot n’a pas atteint la fonction.
+ **Maximum age of record (Âge maximum d’enregistrement)** – Âge maximum d’un enregistrement que Lambda envoie à votre fonction.
+ **Split batch on error (Fractionner le lot en cas d’erreur)** – Quand la fonction renvoie une erreur, diviser le lot en deux avant d’effectuer une nouvelle tentative. Le paramètre de taille de lot d’origine reste inchangé.
+ **Concurrent batches per shard (Lots simultanés par partition)** – Traite simultanément plusieurs lots de la même partition.
+ **Enabled (Activé)** – Valeur définie sur VRAI pour activer le mappage de source d’événement. Définissez la valeur sur « false » pour arrêter le traitement des enregistrements. Lambda garde une trace du dernier enregistrement traité et reprend le traitement à ce point lorsque le mappage est réactivé.
**Note**
Les appels d' GetRecords API invoqués par Lambda dans le cadre des déclencheurs DynamoDB ne vous sont pas facturés.
Pour gérer ultérieurement la configuration de la source d’événement, choisissez le déclencheur dans le concepteur.
## Création d’un mappage des sources d’événements entre comptes
[Amazon DynamoDB prend désormais en charge les politiques basées sur les ressources.](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/access-control-resource-based.html) Grâce à cette fonctionnalité, vous pouvez traiter les données d'un flux DynamoDB dans un compte Compte AWS avec une fonction Lambda dans un autre compte.
Pour créer un mappage de source d'événements pour votre fonction Lambda à l'aide d'un flux DynamoDB dans un autre Compte AWS, vous devez configurer le flux à l'aide d'une politique basée sur les ressources afin d'autoriser votre fonction Lambda à lire des enregistrements. *Pour savoir comment configurer votre flux pour un accès entre comptes, consultez [Partager l'accès avec les fonctions Lambda entre](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/rbac-cross-account-access.html#rbac-analyze-cross-account-lambda-access) comptes dans le guide du développeur Amazon DynamoDB.*
Une fois que vous avez configuré votre flux avec une politique basée sur les ressources qui donne à votre fonction Lambda les autorisations requises, créez le mappage des sources d'événements avec l'ARN de votre flux entre comptes. Vous pouvez trouver l'ARN du flux sous l'onglet **Exportations et flux** du tableau dans la console DynamoDB multicomptes.
Lorsque vous utilisez la console Lambda, collez l'ARN du flux directement dans le champ de saisie de la table DynamoDB sur la page de création du mappage des sources d'événements.
**Remarque :** Les déclencheurs interrégionaux ne sont pas pris en charge.
# Configuration d’une réponse par lots partielle avec DynamoDB et Lambda
Lors de l’utilisation et du traitement de données de streaming à partir d’une source d’événement, par défaut, Lambda n’effectue un point de contrôle sur le numéro de séquence le plus élevé d’un lot que lorsque celui-ci est un succès complet. Lambda traite tous les autres résultats comme un échec complet et recommence à traiter le lot jusqu’à atteindre la limite de nouvelles tentatives. Pour autoriser des succès partiels lors du traitement des lots à partir d’un flux, activez `ReportBatchItemFailures`. Autoriser des succès partiels peut permettre de réduire le nombre de nouvelles tentatives sur un enregistrement, mais cela n’empêche pas entièrement la possibilité de nouvelles tentatives dans un enregistrement réussi.
Pour activer `ReportBatchItemFailures`, incluez la valeur enum **ReportBatchItemFailures** dans la liste[FunctionResponseTypes](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-FunctionResponseTypes). Cette liste indique quels types de réponse sont activés pour votre fonction. Vous pouvez configurer cette liste lorsque vous [créez](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) ou [mettez à jour](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) un mappage des sources d’événements.
**Note**
Même lorsque le code de votre fonction renvoie des réponses d’échec partiel de lot, ces réponses ne seront pas traitées par Lambda à moins que la fonctionnalité `ReportBatchItemFailures` soit explicitement activée pour votre mappage des sources d'événements.
## Syntaxe du rapport
Lors de la configuration des rapports d’échec d’articles de lot, la classe `StreamsEventResponse` est renvoyée avec une liste d’échecs d’articles de lot. Vous pouvez utiliser un objet `StreamsEventResponse` pour renvoyer le numéro de séquence du premier enregistrement ayant échoué dans le lot. Vous pouvez également créer votre classe personnalisée en utilisant la syntaxe de réponse correcte. La structure JSON suivante montre la syntaxe de réponse requise :
```
{
"batchItemFailures": [
{
"itemIdentifier": ""
}
]
}
```
**Note**
Si le tableau `batchItemFailures` contient plusieurs éléments, Lambda utilise l’enregistrement portant le numéro de séquence le plus bas comme point de contrôle. Lambda réessaie ensuite tous les enregistrements à partir de ce point de contrôle.
## Conditions de réussite et d’échec
Lambda traite un lot comme un succès complet si vous renvoyez l’un des éléments suivants :
+ Une liste `batchItemFailure` vide
+ Une liste `batchItemFailure` nulle
+ Une `EventResponse` vide
+ Un nu `EventResponse`
Lambda traite un lot comme un échec complet si vous renvoyez l’un des éléments suivants :
+ Une chaîne vid `itemIdentifier`
+ Un `itemIdentifier` nul
+ Un `itemIdentifier` avec un nom de clé incorrect
Lambda effectue des nouvelles tentatives en cas d’échec en fonction de votre stratégie de nouvelle tentative.
## Diviser un lot
Si votre invocation échoue et que `BisectBatchOnFunctionError` est activé, le lot est divisé en deux quel que soit votre paramètre `ReportBatchItemFailures`.
Quand une réponse de succès partiel de lot est reçue et que les paramètres `BisectBatchOnFunctionError` et `ReportBatchItemFailures` sont activés, le lot est divisé au numéro de séquence renvoyé, et Lambda n’effectue de nouvelle tentative que sur les enregistrements restants.
Pour simplifier la mise en œuvre de la logique de réponse partielle par lots, pensez à utiliser l'[utilitaire Batch Processor](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) de Powertools for AWS Lambda, qui gère automatiquement ces complexités pour vous.
Voici quelques exemples de code de fonction qui renvoie la liste des messages ayant échoué IDs dans le lot :
------
#### [ .NET ]
**SDK pour .NET**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de .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 ]
**Kit SDK pour Go V2**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de 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 pour Java 2.x**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de 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 pour JavaScript (v3)**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signaler les défaillances d'éléments de lot DynamoDB avec Lambda à l'aide de. 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: [] };
};
```
Signaler les défaillances d'éléments de lot DynamoDB avec Lambda à l'aide de. 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 ]
**Kit SDK pour PHP**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de 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 ]
**Kit SDK for Python (Boto3)**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de 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 ]
**Kit SDK pour Ruby**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de 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 pour Rust**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Signalement des échecs d’articles par lots DynamoDB avec Lambda à l’aide de 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
}
```
------
## Utilisation de Powertools pour le AWS Lambda traitement par lots
L'utilitaire de traitement par lots de Powertools for gère AWS Lambda automatiquement la logique de réponse partielle par lots, réduisant ainsi la complexité de la mise en œuvre du signalement des défaillances par lots. Voici des exemples d’utilisation de l’utilitaire de traitement par lots :
**Python**
Pour des exemples complets et des instructions de configuration, consultez la [documentation de l’utilitaire de traitement par lots](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/).
Traitement des enregistrements AWS Lambda de flux DynamoDB à l'aide d'un processeur par lots.
```
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**
Pour des exemples complets et des instructions de configuration, consultez la [documentation de l’utilitaire de traitement par lots](https://docs.aws.amazon.com/powertools/typescript/latest/features/batch/).
Traitement des enregistrements AWS Lambda de flux DynamoDB à l'aide d'un processeur par lots.
```
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**
Pour des exemples complets et des instructions de configuration, consultez la [documentation de l’utilitaire de traitement par lots](https://docs.powertools.aws.dev/lambda/java/latest/utilities/batch/).
Traitement des enregistrements AWS Lambda de flux DynamoDB à l'aide d'un processeur par lots.
```
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**
Pour des exemples complets et des instructions de configuration, consultez la [documentation de l’utilitaire de traitement par lots](https://docs.aws.amazon.com/powertools/dotnet/utilities/batch-processing/).
Traitement des enregistrements AWS Lambda de flux DynamoDB à l'aide d'un processeur par lots.
```
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;
}
}
```
# Retenir les enregistrements ignorés pour une source d’événement DynamoDB dans Lambda
La gestion des erreurs pour les mappages des sources d’événements DynamoDB n’est pas la même selon si l’erreur se produit avant que la fonction ne soit invoquée ou pendant l’invocation de la fonction :
+ **Avant l'appel :** si un mappage de source d'événement Lambda ne parvient pas à appeler la fonction en raison d'un ralentissement ou d'autres problèmes, il réessaie jusqu'à ce que les enregistrements expirent ou dépassent l'âge maximum configuré sur le mappage de source d'événement (). [MaximumRecordAgeInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRecordAgeInSeconds)
+ **Pendant l'appel :** si la fonction est invoquée mais renvoie une erreur, Lambda réessaie jusqu'à ce que les enregistrements expirent, dépassent l'âge maximum [MaximumRecordAgeInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRecordAgeInSeconds)() ou atteignent le quota de nouvelles tentatives configuré (). [MaximumRetryAttempts](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRetryAttempts) Pour les erreurs de fonctionnement, vous pouvez également configurer [BisectBatchOnFunctionError](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-response-BisectBatchOnFunctionError), ce qui divise un lot défaillant en deux lots plus petits, isolant ainsi les mauvais enregistrements et évitant les délais d'attente. Le fractionnement des lots ne consomme pas le quota de nouvelles tentatives.
Si les mesures de gestion des erreurs échouent, Lambda ignore les enregistrements et poursuit le traitement des lots du flux. Avec les paramètres par défaut, cela signifie qu’un enregistrement défectueux peut bloquer le traitement sur la partition affectée pendant jusqu’à une journée. Pour éviter cela, configurez le mappage de source d’événement de votre fonction avec un nombre raisonnable de nouvelles tentatives et un âge maximum d’enregistrement correspondant à votre cas d’utilisation.
## Configuration des destinations pour les invocations ayant échoué
Pour retenir les enregistrements des invocations de mappage de sources d’événements qui ont échoué, ajoutez une destination au mappage des sources d’événements de votre fonction. Chaque enregistrement envoyé à la destination est un document JSON contenant les métadonnées sur l’invocation ayant échoué. Pour les destinations Amazon S3, Lambda envoie également l’intégralité de l’enregistrement d’invocation avec les métadonnées. Vous pouvez configurer n'importe quelle rubrique Amazon SNS, n'importe quelle file d'attente Amazon SQS, n'importe quel compartiment Amazon S3 ou Kafka comme destination.
Avec les destinations Amazon S3, vous pouvez utiliser la fonctionnalité [Notifications d’événements Amazon S3](https://docs.aws.amazon.com/) pour recevoir des notifications lorsque des objets sont chargés dans votre compartiment S3 de destination. Vous pouvez également configurer les notifications d’événements S3 pour invoquer une autre fonction Lambda afin d’effectuer un traitement automatique des lots ayant échoué.
Votre rôle d’exécution doit disposer d’autorisations pour la destination :
+ **Pour une destination SQS : [sqs](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html) :** SendMessage
+ **[Pour une destination SNS : SNS:Publish](https://docs.aws.amazon.com/sns/latest/api/API_Publish.html)**
+ **Pour une destination S3 :** [s3 : PutObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html) et [s3 : ListBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/ListObjectsV2.html)
+ **Pour une destination Kafka : [kafka-cluster](https://docs.aws.amazon.com/msk/latest/developerguide/kafka-actions.html) :** WriteData
Vous pouvez configurer un sujet Kafka comme destination en cas d'échec pour vos mappages de sources d'événements Kafka. Lorsque Lambda ne parvient pas à traiter les enregistrements après avoir épuisé toutes les tentatives ou lorsque les enregistrements dépassent l'âge maximum, Lambda envoie les enregistrements ayant échoué à la rubrique Kafka spécifiée pour un traitement ultérieur. Reportez-vous à [Utiliser un sujet Kafka comme destination en cas d'échec](kafka-on-failure-destination.md).
Si vous avez activé le chiffrement avec votre propre clé KMS pour une destination S3, le rôle d'exécution de votre fonction doit également être autorisé à appeler [kms : GenerateDataKey](https://docs.aws.amazon.com/kms/latest/APIReference/API_GenerateDataKey.html). Si la clé KMS et la destination du compartiment S3 se trouvent dans un compte différent de celui de votre fonction Lambda et de votre rôle d'exécution, configurez la clé KMS pour qu'elle approuve le rôle d'exécution à autoriser. kms: GenerateDataKey
Pour configurer une destination en cas de panne à l’aide de la console, procédez comme suit :
1. Ouvrez la [page Functions](https://console.aws.amazon.com/lambda/home#/functions) (Fonctions) de la console Lambda.
1. Choisissez une fonction.
1. Sous **Function overview (Vue d’ensemble de la fonction)**, choisissez **Add destination (Ajouter une destination)**.
1. Pour **Source**, choisissez **Invocation du mappage des sources d’événements**.
1. Pour le **mappage des sources d’événements**, choisissez une source d’événements configurée pour cette fonction.
1. Pour **Condition**, sélectionnez **En cas d’échec**. Pour les invocations de mappage des sources d’événements, il s’agit de la seule condition acceptée.
1. Pour **Type de destination**, choisissez le type de destination auquel Lambda envoie les enregistrements d’invocation.
1. Pour **Destination**, choisissez une ressource.
1. Choisissez **Enregistrer**.
Vous pouvez également configurer une destination en cas de panne à l'aide de AWS Command Line Interface (AWS CLI). Par exemple, la [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html)commande suivante ajoute un mappage de source d'événement avec une destination SQS en cas de défaillance pour : `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"}}'
```
La [update-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html)commande suivante met à jour le mappage d'une source d'événements afin d'envoyer les enregistrements d'invocation ayant échoué vers une destination SNS après deux tentatives, ou si les enregistrements datent de plus d'une heure.
```
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"}}'
```
Les paramètres mis à jour sont appliqués de façon asynchrone et ne sont pas reflétés dans la sortie tant que le processus n’est pas terminé. Utilisez la commande [get-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html) pour afficher l’état actuel.
Pour supprimer une destination, entrez une chaîne vide comme argument du paramètre `destination-config` :
```
aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--destination-config '{"OnFailure": {"Destination": ""}}'
```
### Pratiques exemplaires en matière de sécurité pour les destinations Amazon S3
La suppression d’un compartiment S3 configuré comme destination sans supprimer la destination de la configuration de votre fonction peut engendrer un risque de sécurité. Si un autre utilisateur connaît le nom de votre compartiment de destination, il peut recréer le compartiment dans son Compte AWS. Les enregistrements des invocations ayant échoué seront envoyés dans son compartiment, exposant potentiellement les données de votre fonction.
**Avertissement**
Pour vous assurer que les enregistrements d'invocation de votre fonction ne peuvent pas être envoyés vers un compartiment S3 d'un autre Compte AWS, ajoutez une condition au rôle d'exécution de votre fonction qui limite `s3:PutObject` les autorisations aux compartiments de votre compte.
L’exemple suivant présente une politique IAM qui limite les autorisations `s3:PutObject` de votre fonction aux seuls compartiments de votre compte. Cette politique donne également à Lambda l’autorisation `s3:ListBucket` dont il a besoin pour utiliser un compartiment S3 comme destination.
```
{
"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"
}
}
}
]
}
```
Pour ajouter une politique d'autorisations au rôle d'exécution de votre fonction à l'aide du AWS Management Console ou AWS CLI, reportez-vous aux instructions des procédures suivantes :
------
#### [ Console ]
**Pour ajouter une politique d’autorisations au rôle d’exécution d’une fonction (console)**
1. Ouvrez la [page Functions](https://console.aws.amazon.com/lambda/home#/functions) (Fonctions) de la console Lambda.
1. Sélectionnez la fonction Lambda dont vous voulez modifier le rôle d’exécution.
1. Sous l’onglet **Configuration**, sélectionnez **Autorisations**.
1. Sous l’onglet **Rôle d’exécution**, sélectionnez le **Nom du rôle** de votre fonction pour ouvrir la page de console IAM du rôle.
1. Ajoutez une politique d’autorisations de au rôle en procédant comme suit :
1. Dans le volet **Politiques d’autorisations**, choisissez **Ajouter des autorisations**, puis **Créer une politique en ligne**.
1. Dans l’**Éditeur de politique**, sélectionnez **JSON**.
1. Collez la politique que vous souhaitez ajouter dans l’éditeur (en remplacement du JSON existant), puis choisissez **Suivant**.
1. Sous **Détails de la politique**, saisissez un **Nom de la politique**.
1. Choisissez **Create Policy** (Créer une politique).
------
#### [ AWS CLI ]
**Pour ajouter une politique d’autorisations au rôle d’exécution d’une fonction (CLI)**
1. Créez un document de politique JSON avec les autorisations requises et enregistrez-le dans un répertoire local.
1. Utilisez la commande `put-role-policy` de la CLI IAM pour ajouter des autorisations au rôle d’exécution de votre fonction. Exécutez la commande suivante depuis le répertoire dans lequel vous avez enregistré votre document de politique JSON et remplacez le nom du rôle, le nom de la politique et le document de politique par vos propres valeurs.
```
aws iam put-role-policy \
--role-name my_lambda_role \
--policy-name LambdaS3DestinationPolicy \
--policy-document file://my_policy.json
```
------
### Exemple d’enregistrement d’invocation Amazon SNS et Amazon SQS
L’exemple suivant illustre un enregistrement d’invocation que Lambda envoie à une destination SQS ou SNS pour un flux 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"
}
}
```
Vous pouvez utiliser ces informations pour récupérer les enregistrements concernés à partir du flux à des fins de résolution de problèmes. Les enregistrements réels n’étant pas inclus, vous devez les récupérer du flux avant qu’ils expirent et soient perdus.
### Exemple d’enregistrement d’invocation Amazon S3
L’exemple suivant illustre un enregistrement d’invocation que Lambda envoie à un compartiment S3 pour un flux DynamoDB. Outre tous les champs de l’exemple précédent pour les destinations SQS et SNS, le champ `payload` contient l’enregistrement d’invocation d’origine sous forme de chaîne JSON échappée.
```
{
"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’objet S3 contenant l’enregistrement d’invocation utilise la convention de dénomination suivante :
```
aws/lambda///YYYY/MM/DD/YYYY-MM-DDTHH.MM.SS-
```
# Implémentation du traitement des flux DynamoDB avec état dans Lambda
Les fonctions Lambda peuvent exécuter des applications de traitement de flux continu. Un flux représente des données illimitées qui circulent en continu dans votre application. Pour analyser les informations provenant de cette entrée de mise à jour continue, vous pouvez lier les enregistrements inclus à l’aide d’une fenêtre de temps définie.
Les fenêtres bascules sont des fenêtres temporelles distinctes qui s’ouvrent et se ferment à intervalles réguliers. Par défaut, les invocations Lambda sont sans état : vous ne pouvez pas les utiliser pour traiter des données sur plusieurs invocations continues sans base de données externe. Cependant, avec les fenêtres bascules, vous pouvez maintenir votre état au long des invocations. Cet état contient le résultat global des messages précédemment traités pour la fenêtre actuelle. Votre état peut être d’un maximum de 1 Mo par partition. S’il dépasse cette taille, Lambda met fin précocement à la fenêtre de traitement.
Chaque enregistrement d’un flux appartient à une fenêtre spécifique. La fonction Lambda traitera chaque enregistrement au moins une fois. Toutefois, elle ne garantit pas un seul traitement pour chaque enregistrement. Dans de rares cas, tels que pour la gestion des erreurs, certains enregistrements peuvent être sujet à de multiples traitements. Les dossiers sont toujours traités dans l’ordre dès la première fois. Si les enregistrements sont traités plusieurs fois, ils peuvent être traités dans le désordre.
## Regroupement et traitement
Votre fonction gérée par l’utilisateur est invoquée tant pour l’agrégation que pour le traitement des résultats finaux de celle-ci. Lambda regroupe tous les enregistrements reçus dans la fenêtre. Vous pouvez recevoir ces enregistrements en plusieurs lots, chacun sous forme d’invocation séparée. Chaque invocation reçoit un état. Ainsi, lorsque vous utilisez des fenêtres bascules, votre réponse de fonction Lambda doit contenir une propriété `state`. Si la réponse ne contient pas de propriété `state`, Lambda considère qu’il s’agit d’une invocation ayant échoué. Pour satisfaire à cette condition, votre fonction peut renvoyer un objet `TimeWindowEventResponse` ayant la forme JSON suivante :
**Example `TimeWindowEventResponse` values**
```
{
"state": {
"1": 282,
"2": 715
},
"batchItemFailures": []
}
```
**Note**
Pour les fonctions Java, nous vous recommandons d’utiliser `Map` pour représenter l’état.
À la fin de la fenêtre, l’indicateur `isFinalInvokeForWindow` est défini sur `true` pour indiquer qu’il s’agit de l’état final et qu’il est prêt pour le traitement. Après le traitement, la fenêtre et votre invocation final se terminent, puis l’état est supprimé.
À la fin de votre fenêtre, Lambda applique un traitement final pour les actions sur les résultats de l’agrégation. Votre traitement final est invoqué de manière synchrone. Une fois l’invocation réussie, votre fonction contrôle le numéro de séquence et le traitement du flux continue. Si l’invocation échoue, votre fonction Lambda suspend le traitement ultérieur jusqu’à ce que l’invocation soit réussie.
**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
}
```
## Configuration
Vous pouvez configurer des fenêtres bascule lorsque vous créez ou mettez à jour un mappage de source d’événement. Pour configurer une fenêtre variable, spécifiez la fenêtre en secondes ([TumblingWindowInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-TumblingWindowInSeconds)). L'exemple de commande suivant AWS Command Line Interface (AWS CLI) crée un mappage des sources d'événements de streaming dont la fenêtre de basculement est de 120 secondes. La fonction Lambda définie pour l’agrégation et le traitement est nommée `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 détermine les limites des fenêtres bascule en fonction de l’heure à laquelle les enregistrements ont été insérés dans le flux. Tous les enregistrements ont un horodatage approximatif disponible que Lambda utilise pour déterminer des limites.
Les agrégations de fenêtres bascule ne prennent pas en charge le repartitionnement. Quand la partition prend fin, Lambda considère que la fenêtre de traitement est fermée, et les partitions enfants entament leur propre fenêtre de traitement dans un nouvel état.
Les fenêtres bascule prennent complètement en charge les stratégies de nouvelle tentative existantes `maxRetryAttempts` et `maxRecordAge`.
**Example Handler.py – Agrégation et traitement**
La fonction Python suivante montre comment regrouper et traiter votre état final :
```
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}
```
# Paramètres Lambda pour les mappages des sources d’événement Amazon DynamoDB
Tous les types de sources d'événements Lambda partagent les mêmes opérations [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html)et les mêmes opérations d'[UpdateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html)API. Cependant, seuls certains paramètres s’appliquent à DynamoDB Streams.
| Paramètre | Obligatoire | Par défaut | Remarques |
| --- | --- | --- | --- |
| BatchSize | N | 100 | Maximum : 10 000. |
| BisectBatchOnFunctionError | N | FAUX | Aucune |
| DestinationConfig | N | N/A | File d’attente Amazon SQS standard ou destination de rubrique Amazon SNS standard pour les enregistrements ignorés |
| Activé | N | VRAI | Aucune |
| EventSourceArn | Y | N/A | ARN du flux de données ou d’un consommateur de flux |
| FilterCriteria | N | N/A | [Contrôle des événements envoyés par Lambda à votre fonction](invocation-eventfiltering.md) |
| FunctionName | Y | N/A | Aucune |
| FunctionResponseTypes | N | N/A | Pour permettre à votre fonction de signaler des échecs spécifiques dans un lot, incluez la valeur `ReportBatchItemFailures` dans `FunctionResponseTypes`. Pour de plus amples informations, veuillez consulter [Configuration d’une réponse par lots partielle avec DynamoDB et Lambda](services-ddb-batchfailurereporting.md). |
| MaximumBatchingWindowInSeconds | N | 0 | Aucune |
| MaximumRecordAgeInSeconds | N | -1 | -1 signifie infini : les enregistrements qui ont échoué sont réessayés jusqu’à ce que l’enregistrement expire. La [limite de conservation des données pour les flux DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.DataRetention) est de 24 heures. Minimum : -1 Maximum : 604 800 |
| MaximumRetryAttempts | N | -1 | -1 signifie infini : les registres qui ont échoué sont réessayés jusqu’à ce que le registre expire. Minimum : 0 Maximum : 10 000. |
| ParallelizationFactor | N | 1 | Maximum : 10 |
| StartingPosition | Y | N/A | TRIM\$1HORIZON ou LATEST |
| TumblingWindowInSeconds | N | N/A | Minimum : 0 Maximum : 900 |
# Utilisation du filtrage des événements avec une source d’événement DynamoDB
Vous pouvez utiliser le filtrage d’événements pour contrôler les enregistrements d’un flux ou d’une file d’attente que Lambda envoie à votre fonction. Pour obtenir des informations générales sur le fonctionnement du filtrage des événements, consultez [Contrôle des événements envoyés par Lambda à votre fonction](invocation-eventfiltering.md).
Cette section porte sur le filtrage des événements pour les sources d’événement DynamoDB.
**Note**
Les mappages des sources d’événements DynamoDB prennent uniquement en charge le filtrage sur la clé `dynamodb`.
**Topics**
+ [Événement DynamoDB](#filtering-ddb)
+ [Filtrage à l’aide des attributs de table](#filtering-ddb-attributes)
+ [Filtrage à l’aide d’expressions booléennes](#filtering-ddb-boolean)
+ [Utilisation de l’opérateur Exists](#filtering-ddb-exists)
+ [Format JSON pour le filtrage DynamoDB](#filtering-ddb-JSON-format)
## Événement DynamoDB
Supposons que vous ayez une table DynamoDB avec la clé primaire `CustomerName` et les attributs `AccountManager` et `PaymentTerms`. La figure suivante montre un exemple d’enregistrement provenant du flux de votre table 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"
}
}
```
Pour filtrer sur la base des valeurs de clé et d’attribut de votre table DynamoDB, utilisez la clé `dynamodb` dans l’enregistrement. Les sections suivantes fournissent des exemples de différents types de filtres.
### Filtrage à l’aide des clés de table
Supposons que vous vouliez que votre fonction ne traite que les enregistrements dont la clé primaire `CustomerName` est « AnyCompany Industries. » L’objet `FilterCriteria` serait le suivant.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"
}
]
}
```
Pour plus de clarté, voici la valeur du `Pattern` de filtre étendu en JSON simple :
```
{
"dynamodb": {
"Keys": {
"CustomerName": {
"S": [ "AnyCompany Industries" ]
}
}
}
}
```
Vous pouvez ajouter votre filtre à l’aide de la console, d’AWS CLI ou d’un modèle AWS SAM.
------
#### [ Console ]
Pour ajouter ce filtre à l’aide de la console, suivez les instructions de [Attacher des critères de filtre à un mappage de sources d’événements (console)](invocation-eventfiltering.md#filtering-console) et saisissez la chaîne suivante comme **critère de filtre**.
```
{ "dynamodb" : { "Keys" : { "CustomerName" : { "S" : [ "AnyCompany Industries" ] } } } }
```
------
#### [ AWS CLI ]
Pour créer un nouveau mappage des sources d’événements avec ces critères de filtrage à l’aide de l’outil AWS Command Line Interface (AWS CLI), exécutez la commande suivante.
```
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\" ] } } } }"}]}'
```
Pour ajouter ces critères de filtre à un mappage des sources d’événements existant, exécutez la commande suivante.
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"}]}'
```
------
#### [ AWS SAM ]
Pour ajouter ce filtre à l’aide d’AWS SAM, ajoutez l’extrait suivant au modèle YAML de votre source d’événement.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "Keys" : { "CustomerName" : { "S" : [ "AnyCompany Industries" ] } } } }'
```
------
## Filtrage à l’aide des attributs de table
Avec DynamoDB, vous pouvez également utiliser les clés `NewImage` et `OldImage` pour filtrer les valeurs d’attributs. Supposons que vous vouliez filtrer les enregistrements où l’attribut `AccountManager` de la dernière image de la table est « Pat Candella » ou « Shirley Rodriguez ». L’objet `FilterCriteria` serait le suivant.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\", \"Shirley Rodriguez\" ] } } } }"
}
]
}
```
Pour plus de clarté, voici la valeur du `Pattern` de filtre étendu en JSON simple :
```
{
"dynamodb": {
"NewImage": {
"AccountManager": {
"S": [ "Pat Candella", "Shirley Rodriguez" ]
}
}
}
}
```
Vous pouvez ajouter votre filtre à l’aide de la console, d’AWS CLI ou d’un modèle AWS SAM.
------
#### [ Console ]
Pour ajouter ce filtre à l’aide de la console, suivez les instructions de [Attacher des critères de filtre à un mappage de sources d’événements (console)](invocation-eventfiltering.md#filtering-console) et saisissez la chaîne suivante comme **critère de filtre**.
```
{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella", "Shirley Rodriguez" ] } } } }
```
------
#### [ AWS CLI ]
Pour créer un nouveau mappage des sources d’événements avec ces critères de filtrage à l’aide de l’outil AWS Command Line Interface (AWS CLI), exécutez la commande suivante.
```
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\" ] } } } }"}]}'
```
Pour ajouter ces critères de filtre à un mappage des sources d’événements existant, exécutez la commande suivante.
```
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 ]
Pour ajouter ce filtre à l’aide d’AWS SAM, ajoutez l’extrait suivant au modèle YAML de votre source d’événement.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella", "Shirley Rodriguez" ] } } } }'
```
------
## Filtrage à l’aide d’expressions booléennes
Vous pouvez également créer des filtres à l’aide des expressions booléennes AND. Ces expressions peuvent inclure les paramètres de clé et d’attribut de votre table. Supposons que vous souhaitiez filtrer les enregistrements dont la valeur `NewImage` d’`AccountManager` est « Pat Candella » et la valeur `OldImage` est « Terry Whitlock ». L’objet `FilterCriteria` serait le suivant.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\" ] } } } , \"dynamodb\" : { \"OldImage\" : { \"AccountManager\" : { \"S\" : [ \"Terry Whitlock\" ] } } } }"
}
]
}
```
Pour plus de clarté, voici la valeur du `Pattern` de filtre étendu en JSON simple :
```
{
"dynamodb" : {
"NewImage" : {
"AccountManager" : {
"S" : [
"Pat Candella"
]
}
}
},
"dynamodb": {
"OldImage": {
"AccountManager": {
"S": [
"Terry Whitlock"
]
}
}
}
}
```
Vous pouvez ajouter votre filtre à l’aide de la console, d’AWS CLI ou d’un modèle AWS SAM.
------
#### [ Console ]
Pour ajouter ce filtre à l’aide de la console, suivez les instructions de [Attacher des critères de filtre à un mappage de sources d’événements (console)](invocation-eventfiltering.md#filtering-console) et saisissez la chaîne suivante comme **critère de filtre**.
```
{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella" ] } } } , "dynamodb" : { "OldImage" : { "AccountManager" : { "S" : [ "Terry Whitlock" ] } } } }
```
------
#### [ AWS CLI ]
Pour créer un nouveau mappage des sources d’événements avec ces critères de filtrage à l’aide de l’outil AWS Command Line Interface (AWS CLI), exécutez la commande suivante.
```
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\" ] } } } } "}]}'
```
Pour ajouter ces critères de filtre à un mappage des sources d’événements existant, exécutez la commande suivante.
```
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 ]
Pour ajouter ce filtre à l’aide d’AWS SAM, ajoutez l’extrait suivant au modèle YAML de votre source d’événement.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella" ] } } } , "dynamodb" : { "OldImage" : { "AccountManager" : { "S" : [ "Terry Whitlock" ] } } } }'
```
------
**Note**
Le filtrage d’événements DynamoDB ne prend pas en charge l’utilisation d’opérateurs numériques (égalité numérique et plage numérique). Même si les éléments de votre table sont stockés sous forme de nombres, ces paramètres sont convertis en chaînes dans l’objet d’enregistrement JSON.
## Utilisation de l’opérateur Exists
En raison de la structure des objets d’événements JSON de DynamoDB, l’utilisation de l’opérateur Exists nécessite une attention particulière. L’opérateur Exists ne fonctionne que sur les nœuds terminaux dans l’événement JSON. Par conséquent, si votre modèle de filtre utilise Exists pour tester un nœud intermédiaire, il ne fonctionnera pas. Observez l’élément de table DynamoDB suivant :
```
{
"UserID": {"S": "12345"},
"Name": {"S": "John Doe"},
"Organizations": {"L": [
{"S":"Sales"},
{"S":"Marketing"},
{"S":"Support"}
]
}
}
```
Vous pourriez avoir besoin de créer un modèle de filtre comme le suivant pour tester les événements contenant `"Organizations"` :
```
{ "dynamodb" : { "NewImage" : { "Organizations" : [ { "exists": true } ] } } }
```
Cependant, ce modèle de filtre ne renverra jamais de correspondance, car `"Organizations"` n’est pas un nœud terminal. L’exemple suivant montre comment utiliser correctement l’opérateur Exists pour construire le modèle de filtre souhaité :
```
{ "dynamodb" : { "NewImage" : {"Organizations": {"L": {"S": [ {"exists": true } ] } } } } }
```
## Format JSON pour le filtrage DynamoDB
Pour filtrer correctement les événements provenant de sources DynamoDB, le champ de données et vos critères de filtre pour le champ de données (`dynamodb`) doivent être au format JSON valide. Si l’un ou l’autre des champs n’est pas dans un format JSON valide, Lambda rejette le message ou lance une exception. Le tableau suivant résume le comportement spécifique :
| Format des données entrantes | Pas de modèle de filtre pour les propriétés des données | Action obtenue. |
| --- | --- | --- |
| JSON valide | JSON valide | Lambda filtre en fonction de vos critères de filtre. |
| JSON valide | Pas de modèle de filtre pour les propriétés des données | Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre. |
| JSON valide | Non JSON | Lambda lance une exception au moment de la création ou de la mise à jour du mappage de sources d’événements. Le modèle de filtre des propriétés de données doit être au format JSON valide. |
| Non JSON | JSON valide | Lambda rejette l’enregistrement. |
| Non JSON | Pas de modèle de filtre pour les propriétés des données | Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre. |
| Non JSON | Non JSON | Lambda lance une exception au moment de la création ou de la mise à jour du mappage de sources d’événements. Le modèle de filtre des propriétés de données doit être au format JSON valide. |
# Tutoriel : Utilisation AWS Lambda avec les flux Amazon DynamoDB
Dans ce didacticiel, vous allez créer une fonction Lambda afin d’utiliser des événements d’un flux Amazon DynamoDB.
## Conditions préalables
### Installez le AWS Command Line Interface
Si vous ne l'avez pas encore installé AWS Command Line Interface, suivez les étapes décrites dans la [section Installation ou mise à jour de la dernière version du AWS CLI pour l'](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html)installer.
Ce tutoriel nécessite un terminal de ligne de commande ou un shell pour exécuter les commandes. Sous Linux et macOS, utilisez votre gestionnaire de shell et de package préféré.
**Note**
Sous Windows, certaines commandes CLI Bash que vous utilisez couramment avec Lambda (par exemple `zip`) ne sont pas prises en charge par les terminaux intégrés du système d’exploitation. [Installez le sous-système Windows pour Linux](https://docs.microsoft.com/en-us/windows/wsl/install-win10) afin d’obtenir une version intégrée à Windows d’Ubuntu et Bash.
## Créer le rôle d’exécution
Créez le [rôle d'exécution](lambda-intro-execution-role.md) qui autorise votre fonction à accéder aux AWS ressources.
**Pour créer un rôle d’exécution**
1. Ouvrez la page [Roles (Rôles)](https://console.aws.amazon.com/iam/home#/roles) dans la console IAM.
1. Sélectionnez **Créer un rôle**.
1. Créez un rôle avec les propriétés suivantes :
+ **Entité de confiance** – Lambda.
+ **Autorisations** — **DBExecutionRôle AWSLambda Dynamo**.
+ **Nom de rôle** – **lambda-dynamodb-role**.
Le **DBExecutionrôle AWSLambda Dynamo** dispose des autorisations dont la fonction a besoin pour lire des éléments depuis DynamoDB et écrire des journaux dans des journaux. CloudWatch
## Créer la fonction
Créez une fonction Lambda qui traite vos événements DynamoDB. Le code de fonction écrit certaines des données d'événements entrants dans CloudWatch Logs.
------
#### [ .NET ]
**SDK pour .NET**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant .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 ]
**Kit SDK pour Go V2**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant 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 pour Java 2.x**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant 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 pour JavaScript (v3)**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d'un événement DynamoDB avec Lambda en utilisant. 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)}`);
};
```
Consommation d'un événement DynamoDB avec Lambda en utilisant. 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 ]
**Kit SDK pour PHP**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant 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 ]
**Kit SDK for Python (Boto3)**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant 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 ]
**Kit SDK pour Ruby**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant 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 pour Rust**
Il y en a plus à ce sujet GitHub. Trouvez l’exemple complet et découvrez comment le configurer et l’exécuter dans le référentiel d’[exemples sans serveur](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consommation d’un événement DynamoDB avec Lambda en utilisant 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(())
}
```
------
**Pour créer la fonction**
1. Copiez l’exemple de code dans un fichier nommé `example.js`.
1. Créez un package de déploiement.
```
zip function.zip example.js
```
1. Créez une fonction Lambda à l’aide de la commande `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 de la fonction Lambda
Au cours de cette étape, vous appelez votre fonction Lambda manuellement à l'aide de la commande `invoke` AWS Lambda CLI et de l'exemple d'événement DynamoDB suivant. Copiez le code suivant dans un fichier nommé `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"
}
]
}
```
Exécutez la commande suivante `invoke`.
```
aws lambda invoke --function-name ProcessDynamoDBRecords \
--cli-binary-format raw-in-base64-out \
--payload file://input.txt outputfile.txt
```
L'**cli-binary-format**option est obligatoire si vous utilisez AWS CLI la version 2. Pour faire de ce paramètre le paramètre par défaut, exécutez `aws configure set cli-binary-format raw-in-base64-out`. Pour plus d’informations, consultez les [options de ligne de commande globales prises en charge par l’AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-options.html#cli-configure-options-list) dans le *Guide de l’utilisateur AWS Command Line Interface version 2*.
La fonction renvoie la chaîne `message` dans le corps de la réponse.
Vérifiez la sortie dans le fichier `outputfile.txt`.
## Créer une table DynamoDB avec un flux activé
Créez une table Amazon DynamoDB avec un flux activé.
**Pour créer une table DynamoDB**
1. Ouvrez la [console DynamoDB](https://console.aws.amazon.com/dynamodb).
1. Choisissez **Créer un tableau**.
1. Créez une table avec les paramètres suivants.
+ **Nom de la table** – **lambda-dynamodb-stream**
+ **Clé primaire** – **id** (chaîne)
1. Choisissez **Create (Créer)**.
**Pour activer les flux**
1. Ouvrez la [console DynamoDB](https://console.aws.amazon.com/dynamodb).
1. Choisissez **Tables**.
1. Choisissez la table **lambda-dynamodb-stream**.
1. Sous la section **Exports and streams (Exportations et flux)**, choisissez **DynamoDB stream details (Détails du flux DynamoDB)**.
1. Choisissez **Activer**.
1. Pour **Type de vue**, sélectionnez **Attributs de clé uniquement**.
1. Choisissez **Activer le streaming**.
Écrivez l’ARN du flux. Vous en aurez besoin à l’étape suivante pour associer le flux à votre fonction Lambda. Pour plus d’informations sur l’activation des flux, consultez [Capture d’activité Table avec DynamoDB Streams](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html).
## Ajouter une source d'événement dans AWS Lambda
Créez un mappage de source d’événement dans AWS Lambda. Ce mappage de source d’événement associe le flux DynamoDB avec votre fonction Lambda. Après avoir créé ce mappage des sources d'événements, AWS Lambda commence à interroger le flux.
Exécutez la commande suivante AWS CLI `create-event-source-mapping`. Une fois la commande exécutée, notez l’UUID. Vous aurez besoin de l’UUID pour faire référence au mappage de source d’événement dans les commandes (par exemple, lors de la suppression du mappage).
```
aws lambda create-event-source-mapping --function-name ProcessDynamoDBRecords \
--batch-size 100 --starting-position LATEST --event-source DynamoDB-stream-arn
```
Cela a pour effet de créer un mappage entre le flux DynamoDB spécifié et la fonction Lambda. Vous pouvez associer un flux diffuser à plusieurs fonctions Lambda, et associer la même fonction Lambda à plusieurs flux. Toutefois, les fonctions Lambda partagent le débit de lecture du flux qu’elles partagent.
Pour obtenir la liste des mappages de source d’événement, exécutez la commande suivante.
```
aws lambda list-event-source-mappings
```
Cette liste renvoie tous les mappages de source d’événement que vous avez créés et indique la valeur `LastProcessingResult` pour chacun d’eux, entre autres. Ce champ est utilisé pour fournir un message d’information en cas de problème. Des valeurs telles que `No records processed` (indique que le sondage n' AWS Lambda a pas commencé ou qu'il n'y a aucun enregistrement dans le flux) et `OK` (indique que les enregistrements du flux AWS Lambda ont été lus avec succès et que votre fonction Lambda a été invoquée) indiquent qu'il n'y a aucun problème. Dans le cas contraire, vous recevez un message d’erreur.
Si vous avez un grand nombre de mappages de source d’événement, utilisez le paramètre du nom de la fonction pour affiner les résultats.
```
aws lambda list-event-source-mappings --function-name ProcessDynamoDBRecords
```
## Tester la configuration
Testez l' end-to-endexpérience. A mesure que vous mettez la table à jour, DynamoDB écrit les enregistrements d’événement dans le flux. Quand AWS Lambda interroge le flux, il y détecte les nouveaux enregistrements et invoque pour vous la fonction Lambda en transmettant les événements à la fonction.
1. Dans la console DynamoDB, vous pouvez ajouter, mettre à jour et supprimer des éléments dans la table. DynamoDB écrit des enregistrements de ces actions dans le flux.
1. AWS Lambda interroge le flux et lorsqu'il détecte des mises à jour du flux, il invoque votre fonction Lambda en transmettant les données d'événements qu'il trouve dans le flux.
1. Votre fonction s'exécute et crée des journaux sur Amazon CloudWatch. Vous pouvez vérifier les journaux signalés dans la CloudWatch console Amazon.
## Étapes suivantes
Ce didacticiel vous a montré les principes de base du traitement des événements de flux DynamoDB avec Lambda. Pour les charges de travail de production, envisagez de mettre en œuvre une logique de réponse partielle de lot afin de gérer plus efficacement les échecs d’enregistrement individuels. L'[utilitaire de traitement par lots](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) de Powertools for AWS Lambda est disponible en Python TypeScript, .NET et Java et fournit une solution robuste pour cela, en gérant automatiquement la complexité des réponses partielles par lots et en réduisant le nombre de tentatives pour les enregistrements traités avec succès.
## Nettoyage de vos ressources
Vous pouvez maintenant supprimer les ressources que vous avez créées pour ce didacticiel, sauf si vous souhaitez les conserver. En supprimant AWS les ressources que vous n'utilisez plus, vous évitez des frais inutiles pour votre Compte AWS.
**Pour supprimer la fonction Lambda**
1. Ouvrez la [page Functions (Fonctions)](https://console.aws.amazon.com/lambda/home#/functions) de la console Lambda.
1. Sélectionnez la fonction que vous avez créée.
1. Sélectionnez **Actions**, **Supprimer**.
1. Saisissez **confirm** dans la zone de saisie de texte et choisissez **Delete** (Supprimer).
**Pour supprimer le rôle d’exécution**
1. Ouvrez la [page Roles (Rôles)](https://console.aws.amazon.com/iam/home#/roles) de la console IAM.
1. Sélectionnez le rôle d’exécution que vous avez créé.
1. Sélectionnez **Delete (Supprimer)**.
1. Saisissez le nom du rôle dans le champ de saisie de texte et choisissez **Delete** (Supprimer).
**Pour supprimer la table DynamoDB**
1. Ouvrez la [page Tables (Tables)](https://console.aws.amazon.com//dynamodb/home#tables:) de la console DynamoDB.
1. Sélectionnez la table que vous avez créée.
1. Choisissez **Supprimer**.
1. Saisissez **delete** dans la zone de texte.
1. Choisissez **Supprimer la table**.