# Uso de AWS Lambda con Amazon DynamoDB
**nota**
Si desea enviar datos a un destino que no sea una función de Lambda o enriquecer los datos antes de enviarlos, consulte [Amazon EventBridge Pipes](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) (Canalizaciones de Amazon EventBridge).
Puede utilizar una función AWS Lambda para procesar los registros de un [flujo de Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html). Con DynamoDB Streams, puede activar una función de Lambda para realizar trabajo adicional cada vez que se actualice una tabla de DynamoDB.
Cuando se procesan transmisiones de DynamoDB, se debe implementar una lógica de respuesta por lotes parcial para evitar que se intente reprocesar los registros ya procesados correctamente cuando algunos registros de un lote fallan. La [utilidad de procesador por lotes](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) de Powertools para AWS Lambda está disponible en Python, TypeScript, .NET y Java y simplifica esta implementación al gestionar de manera automática la lógica de respuesta parcial por lotes, lo que reduce el tiempo de desarrollo y mejora la fiabilidad.
**Topics**
+ [Sondeo y procesamiento por lotes de flujos](#dynamodb-polling-and-batching)
+ [Posiciones iniciales de flujos y sondeo](#dyanmo-db-stream-poll)
+ [Lectores simultáneos de una partición en DynamoDB Streams](#events-dynamodb-simultaneous-readers)
+ [Evento de ejemplo](#events-sample-dynamodb)
+ [Proceso de registros de DynamoDB con Lambda](services-dynamodb-eventsourcemapping.md)
+ [Configuración de la respuesta por lotes parcial con DynamoDB y Lambda](services-ddb-batchfailurereporting.md)
+ [Conservación de registros descartados para un origen de eventos de DynamoDB en Lambda](services-dynamodb-errors.md)
+ [Implementación del procesamiento con estado del flujo DynamoDB en Lambda](services-ddb-windows.md)
+ [Parámetros de Lambda para las asignaciones de orígenes de eventos de Amazon DynamoDB](services-ddb-params.md)
+ [Uso del filtrado de eventos con una fuente de eventos de DynamoDB](with-ddb-filtering.md)
+ [Tutorial: Uso de AWS Lambda con Amazon DynamoDB Streams](with-ddb-example.md)
## Sondeo y procesamiento por lotes de flujos
Lambda sondea las particiones del DynamoDB Stream y busca registros 4 veces por segundo. Cuando hay registros disponibles, Lambda invoca la función y espera el resultado. Si el procesamiento se realiza correctamente, Lambda reanuda el sondeo hasta que recibe más registros.
De forma predeterminada, Lambda invoca su función tan pronto como los registros estén disponibles. Si el lote que Lambda lee del origen de eventos solo tiene un registro, Lambda envía solo un registro a la función. Para evitar invocar la función con un número de registros pequeño, puede indicar al origen de eventos que almacene en búfer registros durante hasta 5 minutos configurando un *plazo de procesamiento por lotes*. Antes de invocar la función, Lambda continúa leyendo los registros del origen de eventos hasta que haya recopilado un lote completo, venza el plazo de procesamiento por lotes o el lote alcance el límite de carga de 6 MB. Para obtener más información, consulte [Comportamiento de procesamiento por lotes](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching).
**aviso**
Las asignaciones de orígenes de eventos de Lambda procesan cada evento al menos una vez, y puede producirse un procesamiento duplicado de registros. Para evitar posibles problemas relacionados con la duplicación de eventos, le recomendamos encarecidamente que haga que el código de la función sea idempotente. Para obtener más información, consulte [¿Cómo puedo hacer que mi función de Lambda sea idempotente?](https://repost.aws/knowledge-center/lambda-function-idempotent) en el Centro de conocimientos de AWS.
Lambda no espera a que se completen las [extensiones](lambda-extensions.md) configuradas antes de enviar el siguiente lote para su procesamiento. En otras palabras, las extensiones pueden seguir ejecutándose mientras Lambda procesa el siguiente lote de registros. Esto puede provocar problemas de limitación si infringe alguno de los ajustes o límites de [simultaneidad](lambda-concurrency.md) de la cuenta. Para detectar si se trata de un posible problema, supervise sus funciones y compruebe si ve [métricas de simultaneidad](monitoring-concurrency.md#general-concurrency-metrics) más elevadas de lo esperado para la asignación de orígenes de eventos. Debido a los tiempos cortos entre invocaciones, Lambda puede informar brevemente un uso de simultaneidad superior al número de particiones. Esto puede ser cierto incluso para las funciones de Lambda sin extensiones.
Ajuste la configuración de [ParallelizationFactor](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-ParallelizationFactor) para procesar una partición de un flujo de DynamoDB con más de una invocación de Lambda simultáneamente. Puede especificar el número de lotes simultáneos que Lambda sondea desde una partición a través de un factor de paralelización de 1 (predeterminado) a 10. Por ejemplo, cuando establece `ParallelizationFactor` en 2, puede tener un máximo de 200 invocaciones de Lambda simultáneas para procesar 100 particiones de flujos de DynamoDB (aunque, en la práctica, es posible que observe diferentes valores para la métrica `ConcurrentExecutions`). Esto ayuda a escalar verticalmente el rendimiento de procesamiento cuando el volumen de datos es volátil y la [IteratorAge](monitoring-metrics-types.md#performance-metrics) es alta. Si aumenta el número de lotes simultáneos por partición, Lambda sigue garantizando el procesamiento en orden del nivel del elemento (partición y clave de clasificación).
## Posiciones iniciales de flujos y sondeo
Tenga en cuenta que el sondeo de flujos durante la creación y las actualizaciones de la asignación de orígenes de eventos es, en última instancia, coherente.
+ Durante la creación de la asignación de orígenes de eventos, es posible que se demore varios minutos en iniciar el sondeo de los eventos del flujo.
+ Durante las actualizaciones de la asignación de orígenes de eventos, es posible que se demore varios minutos en detener y reiniciar el sondeo de los eventos del flujo.
Este comportamiento significa que, si especifica `LATEST` como posición inicial del flujo, la asignación de orígenes de eventos podría omitir eventos durante la creación o las actualizaciones. Para garantizar que no se pierda ningún evento, especifique la posición inicial del flujo como `TRIM_HORIZON`.
## Lectores simultáneos de una partición en DynamoDB Streams
En el caso de las tablas de una sola región que no sean tablas globales, puede diseñar hasta dos funciones de Lambda para leer desde la misma partición de DynamoDB Streams al mismo tiempo. Si excede este límite, puede producirse una limitación controlada de las solicitudes. En el caso de las tablas globales, le recomendamos que limite el número de funciones simultáneas a uno para evitar la limitación de solicitudes.
## Evento de ejemplo
**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"
}
]}
```
# Proceso de registros de DynamoDB con Lambda
Cree una asignación de orígenes de eventos para indicar a Lambda que envíe registros desde un flujo a una función de Lambda. Puede crear varias asignaciones de orígenes de eventos para procesar los mismos datos con distintas funciones de Lambda o para procesar elementos de varios flujos con una sola función.
Puede configurar las asignaciones de orígenes de eventos para procesar los registros de un flujo en una Cuenta de AWS diferente. Para obtener más información, consulte [Creación de asignaciones de orígenes de eventos entre cuentas](#services-dynamodb-eventsourcemapping-cross-account).
Para configurar la función para que lea desde el flujo de DynamoDB, adjunte la política administrada de AWS [AWSLambdaDynamoDBExecutionRole](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaDynamoDBExecutionRole.html) al rol de ejecución y, a continuación, cree un desencadenador de **DynamoDB**.
**Cómo agregar permisos y crear un desencadenador**
1. Abra la página de [Funciones](https://console.aws.amazon.com/lambda/home#/functions) en la consola de Lambda.
1. Elija el nombre de una función.
1. Elija la pestaña **Configuración** y, a continuación, elija **Permisos**.
1. En **Nombre del rol**, elija el enlace al rol de ejecución. Este enlace abre el rol en la consola de IAM.
![\[\]](http://docs.aws.amazon.com/es_es/lambda/latest/dg/images/execution-role.png)
1. Elija **Agregar permisos** y luego **Adjuntar políticas**.
![\[\]](http://docs.aws.amazon.com/es_es/lambda/latest/dg/images/attach-policies.png)
1. En el campo de búsqueda, escriba `AWSLambdaDynamoDBExecutionRole`. Agregue esta política al rol de ejecución. Se trata de una política administrada por AWS que contiene los permisos que la función necesita para leer desde un flujo de DynamoDB. Para obtener más información acerca de esta política, consulte [AWSLambdaDynamoDBExecutionRole](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaDynamoDBExecutionRole.html) en la *Referencia de políticas administradas de AWS*.
1. Regrese a la función en la consola de Lambda. En **Descripción general de la función**, elija **Agregar desencadenador**.
![\[\]](http://docs.aws.amazon.com/es_es/lambda/latest/dg/images/add-trigger.png)
1. Elija un tipo de desencadenador.
1. Configure las opciones requeridas y luego elija **Add** (Agregar).
Lambda admite las siguientes opciones para los orígenes de eventos de DynamoDB:
**Opciones de origen de eventos**
+ **Tabla DynamoDB**: la tabla de DynamoDB de la que leer registros.
+ **Tamaño del lote**: número de registros que se enviarán a la función en cada lote, hasta 10 000. Lambda pasa todos los registros del lote a la función en una sola llamada, siempre y cuando el tamaño total de los eventos no exceda el [límite de carga](gettingstarted-limits.md) para la invocación síncrona (6 MB).
+ **Ventana de lote**: especifique la cantidad de tiempo máxima para recopilar registros antes de invocar la función, en segundos.
+ **Posición inicial**: procesar solo los registros nuevos o todos los registros existentes.
+ **Más recientes**: procesar los registros nuevos que se agreguen al flujo principal.
+ **Horizonte de supresión**: procesar todos los registros del flujo.
Tras procesar cualquier registro existente, la función es alcanzada y continúa procesando registros nuevos.
+ **Destino en caso de error**: una cola de SQS estándar o un tema de SNS estándar para los registros que no se puedan procesar. Cuando Lambda descarta un lote de registros demasiado antiguo o que ha agotado todos los reintentos, Lambda envía detalles sobre el lote a la cola o al tema.
+ **Número de reintentos**: número máximo de reintentos que Lambda realiza cuando la función devuelve un error. Esto no se aplica a errores de servicio o limitaciones controladas en los que el lote no alcanzó la función.
+ **Edad máxima de registro**: antigüedad máxima de un registro que Lambda envía a su función.
+ **División del lote en caso de error**: cuando la función devuelve un error, divida el lote en dos antes de volver a intentarlo. La configuración de tamaño de lote original permanece sin cambios.
+ **Lotes simultáneos por partición**: procese simultáneamente varios lotes desde la misma partición.
+ **Habilitado**: establézcalo en verdadero para habilitar la asignación de orígenes de eventos. Establézcalo en falso para detener el procesamiento de registros. Lambda toma nota del último registro procesado y sigue procesando desde ese punto cuando se habilita de nuevo el mapeo.
**nota**
No se cobran las llamadas a la API GetRecords invocadas por Lambda como parte de los desencadenadores de DynamoDB.
Para administrar la configuración de origen de evento más tarde, elija el desencadenador en el diseñador.
## Creación de asignaciones de orígenes de eventos entre cuentas
Ahora Amazon DynamoDB admite las [políticas basadas en recursos](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/access-control-resource-based.html). Con esta capacidad, puede procesar los datos de un flujo de DynamoDB en una Cuenta de AWS con una función de Lambda en otra cuenta.
Para crear una asignación de orígenes de eventos para la función de Lambda mediante un flujo de DynamoDB en otra Cuenta de AWS, debe configurar el flujo mediante una política basada en recursos para conceder a la función de Lambda permiso para leer registros. Para obtener información sobre cómo configurar el flujo para permitir el acceso entre cuentas, consulte [Acceso compartido con funciones de Lambda entre cuentas](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/rbac-cross-account-access.html#rbac-analyze-cross-account-lambda-access) en la *Guía para desarrolladores de Amazon DynamoDB*.
Una vez que haya configurado el flujo con una política basada en recursos que otorgue a la función de Lambda los permisos necesarios, cree la asignación de orígenes de eventos con el ARN del flujo entre cuentas. Puede encontrar el ARN del flujo en la pestaña **Exportaciones y flujos** de la tabla de la consola de DynamoDB entre cuentas.
Cuando utilice la consola de Lambda, pegue el ARN del flujo directamente en el campo de entrada de la tabla de DynamoDB en la página de creación de asignaciones de orígenes de eventos.
**Nota:** No se admiten los activadores entre regiones.
# Configuración de la respuesta por lotes parcial con DynamoDB y Lambda
Al consumir y procesar datos de streaming desde un origen de eventos, de forma predeterminada, Lambda comprueba hasta el número de secuencia más alto de un lote solo cuando el lote se ha completado con éxito. Lambda trata todos los demás resultados como un error completo y vuelve a intentar procesar el lote hasta el límite de reintentos. Para permitir éxitos parciales al procesar lotes de una secuencia, active `ReportBatchItemFailures`. Permitir éxitos parciales puede ayudar a reducir el número de reintentos en un registro, aunque no impide por completo la posibilidad de reintentos en un registro exitoso.
Para activar `ReportBatchItemFailures`, incluya el valor enumerado **ReportBatchItemFailures** en la lista [FunctionResponseTypes](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-FunctionResponseTypes). Esta lista indica qué tipos de respuesta están habilitados para su función. Puede configurar esta lista al [crear](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) o [actualizar](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) una asignación de orígenes de eventos.
**nota**
Incluso cuando el código de función devuelva respuestas de fallos parciales por lotes, Lambda no procesará estas respuestas a menos que la característica `ReportBatchItemFailures` esté activada explícitamente para la asignación de orígenes de eventos.
## Sintaxis del informe
Al configurar los informes sobre errores de elementos por lotes, la clase `StreamsEventResponse` se devuelve con una lista de errores de elementos de lote. Puede utilizar un objeto `StreamsEventResponse` para devolver el número de secuencia del primer registro fallido del lote. También puede crear su propia clase personalizada usando la sintaxis de respuesta correcta. La siguiente estructura JSON muestra la sintaxis de respuesta requerida:
```
{
"batchItemFailures": [
{
"itemIdentifier": ""
}
]
}
```
**nota**
Si la matriz `batchItemFailures` contiene varios elementos, Lambda usa el registro con el número de secuencia más bajo como punto de control. Luego Lambda vuelve a probar todos los registros a partir de ese punto de control.
## Condiciones de éxito y fracaso
Lambda trata un lote como un éxito completo si devuelve cualquiera de los siguientes elementos:
+ Una lista `batchItemFailure` vacía
+ Una lista `batchItemFailure` nula
+ Una `EventResponse` vacía
+ Un nulo `EventResponse`
Lambda trata un lote como un error completo si devuelve cualquiera de los siguientes elementos:
+ Una cadena `itemIdentifier` vacía
+ Una nula `itemIdentifier`
+ Un `itemIdentifier` con un mal nombre de clave
Lambda reintentos fallidos basados en su estrategia de reintento.
## Bisecar un lote
Si su invocación falla y `BisectBatchOnFunctionError` está activada, el lote se divide en bisectos independientemente de su configuración `ReportBatchItemFailures`.
Cuando se recibe una respuesta de éxito parcial de lote y se activan tanto `BisectBatchOnFunctionError` como `ReportBatchItemFailures`, el lote se divide en el número de secuencia devuelto y Lambda vuelve a intentar solo los registros restantes.
Para simplificar la implementación de la lógica de respuesta parcial por lotes, considere la posibilidad de utilizar la [utilidad Batch Processor](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) de PowertoolsAWS Lambda, que gestiona automáticamente estas complejidades para usted.
Estos son algunos ejemplos de código de función que devuelven la lista de IDs de mensajes fallidos del lote:
------
#### [ .NET ]
**SDK para .NET**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante .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 para Go V2**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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 para Java 2.x**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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 para JavaScript (v3)**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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: [] };
};
```
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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 para PHP**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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 para Python (Boto3)**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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 para Ruby**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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 para Rust**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling).
Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante 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
}
```
------
## Uso de Powertools para el procesador por lotes de AWS Lambda
La utilidad de procesamiento por lotes de Powertools para AWS Lambda gestiona de manera automática la lógica de respuesta parcial de los lotes, lo que reduce la complejidad de implementar la notificación de fallas en los lotes. Estos son algunos ejemplos del uso del procesador por lotes:
**Python**
Para ver ejemplos completos e instrucciones de configuración, consulte la [documentación del procesador por lotes](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/).
Procesamiento de registros de flujos de DynamoDB con un procesador por lotes AWS Lambda.
```
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**
Para ver ejemplos completos e instrucciones de configuración, consulte la [documentación del procesador por lotes](https://docs.aws.amazon.com/powertools/typescript/latest/features/batch/).
Procesamiento de registros de flujos de DynamoDB con un procesador por lotes AWS Lambda.
```
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**
Para ver ejemplos completos e instrucciones de configuración, consulte la [documentación del procesador por lotes](https://docs.powertools.aws.dev/lambda/java/latest/utilities/batch/).
Procesamiento de registros de flujos de DynamoDB con un procesador por lotes AWS Lambda.
```
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**
Para ver ejemplos completos e instrucciones de configuración, consulte la [documentación del procesador por lotes](https://docs.aws.amazon.com/powertools/dotnet/utilities/batch-processing/).
Procesamiento de registros de flujos de DynamoDB AWS Lambdacon un procesador por lotes.
```
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;
}
}
```
# Conservación de registros descartados para un origen de eventos de DynamoDB en Lambda
La gestión de errores en las asignaciones de orígenes de eventos de DynamoDB depende de si el error se produce antes de que se invoque la función o durante la invocación de la función:
+ **Antes de la invocación:** si una asignación de orígenes de eventos de Lambda no puede invocar la función debido a una limitación u otros problemas, lo vuelve a intentar hasta que los registros caduquen o superen la antigüedad máxima configurada en la asignación de orígenes de eventos ([MaximumRecordageInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRecordAgeInSeconds)).
+ **Durante invocación:** [si se invoca la función pero devuelve un error, Lambda vuelve a intentarlo hasta que los registros caduquen, superen la antigüedad [máxima (MaximumRecordageInSeconds) o alcancen la cuota de reintento configurada (MaximumRetryAttempts](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRecordAgeInSeconds)).](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-MaximumRetryAttempts) En el caso de errores de función, también puede configurar [BisectBatchOnFunctionError](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-response-BisectBatchOnFunctionError), que divide un lote fallido en dos lotes más pequeños, aislando registros fallidos y evitando tiempos de espera. La división de lotes no consume la cuota de reintentos.
Si las medidas de administración de errores fallan, Lambda descarta los registros y continúa procesando lotes del flujo. Con la configuración predeterminada, esto significa que un registro incorrecto puede bloquear el procesamiento en la partición afectada durante un máximo de un día. Para evitar esto, configure el mapeo de fuente de eventos de su función con un número razonable de reintentos y una antigüedad máxima de registro que se ajuste a su caso de uso.
## Configuración de destinos para invocaciones fallidas
Para retener los registros de las invocaciones de asignación de orígenes de eventos fallidos, agregue un destino a la asignación de orígenes de eventos de su función. Cada registro enviado al destino es un documento JSON que contiene metadatos sobre la invocación fallida. Para los destinos de Amazon S3, Lambda envía también todo el registro de invocación junto con los metadatos. Puede configurar cualquier tema de Amazon SNS, cola de Amazon SQS, bucket de Amazon S3 o Kafka como destino.
Con los destinos de Amazon S3, puede utilizar la característica de [notificaciones de eventos de Amazon S3](https://docs.aws.amazon.com/) para recibir notificaciones cuando se carguen objetos en el bucket de S3 de destino. También puede configurar las notificaciones de eventos de S3 para que invoquen otra función de Lambda para realizar el procesamiento automatizado de los lotes fallidos.
Su rol de ejecución debe tener permisos para el destino:
+ **En el caso de un destino de SQS:** [sqs:SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html)
+ **En el caso de un destino de SNS:** [sns:Publish](https://docs.aws.amazon.com/sns/latest/api/API_Publish.html)
+ **En el caso de un destino de S3:** [ s3:PutObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html) y [s3:ListBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/ListObjectsV2.html)
+ **Para un destino de Kafka:** [kafka-cluster:WriteData](https://docs.aws.amazon.com/msk/latest/developerguide/kafka-actions.html)
Puede configurar un tema de Kafka como destino en caso de error para las asignaciones de orígenes de eventos de Kafka. Cuando Lambda no puede procesar los registros tras agotar los reintentos o cuando los registros superan la antigüedad máxima, Lambda envía los registros fallidos al tema de Kafka especificado para su posterior procesamiento. Consulte [Uso de un tema de Kafka como destino en caso de error](kafka-on-failure-destination.md).
Si tiene activado el cifrado con su propia clave de KMS para un destino de S3, el rol de ejecución de la función también debe tener permiso para llamar a [kms:GenerateDataKey](https://docs.aws.amazon.com/kms/latest/APIReference/API_GenerateDataKey.html). Si la clave de KMS y el destino del bucket de S3 están en una cuenta diferente a la de su función de Lambda y rol de ejecución, configure la clave de KMS para que confíe en el rol de ejecución y permita kms:GenerateDataKey.
Para configurar un destino en caso de error mediante la consola, siga estos pasos:
1. Abra la [página de Funciones](https://console.aws.amazon.com/lambda/home#/functions) en la consola de Lambda.
1. Elija una función.
1. En **Descripción general de la función**, elija **Agregar destino**.
1. En **Origen**, elija **Invocación de asignación de orígenes de eventos**.
1. Para la **Asignación de orígenes de eventos**, elija un origen de eventos que esté configurado para esta función.
1. En **Condición**, seleccione **En caso de error**. Para las invocaciones de asignación de orígenes de eventos, esta es la única condición aceptada.
1. En **Tipo de destino**, elija el tipo de destino al que Lambda envía los registros de invocación.
1. En **Destino**, elija un recurso.
1. Seleccione **Save**.
También puede configurar un destino en caso de error mediante la AWS Command Line Interface (AWS CLI). Por ejemplo, el siguiente comando [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) agrega una asignación de orígenes de eventos con un destino de SQS en caso de error 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"}}'
```
El siguiente comando [update-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html) actualiza una asignación de orígenes de eventos para enviar registros de invocación fallida a un destino de SNS después de dos intentos de reintento, o si los registros tienen más de una hora de antigüedad.
```
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"}}'
```
La configuración actualizada se aplica de forma asincrónica y no se refleja en la salida hasta que se completa el proceso. Utilice el comando [get-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html) para ver el estado actual.
Para eliminar un destino, introduzca una cadena vacía como argumento del parámetro `destination-config`:
```
aws lambda update-event-source-mapping \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--destination-config '{"OnFailure": {"Destination": ""}}'
```
### Prácticas recomendadas de seguridad para destinos de Amazon S3
Eliminar un bucket de S3 que está configurado como destino sin eliminar el destino de la configuración de la función puede suponer un riesgo de seguridad. Si otro usuario conoce el nombre del bucket de destino, puede volver a crear el bucket en su Cuenta de AWS. Los registros de las invocaciones fallidas se enviarán a su bucket, lo que podría exponer los datos de su función.
**aviso**
Para asegurarse de que los registros de invocación de su función no se puedan enviar a un bucket de S3 de otra Cuenta de AWS, agregue una condición al rol de ejecución de la función que limite los permisos `s3:PutObject` a los buckets de su cuenta.
En el siguiente ejemplo, se muestra una política de IAM que limita los permisos `s3:PutObject` de la función a los buckets de la cuenta. Esta política también otorga a Lambda el permiso `s3:ListBucket` que necesita para usar un bucket de S3 como destino.
```
{
"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"
}
}
}
]
}
```
Para agregar una política de permisos al rol de ejecución de su función mediante la Consola de administración de AWS o la AWS CLI, consulte las instrucciones de los siguientes procedimientos:
------
#### [ Console ]
**Cómo agregar una política de permisos al rol de ejecución de la función (consola)**
1. Abra la página de [Funciones](https://console.aws.amazon.com/lambda/home#/functions) en la consola de Lambda.
1. Seleccione la función de Lambda cuyo rol de ejecución desee modificar.
1. En la pestaña **Configuración**, elija **Permisos**.
1. En la pestaña **Rol de ejecución**, seleccione el **nombre del rol** de la función para abrir la página de la consola de IAM del rol.
1. Agregue una política de permisos al rol de la siguiente manera:
1. En el panel **Política de permisos**, elija **Agregar permisos** y seleccione **Crear política insertada**.
1. En el **editor de políticas**, seleccione **JSON**.
1. Pegue la política que desee agregar en el editor (sustituyendo el JSON existente) y, a continuación, seleccione **Siguiente**.
1. En **Detalles de política**, ingrese un **Nombre de política**.
1. Seleccione **Crear política**.
------
#### [ AWS CLI ]
**Cómo agregar una política de permisos al rol de ejecución de la función (CLI)**
1. Cree un documento de política de JSON con los permisos necesarios y guárdelo en un directorio local.
1. Utilice el comando de la CLI de IAM `put-role-policy` para agregar permisos al rol de ejecución de la función. Ejecute el siguiente comando desde el directorio en el que guardó el documento de política JSON y sustituya el nombre del rol, el nombre de la política y el documento de política por sus propios valores.
```
aws iam put-role-policy \
--role-name my_lambda_role \
--policy-name LambdaS3DestinationPolicy \
--policy-document file://my_policy.json
```
------
### Ejemplo de registro de invocación de Amazon SNS y Amazon SQS
En el siguiente ejemplo, se muestra un registro de invocación que Lambda envía a un destino de SQS o SNS para un flujo de 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"
}
}
```
Puede utilizar esta información para recuperar los registros afectados del flujo para solucionar problemas. Los registros reales no están incluidos, por lo que debe procesar este registro y recuperarlos del flujo antes de que caduquen y se pierdan.
### Ejemplo de registro de invocación de Amazon S3
En el siguiente ejemplo, se muestra un registro de invocación que Lambda envía a un bucket de S3 para un flujo de DynamoDB. Además de todos los campos del ejemplo anterior para los destinos de SQS y SNS, el campo `payload` contiene el registro de invocación original en forma de cadena JSON de 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
}
```
El objeto de S3 que contiene el registro de invocación utiliza la siguiente convención de nomenclatura:
```
aws/lambda///YYYY/MM/DD/YYYY-MM-DDTHH.MM.SS-
```
# Implementación del procesamiento con estado del flujo DynamoDB en Lambda
Las funciones de Lambda pueden ejecutar aplicaciones de procesamiento de flujo continuo. Una secuencia representa datos ilimitados que fluyen de forma continua a través de su aplicación. Para analizar la información de esta entrada de actualización continua, puede enlazar los registros incluidos mediante una ventana definida en términos de tiempo.
Las ventanas de salto constante son ventanas de tiempo distintas que se abren y cierran a intervalos regulares. De forma predeterminada, las invocaciones de Lambda no tienen estado: no se pueden utilizar para procesar datos en múltiples invocaciones continuas sin una base de datos externa. Sin embargo, con las ventanas de salto constante, puede mantener su estado en todas las invocaciones. Este estado contiene el resultado agregado de los mensajes procesados previamente para la ventana actual. Su estado puede ser un máximo de 1 MB por partición. Si supera ese tamaño, Lambda finaliza la ventana antes de tiempo.
Cada registro de una secuencia pertenece a un periodo específico. Lambda procesará cada registro al menos una vez, pero no garantiza que cada registro se procese solo una vez. En casos excepcionales, como el manejo de errores, es posible que algunos registros se procesen más de una vez. Los registros siempre se procesan en orden la primera vez. Si los registros se procesan más de una vez, es posible que lo hagan de forma desordenada.
## Agregación y procesamiento
Su función administrada por el usuario se invoca tanto para la agregación como para procesar los resultados finales de esa agregación. Lambda agrega todos los registros recibidos en la ventana. Puede recibir estos registros en varios lotes, cada uno como una invocación independiente. Cada invocación recibe un estado. Por lo tanto, al usar las ventanas de salto constante, su respuesta de la función de Lambda debe contener una propiedad de `state`. Si la respuesta no contiene una propiedad de `state`, Lambda considera que esto es una invocación fallida. Para satisfacer esta condición, la función puede devolver un objeto de `TimeWindowEventResponse`, que tiene la siguiente forma JSON:
**Example `TimeWindowEventResponse`Valores**
```
{
"state": {
"1": 282,
"2": 715
},
"batchItemFailures": []
}
```
**nota**
Para las funciones Java, se recomienda utilizar un `Map` para representar el estado.
Al final de la ventana, el indicador `isFinalInvokeForWindow` está configurado en `true` para indicar que este es el estado final y que está listo para su procesamiento. Después del procesamiento, la ventana se completa y su invocación final se completa, y luego se elimina el estado.
Al final de la ventana, Lambda utiliza el procesamiento final para las acciones en los resultados de agregación. Su procesamiento final se invoca sincrónicamente. Después de la invocación exitosa, los puntos de control de la función, el número de secuencia y el procesamiento de flujo continúa. Si la invocación no tiene éxito, su función de Lambda suspende el procesamiento posterior hasta una invocación exitosa.
**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
}
```
## Configuración
Puede configurar ventanas de salto constante al crear o actualizar una asignación de orígenes de eventos. Para configurar una ventana de saltos de tamaño constante, especifique la ventana en segundos ([TumblingWindowInSeconds](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-TumblingWindowInSeconds)). El siguiente comando de ejemplo AWS Command Line Interface (AWS CLI) crea una asignación de origen de eventos de streaming que tiene una ventana de salto constante de 120 segundos. Se nombra la función de Lambda definida para la agregación y el procesamiento se llama `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 los límites de la ventana de salto constante en función de la hora en que se insertaron los registros en la secuencia. Todos los registros tienen una marca de hora aproximada disponible que Lambda utiliza en las determinaciones de límites.
Las agregaciones de ventanas de saltos constantes no admiten el reendurecimiento. Cuando el fragmento termina, Lambda considera la ventana cerrada y las particiones secundarias comienzan su propia ventana en un estado fresco.
Ventanas de saltos constantes son totalmente compatibles con las directivas de reintento existentes `maxRetryAttempts` y `maxRecordAge`.
**Example Handler.py: agregación y procesamiento**
La siguiente función de Python muestra cómo agregar y luego procesar su estado 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}
```
# Parámetros de Lambda para las asignaciones de orígenes de eventos de Amazon DynamoDB
Todos los tipos de fuente de eventos Lambda comparten las mismas operaciones [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) y [UpdateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) de la API. Sin embargo, solo algunos de los parámetros se aplican a Amazon DynamoDB Streams.
| Parámetro | Obligatorio | Predeterminado | Notas |
| --- | --- | --- | --- |
| BatchSize | N | 100 | Máximo: 10 000 |
| BisectBatchOnFunctionError | N | false | none |
| DestinationConfig | N | N/A | Cola de Amazon SQS estándar o destino de tema de Amazon SNS estándar para registros descartados |
| Habilitado | N | true | none |
| EventSourceArn | S | N/A | ARN del flujo de datos o un consumidor de flujos |
| FilterCriteria | N | N/A | [Controle qué eventos envía Lambda a la función](invocation-eventfiltering.md) |
| FunctionName | S | N/A | none |
| FunctionResponseTypes | N | N/A | Para permitir que la función informe de errores específicos de un lote, incluya el valor `ReportBatchItemFailures` en `FunctionResponseTypes`. Para obtener más información, consulte [Configuración de la respuesta por lotes parcial con DynamoDB y Lambda](services-ddb-batchfailurereporting.md). |
| MaximumBatchingWindowInSeconds | N | 0 | none |
| MaximumRecordAgeInSeconds | N | -1 | -1 significa infinito: se vuelven a intentar los registros que han producido error hasta que caduque el registro. El [límite de retención de datos del flujo de DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.DataRetention) es de 24 horas. Mínimo: -1 Máximo: 604 800 |
| MaximumRetryAttempts | N | -1 | -1 significa infinito: se vuelven a intentar los registros que han producido error hasta que caduque el registro Mínimo: 0 Máximo: 10 000 |
| ParallelizationFactor | N | 1 | Máximo: 10 |
| StartingPosition | S | N/A | TRIM\$1HORIZON o LATEST |
| TumblingWindowInSeconds | N | N/A | Mínimo: 0 Máximo: 900 |
# Uso del filtrado de eventos con una fuente de eventos de DynamoDB
Puede utilizar el filtrado de eventos para controlar qué registros de un flujo o una cola envía Lambda a su función. Para obtener información general sobre cómo funciona el filtrado de eventos, consulte [Controle qué eventos envía Lambda a la función](invocation-eventfiltering.md).
Esta sección se centra en el filtrado de eventos para las fuentes de eventos de DynamoDB.
**nota**
Las asignaciones de orígenes de eventos de DynamoDB solo admiten el filtrado en la clave `dynamodb`.
**Topics**
+ [evento de DynamoDB](#filtering-ddb)
+ [Filtrar con atributos de tabla](#filtering-ddb-attributes)
+ [Filtrar con expresiones booleanas](#filtering-ddb-boolean)
+ [Uso del operador Exists](#filtering-ddb-exists)
+ [Formato JSON para filtrado de DynamoDB](#filtering-ddb-JSON-format)
## evento de DynamoDB
Supongamos que tiene una tabla de DynamoDB con la clave principal `CustomerName` y los atributos `AccountManager` y `PaymentTerms`. El siguiente es un registro de ejemplo de flujo de la tabla de 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"
}
}
```
Para filtrar en función de los valores de clave y atributos en la tabla de DynamoDB, utilice la clave `dynamodb` del registro. En las siguientes secciones, se muestran ejemplos de diferentes tipos de filtros.
### Filtrar con claves de tabla
Supongamos que desea que su función procese únicamente los registros en los que la clave principal `CustomerName` sea “AnyCompany Industries”. El objeto `FilterCriteria` sería el siguiente.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"
}
]
}
```
Para mayor claridad, este es el valor del `Pattern` del filtro ampliado en JSON no cifrado.
```
{
"dynamodb": {
"Keys": {
"CustomerName": {
"S": [ "AnyCompany Industries" ]
}
}
}
}
```
Puede agregar el filtro mediante la consola, la AWS CLI o una plantilla de AWS SAM.
------
#### [ Console ]
Para agregar este filtro mediante la consola, siga las instrucciones que se indican en [Adjuntar criterios de filtro a una asignación de origen de eventos (consola)](invocation-eventfiltering.md#filtering-console) e ingrese la siguiente cadena para los **Criterios de filtro**.
```
{ "dynamodb" : { "Keys" : { "CustomerName" : { "S" : [ "AnyCompany Industries" ] } } } }
```
------
#### [ AWS CLI ]
Para crear una nueva asignación de orígenes de eventos con estos criterios de filtro mediante la AWS Command Line Interface (AWS CLI), ejecute el siguiente comando.
```
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\" ] } } } }"}]}'
```
Para agregar estos criterios de filtro a una asignación de orígenes de eventos existente, ejecute el siguiente comando.
```
aws lambda update-event-source-mapping \
--uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
--filter-criteria '{"Filters": [{"Pattern": "{ \"dynamodb\" : { \"Keys\" : { \"CustomerName\" : { \"S\" : [ \"AnyCompany Industries\" ] } } } }"}]}'
```
------
#### [ AWS SAM ]
Para agregar este filtro mediante AWS SAM, agregue el siguiente fragmento a la plantilla YAML de su origen de eventos.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "Keys" : { "CustomerName" : { "S" : [ "AnyCompany Industries" ] } } } }'
```
------
## Filtrar con atributos de tabla
Con DynamoDB, también puede utilizar las claves `NewImage` y `OldImage` para filtrar por los valores de los atributos. Supongamos que desea filtrar los registros en los que el atributo `AccountManager` de la última imagen de la tabla sea “Pat Candella” o “Shirley Rodriguez”. El objeto `FilterCriteria` sería el siguiente.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\", \"Shirley Rodriguez\" ] } } } }"
}
]
}
```
Para mayor claridad, este es el valor del `Pattern` del filtro ampliado en JSON no cifrado.
```
{
"dynamodb": {
"NewImage": {
"AccountManager": {
"S": [ "Pat Candella", "Shirley Rodriguez" ]
}
}
}
}
```
Puede agregar el filtro mediante la consola, la AWS CLI o una plantilla de AWS SAM.
------
#### [ Console ]
Para agregar este filtro mediante la consola, siga las instrucciones que se indican en [Adjuntar criterios de filtro a una asignación de origen de eventos (consola)](invocation-eventfiltering.md#filtering-console) e ingrese la siguiente cadena para los **Criterios de filtro**.
```
{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella", "Shirley Rodriguez" ] } } } }
```
------
#### [ AWS CLI ]
Para crear una nueva asignación de orígenes de eventos con estos criterios de filtro mediante la AWS Command Line Interface (AWS CLI), ejecute el siguiente comando.
```
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\" ] } } } }"}]}'
```
Para agregar estos criterios de filtro a una asignación de orígenes de eventos existente, ejecute el siguiente comando.
```
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 ]
Para agregar este filtro mediante AWS SAM, agregue el siguiente fragmento a la plantilla YAML de su origen de eventos.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella", "Shirley Rodriguez" ] } } } }'
```
------
## Filtrar con expresiones booleanas
También puede crear filtros mediante expresiones booleanas AND. Estas expresiones pueden incluir tanto los parámetros de clave como los de atributo de la tabla. Supongamos que desea filtrar los registros en los que el valor `NewImage` de `AccountManager` es “Pat Candella” y el valor `OldImage` es “Terry Whitlock”. El objeto `FilterCriteria` sería el siguiente.
```
{
"Filters": [
{
"Pattern": "{ \"dynamodb\" : { \"NewImage\" : { \"AccountManager\" : { \"S\" : [ \"Pat Candella\" ] } } } , \"dynamodb\" : { \"OldImage\" : { \"AccountManager\" : { \"S\" : [ \"Terry Whitlock\" ] } } } }"
}
]
}
```
Para mayor claridad, este es el valor del `Pattern` del filtro ampliado en JSON no cifrado.
```
{
"dynamodb" : {
"NewImage" : {
"AccountManager" : {
"S" : [
"Pat Candella"
]
}
}
},
"dynamodb": {
"OldImage": {
"AccountManager": {
"S": [
"Terry Whitlock"
]
}
}
}
}
```
Puede agregar el filtro mediante la consola, la AWS CLI o una plantilla de AWS SAM.
------
#### [ Console ]
Para agregar este filtro mediante la consola, siga las instrucciones que se indican en [Adjuntar criterios de filtro a una asignación de origen de eventos (consola)](invocation-eventfiltering.md#filtering-console) e ingrese la siguiente cadena para los **Criterios de filtro**.
```
{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella" ] } } } , "dynamodb" : { "OldImage" : { "AccountManager" : { "S" : [ "Terry Whitlock" ] } } } }
```
------
#### [ AWS CLI ]
Para crear una nueva asignación de orígenes de eventos con estos criterios de filtro mediante la AWS Command Line Interface (AWS CLI), ejecute el siguiente comando.
```
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\" ] } } } } "}]}'
```
Para agregar estos criterios de filtro a una asignación de orígenes de eventos existente, ejecute el siguiente comando.
```
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 ]
Para agregar este filtro mediante AWS SAM, agregue el siguiente fragmento a la plantilla YAML de su origen de eventos.
```
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb" : { "NewImage" : { "AccountManager" : { "S" : [ "Pat Candella" ] } } } , "dynamodb" : { "OldImage" : { "AccountManager" : { "S" : [ "Terry Whitlock" ] } } } }'
```
------
**nota**
El filtrado de eventos de DynamoDB no es compatible con el uso de operadores numéricos (equivalentes numéricos e intervalo numérico). Incluso si los elementos de la tabla se almacenan como números, estos parámetros se convierten en cadenas en el objeto de registro JSON.
## Uso del operador Exists
Debido a la forma en que están estructurados los objetos de eventos JSON de DynamoDB, el uso del operador Exists requiere un cuidado especial. El operador Exists solo funciona en los nodos hoja en el evento JSON, por lo que si el patrón de filtro usa Exists para probar la presencia de un nodo intermedio, no funcionará. Considere el siguiente elemento de la tabla de DynamoDB:
```
{
"UserID": {"S": "12345"},
"Name": {"S": "John Doe"},
"Organizations": {"L": [
{"S":"Sales"},
{"S":"Marketing"},
{"S":"Support"}
]
}
}
```
Es posible que deba crear un patrón de filtro como el siguiente para comprobar si hay eventos que contengan `"Organizations"`:
```
{ "dynamodb" : { "NewImage" : { "Organizations" : [ { "exists": true } ] } } }
```
Sin embargo, este patrón de filtro nunca devolvería una coincidencia porque `"Organizations"` no es un nodo hoja. El siguiente ejemplo muestra cómo utilizar correctamente el operador Exists para crear el patrón de filtro deseado:
```
{ "dynamodb" : { "NewImage" : {"Organizations": {"L": {"S": [ {"exists": true } ] } } } } }
```
## Formato JSON para filtrado de DynamoDB
Para filtrar correctamente los eventos de orígenes de DynamoDB, tanto el campo de datos como los criterios de filtro del campo de datos (`dynamodb`) deben estar en un formato JSON válido. Si el formato JSON de alguno de los campos no es válido, Lambda elimina el mensaje o genera una excepción. En la siguiente tabla se resume el comportamiento específico:
| Formato de los datos entrantes | Formato del patrón de filtro para las propiedades de datos | Acción resultante |
| --- | --- | --- |
| JSON válido | JSON válido | Lambda filtra en función de los criterios de filtro. |
| JSON válido | Sin patrón de filtro para las propiedades de datos | Lambda filtra (solo en las demás propiedades de metadatos) en función de los criterios de filtro. |
| JSON válido | No JSON | Lambda genera una excepción al crear o actualizar la asignación de origen de eventos. El formato JSON del patrón de filtro de las propiedades de datos debe ser válido. |
| No JSON | JSON válido | Lambda elimina el registro. |
| No JSON | Sin patrón de filtro para las propiedades de datos | Lambda filtra (solo en las demás propiedades de metadatos) en función de los criterios de filtro. |
| No JSON | No JSON | Lambda genera una excepción al crear o actualizar la asignación de origen de eventos. El formato JSON del patrón de filtro de las propiedades de datos debe ser válido. |
# Tutorial: Uso de AWS Lambda con Amazon DynamoDB Streams
En este tutorial, se crea una función de Lambda para consumir eventos de Amazon DynamoDB Stream.
## Requisitos previos
### Instala la AWS Command Line Interface
Si aún no ha instalado AWS Command Line Interface, siga los pasos que se indican en [Instalación o actualización de la versión más reciente de AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) para instalarlo.
El tutorial requiere un intérprete de comandos o un terminal de línea de comando para ejecutar los comandos. En Linux y macOS, use su administrador de intérprete de comandos y paquetes preferido.
**nota**
En Windows, algunos comandos de la CLI de Bash que se utilizan habitualmente con Lambda (por ejemplo, `zip`) no son compatibles con los terminales integrados del sistema operativo. Para obtener una versión de Ubuntu y Bash integrada con Windows, [instale el subsistema de Windows para Linux](https://docs.microsoft.com/en-us/windows/wsl/install-win10).
## Creación del rol de ejecución
Cree el [rol de ejecución](lambda-intro-execution-role.md) que concederá a su función permiso para obtener acceso a los recursos de AWS.
**Para crear un rol de ejecución**
1. Abra la [página Roles](https://console.aws.amazon.com/iam/home#/roles) en la consola de IAM.
1. Elija **Crear rol**.
1. Cree un rol con las propiedades siguientes.
+ **Entidad de confianza**: Lambda.
+ **Permisos**: **AWSLambdaDynamoDBExecutionRole**.
+ **Role name (Nombre de rol** – **lambda-dynamodb-role**.
El **AWSLambdaDynamoDBExecutionRole** tiene los permisos que la función necesita para leer elementos desde DynamoDB y escribir registros al CloudWatch Logs.
## Creación de la función
Cree una función de Lambda que procese los eventos de DynamoDB. El código de la función escribe algunos de los datos del evento entrante en CloudWatch Logs.
------
#### [ .NET ]
**SDK para .NET**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante .NET.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text.Json;
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.DynamoDBEvents;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace AWSLambda_DDB;
public class Function
{
public void FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context)
{
context.Logger.LogInformation($"Beginning to process {dynamoEvent.Records.Count} records...");
foreach (var record in dynamoEvent.Records)
{
context.Logger.LogInformation($"Event ID: {record.EventID}");
context.Logger.LogInformation($"Event Name: {record.EventName}");
context.Logger.LogInformation(JsonSerializer.Serialize(record));
}
context.Logger.LogInformation("Stream processing complete.");
}
}
```
------
#### [ Go ]
**SDK para Go V2**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante Go.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-lambda-go/events"
"fmt"
)
func HandleRequest(ctx context.Context, event events.DynamoDBEvent) (*string, error) {
if len(event.Records) == 0 {
return nil, fmt.Errorf("received empty event")
}
for _, record := range event.Records {
LogDynamoDBRecord(record)
}
message := fmt.Sprintf("Records processed: %d", len(event.Records))
return &message, nil
}
func main() {
lambda.Start(HandleRequest)
}
func LogDynamoDBRecord(record events.DynamoDBEventRecord){
fmt.Println(record.EventID)
fmt.Println(record.EventName)
fmt.Printf("%+v\n", record.Change)
}
```
------
#### [ Java ]
**SDK para Java 2.x**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante 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 para JavaScript (v3)**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante 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 de un evento de DynamoDB con Lambda mediante TypeScript.
```
export const handler = async (event, context) => {
console.log(JSON.stringify(event, null, 2));
event.Records.forEach(record => {
logDynamoDBRecord(record);
});
}
const logDynamoDBRecord = (record) => {
console.log(record.eventID);
console.log(record.eventName);
console.log(`DynamoDB Record: ${JSON.stringify(record.dynamodb)}`);
};
```
------
#### [ PHP ]
**SDK para PHP**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante 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 para Python (Boto3)**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante Python.
```
import json
def lambda_handler(event, context):
print(json.dumps(event, indent=2))
for record in event['Records']:
log_dynamodb_record(record)
def log_dynamodb_record(record):
print(record['eventID'])
print(record['eventName'])
print(f"DynamoDB Record: {json.dumps(record['dynamodb'])}")
```
------
#### [ Ruby ]
**SDK para Ruby**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante Ruby.
```
def lambda_handler(event:, context:)
return 'received empty event' if event['Records'].empty?
event['Records'].each do |record|
log_dynamodb_record(record)
end
"Records processed: #{event['Records'].length}"
end
def log_dynamodb_record(record)
puts record['eventID']
puts record['eventName']
puts "DynamoDB Record: #{JSON.generate(record['dynamodb'])}"
end
```
------
#### [ Rust ]
**SDK para Rust**
Hay más en GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de [ejemplos de tecnología sin servidor](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda).
Consumo de un evento de DynamoDB con Lambda mediante 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(())
}
```
------
**Cómo crear la función**
1. Copie el código de muestra en un archivo con el nombre `example.js`.
1. Cree un paquete de implementación.
```
zip function.zip example.js
```
1. Cree una función de Lambda con el 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
```
## Prueba de la función de Lambda
En este paso, invocará la función de Lambda manualmente mediante el comando `invoke` de la CLI de AWS Lambda y el siguiente evento de muestra de DynamoDB. Copie lo siguiente en un archivo denominado `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"
}
]
}
```
Ejecute el comando `invoke` siguiente.
```
aws lambda invoke --function-name ProcessDynamoDBRecords \
--cli-binary-format raw-in-base64-out \
--payload file://input.txt outputfile.txt
```
La opción **cli-binary-format** es obligatoria si va a utilizar la versión 2 de la AWS CLI. Para que esta sea la configuración predeterminada, ejecute `aws configure set cli-binary-format raw-in-base64-out`. Para obtener más información, consulte [Opciones de la línea de comandos globales compatibles con AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-options.html#cli-configure-options-list) en la *Guía del usuario de la AWS Command Line Interface versión 2*.
La función devuelve la cadena `message` en el cuerpo de la respuesta.
Verifique la salida en el archivo `outputfile.txt`.
## Crear una tabla de DynamoDB con un flujo habilitado
Crear una tabla de Amazon DynamoDB con un flujo habilitado.
**Para crear una tabla de DynamoDB**
1. Abra la [consola de DynamoDB](https://console.aws.amazon.com/dynamodb).
1. Seleccione **Create table (Creación de tabla)**.
1. Cree una tabla con la siguiente configuración:
+ **Nombre de la tabla** – **lambda-dynamodb-stream**
+ **Clave principal**: **id** (cadena)
1. Seleccione **Crear**.
**Habilitar secuencias**
1. Abra la [consola de DynamoDB](https://console.aws.amazon.com/dynamodb).
1. Elija **Tables (Tablas)**.
1. Elija la tabla **lambda-dynamodb-stream (Flujo de dynamodb lambda)**.
1. En **Exports and streams** (Exportaciones y flujos), elija **DynamoDB stream details** (Detalles del flujo de DynamoDB).
1. Elija **Turn on**.
1. En **Tipo de vista**, elija **Solo atributos clave**.
1. Seleccione **Activar flujo**.
Anote el ARN del flujo. Lo necesitará en el siguiente paso al asociar el flujo a la función de Lambda. Para obtener más información acerca de cómo habilitar flujos, consulte [Captura de la actividad de las tablas con DynamoDB Streams](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html).
## Añadir un origen de eventos en AWS Lambda
Crear un mapeo de origen de eventos en AWS Lambda. Esta asignación de orígenes de eventos asocia el flujo de DynamoDB a la función de Lambda. Una vez creado este mapeo de origen de eventos, AWS Lambda comienza a sondear el flujo.
Ejecute el siguiente comando `create-event-source-mapping` de la AWS CLI. Después de ejecutar el comando, anote el UUID. Necesitará este UUID para hacer referencia al mapeo de origen de eventos en los comandos, por ejemplo, al eliminar el mapeo de origen de eventos.
```
aws lambda create-event-source-mapping --function-name ProcessDynamoDBRecords \
--batch-size 100 --starting-position LATEST --event-source DynamoDB-stream-arn
```
Esto crea un mapeo entre el flujo de DynamoDB especificado y la función de Lambda. Puede asociar un flujo de DynamoDB a varias funciones de Lambda, así como asociar la misma función de Lambda a varios flujos. Sin embargo, las funciones de Lambda compartirán el rendimiento de lectura para el flujo que comparten.
Para obtener la lista de mapeos de orígenes de eventos, ejecute el siguiente comando.
```
aws lambda list-event-source-mappings
```
La lista devuelve todos los mapeos de orígenes de eventos creados, y para cada uno de ellos muestra el `LastProcessingResult`, entre otras cosas. Este campo se utiliza para proporcionar un mensaje informativo en caso de que surja algún problema. Los valores como `No records processed` (indica que AWS Lambda no ha comenzado el sondeo o que no hay registros en el flujo) y `OK` (indica que AWS Lambda ha leído correctamente los registros del flujo y ha invocado la función de Lambda) indican que no hay ningún problema. Si hay algún problema, recibirá un mensaje de error.
Si tiene muchos mapeos de origen de eventos, use la función nombrar parámetro para reducir los resultados.
```
aws lambda list-event-source-mappings --function-name ProcessDynamoDBRecords
```
## Prueba de la configuración
Probar la experiencia integral. A medida que se actualiza la tabla, DynamoDB escribe los registros de eventos en el flujo. Cuando AWS Lambda sondea el flujo, detecta nuevos registros en él e invoca la función de Lambda en nombre del usuario pasando eventos a esta.
1. En la consola de DynamoDB, agregue, actualice y elimine elementos de la tabla. DynamoDB escribe registros de estas acciones en el flujo.
1. AWS Lambda sondea el flujo y, cuando detecta actualizaciones en este, invoca la función de Lambda pasando los datos de eventos que encuentra en el flujo.
1. La función se ejecuta y crea registros en Amazon CloudWatch. Puede verificar los registros reportados en la consola de Amazon CloudWatch.
## Siguientes pasos
En este instructivo, se muestran los conceptos básicos del procesamiento de eventos de transmisión de DynamoDB con Lambda. En el caso de las cargas de trabajo de producción, puede implementar una lógica de respuesta por lotes parcial para gestionar los errores de registro individuales de forma más eficiente. La [utilidad de procesador por lotes](https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/) de Powertools for AWS Lambda está disponible en Python, TypeScript, .NET y Java y ofrece una solución sólida para este fin, ya que gestiona automáticamente la complejidad de las respuestas de lotes parciales y reduce el número de reintentos para los registros que se han procesado correctamente.
## Eliminación de sus recursos
A menos que desee conservar los recursos que creó para este tutorial, puede eliminarlos ahora. Si elimina los recursos de AWS que ya no utiliza, evitará gastos innecesarios en su Cuenta de AWS.
**Cómo eliminar la función de Lambda**
1. Abra la [página de Funciones](https://console.aws.amazon.com/lambda/home#/functions) en la consola de Lambda.
1. Seleccione la función que ha creado.
1. Elija **Acciones**, **Eliminar**.
1. Escriba **confirm** en el campo de entrada de texto y elija **Delete**(Eliminar).
**Cómo eliminar el rol de ejecución**
1. Abra la página [Roles](https://console.aws.amazon.com/iam/home#/roles) en la consola de IAM.
1. Seleccione el rol de ejecución que creó.
1. Elija **Eliminar**.
1. Si desea continuar, escriba el nombre del rol en el campo de entrada de texto y elija **Delete** (Eliminar).
**Para eliminar la tabla de DynamoDB**
1. Abra la página [Tables (Tablas)](https://console.aws.amazon.com//dynamodb/home#tables:) en la consola de DynamoDB.
1. Seleccione la tabla que ha creado.
1. Elija **Eliminar**.
1. Escriba **delete** en el cuadro de texto.
1. Elija **Delete table (Eliminar tabla)**.