Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
# Streams e trigger DynamoDB AWS Lambda
Amazon DynamoDB è integrato AWS Lambda in modo da poter *creare* trigger, parti di codice che rispondono automaticamente agli eventi in DynamoDB Streams. Con i trigger è possibile creare applicazioni che rispondono alle modifiche di dati nelle tabelle DynamoDB.
**Topics**
+ [Tutorial \$11: Utilizzo dei filtri per elaborare tutti gli eventi con Amazon DynamoDB e utilizzo di AWS Lambda AWS CLI](Streams.Lambda.Tutorial.md)
+ [Tutorial n. 2: utilizzo di filtri per elaborare alcuni eventi con DynamoDB e Lambda](Streams.Lambda.Tutorial2.md)
+ [Best practice per l’utilizzo dei flussi DynamoDB con Lambda](Streams.Lambda.BestPracticesWithDynamoDB.md)
Se abiliti DynamoDB Streams su una tabella, puoi associare lo stream Amazon Resource Name (ARN) a una funzione che scrivi. AWS Lambda Tutte le operazioni di mutazione su quella tabella DynamoDB possono quindi essere acquisite come elemento nel flusso. Ad esempio, è possibile impostare un trigger in modo che quando un elemento in una tabella viene modificato, nel flusso della tabella venga immediatamente visualizzato un nuovo record.
**Nota**
In caso di sottoscrizione di più di due funzioni Lambda in un flusso DynamoDB, potrebbe verificarsi una limitazione (della larghezza di banda della rete) della lettura.
Il servizio [AWS Lambda](https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html) esegue il polling del flusso alla ricerca di nuovi record quattro volte al secondo. Quando sono disponibili nuovi record di flusso, la funzione Lambda viene richiamata in modo sincrono. Puoi sottoscrivere fino a due funzioni Lambda allo stesso flusso DynamoDB. In caso di sottoscrizione di più di due funzioni Lambda nello stesso flusso DynamoDB, potrebbe verificarsi una limitazione (della larghezza di banda della rete) della lettura.
La funzione Lambda può inviare una notifica, avviare un flusso di lavoro o eseguire numerose altre operazioni specificate. Ad esempio, è possibile scrivere una funzione Lambda semplicemente per copiare ogni record di flusso in un'archiviazione persistente, come il Gateway di file di Amazon S3 (Amazon S3), per creare un percorso di verifica permanente dell'attività di scrittura della tabella. Oppure, supponi di avere un'applicazione di gioco per dispositivi mobili che scrive in una tabella `GameScores`. Quando l'attributo `TopScore` della tabella `GameScores` viene aggiornato, viene scritto un record di flusso corrispondente nel flusso della tabella. Questo evento potrebbe quindi attivare una funzione Lambda che pubblica un messaggio di congratulazioni su un social network. È anche possibile scrivere questa funzione per ignorare tutti i record di flusso che non sono aggiornamenti di `GameScores` o che non modificano l'attributo `TopScore`.
Se la funzione restituisce un errore, Lambda ritenta il batch fino a quando l'elaborazione non riesce o i dati scadono. È inoltre possibile configurare Lambda in modo da riprovare con un batch di dimensioni inferiori, limitare il numero di tentativi, eliminare i record una volta che diventano troppo vecchi e altre opzioni.
Come best practice in materia di prestazioni, la funzione Lambda deve essere di breve durata. Per evitare di introdurre ritardi di elaborazione non necessari, inoltre, non dovrebbe eseguire una logica complessa. In particolare, per un flusso a velocità elevata, è meglio attivare flussi di lavoro Step Function di post-elaborazione asincrona rispetto a funzioni Lambda sincrone a lunga durata.
Puoi utilizzare i trigger Lambda su diversi AWS account configurando una policy basata sulle risorse sul flusso DynamoDB per concedere l'accesso in lettura tra account diversi alla funzione Lambda. Per ulteriori informazioni su come configurare lo stream per consentire l'accesso tra account diversi, consulta [Condividi l'accesso con le funzioni AWS Lambda tra account nella](rbac-cross-account-access.md#shared-access-cross-acount-lambda) DynamoDB Developer Guide.
[Per ulteriori informazioni, consulta la Guida per gli AWS Lambda sviluppatori.AWS Lambda](https://docs.aws.amazon.com/lambda/latest/dg/)
# Tutorial \$11: Utilizzo dei filtri per elaborare tutti gli eventi con Amazon DynamoDB e utilizzo di AWS Lambda AWS CLI
In questo tutorial, creerai un AWS Lambda trigger per elaborare un flusso da una tabella DynamoDB.
**Topics**
+ [Fase 1: creazione di una tabella DynamoDB con un flusso abilitato](#Streams.Lambda.Tutorial.CreateTable)
+ [Fase 2: creazione di un ruolo di esecuzione Lambda](#Streams.Lambda.Tutorial.CreateRole)
+ [Fase 3: creazione di un argomento Amazon SNS](#Streams.Lambda.Tutorial.SNSTopic)
+ [Fase 4: creazione e test di una funzione Lambda](#Streams.Lambda.Tutorial.LambdaFunction)
+ [Fase 5: creazione e test di un trigger](#Streams.Lambda.Tutorial.CreateTrigger)
Lo scenario di questo tutorial è Woofer, un semplice social network. Gli utenti di Woofer comunicano tramite i *bark*, brevi messaggi di testo che vengono inviati ad altri utenti di Woofer. Il seguente diagramma illustra i componenti e il flusso di lavoro di questa applicazione.
![\[Flusso di lavoro dell’applicazione Woofer di una tabella DynamoDB, record dei flussi, funzione Lambda e argomento Amazon SNS.\]](http://docs.aws.amazon.com/it_it/amazondynamodb/latest/developerguide/images/StreamsAndTriggers.png)
1. Un utente scrive un elemento in una tabella DynamoDB (`BarkTable`). Ogni item della tabella rappresenta un bark.
1. Viene scritto un nuovo record di flusso per riflettere l'aggiunta di un nuovo item a `BarkTable`.
1. Il nuovo record di flusso attiva una AWS Lambda funzione (). `publishNewBark`
1. Se il record di flusso indica che è stato aggiunto un nuovo elemento a `BarkTable`, la funzione Lambda legge i dati dal record di flusso e pubblica un messaggio in un argomento di Amazon Simple Notification Service (Amazon SNS).
1. Questo messaggio è ricevuto dai sottoscrittori dell'argomento Amazon SNS. (In questo tutorial, l'unico sottoscrittore è un indirizzo e-mail).
**Prima di iniziare**
Questo tutorial utilizza il. AWS Command Line Interface AWS CLI Se non è stato ancora fatto, seguire le istruzioni contenute nella [Guida per l'utente di AWS Command Line Interface](https://docs.aws.amazon.com/cli/latest/userguide/) per installare e configurare la AWS CLI.
## Fase 1: creazione di una tabella DynamoDB con un flusso abilitato
In questa fase, viene creata crea una tabella DynamoDB (`BarkTable`) per memorizzare tutti i bark degli utenti di Woofer. La chiave primaria è costituita da `Username` (chiave di partizione) e `Timestamp` (chiave di ordinamento). Entrambi questi attributi sono di tipo stringa.
In `BarkTable` è abilitato un flusso. Più avanti in questo tutorial, crei un trigger associando una AWS Lambda funzione allo stream.
1. Immetti il seguente comando per creare la tabella.
```
aws dynamodb create-table \
--table-name BarkTable \
--attribute-definitions AttributeName=Username,AttributeType=S AttributeName=Timestamp,AttributeType=S \
--key-schema AttributeName=Username,KeyType=HASH AttributeName=Timestamp,KeyType=RANGE \
--provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
--stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES
```
1. Nell'output, cerca `LatestStreamArn`.
```
...
"LatestStreamArn": "arn:aws:dynamodb:region:accountID:table/BarkTable/stream/timestamp
...
```
Prendi nota dei valori `region` e `accountID`, in quanto sono necessari nelle altre fasi di questo tutorial.
## Fase 2: creazione di un ruolo di esecuzione Lambda
In questo passaggio, crei un ruolo AWS Identity and Access Management (IAM) (`WooferLambdaRole`) e gli assegni le autorizzazioni. Questo ruolo viene utilizzato dalla funzione Lambda creata in [Fase 4: creazione e test di una funzione Lambda](#Streams.Lambda.Tutorial.LambdaFunction).
Puoi creare anche una policy per il ruolo. La policy conterrà tutte le autorizzazioni necessarie alla funzione Lambda nella fase di runtime.
1. Crea un file denominato `trust-relationship.json` con i seguenti contenuti.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
1. Immetti il seguente comando per creare `WooferLambdaRole`.
```
aws iam create-role --role-name WooferLambdaRole \
--path "/service-role/" \
--assume-role-policy-document file://trust-relationship.json
```
1. Crea un file denominato `role-policy.json` con i seguenti contenuti. (Sostituisci `region` e inserisci `accountID` la tua AWS regione e l'ID dell'account.)
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "arn:aws:logs:us-east-1:111122223333:*"
},
{
"Effect": "Allow",
"Action": [
"dynamodb:DescribeStream",
"dynamodb:GetRecords",
"dynamodb:GetShardIterator",
"dynamodb:ListStreams"
],
"Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/BarkTable/stream/*"
},
{
"Effect": "Allow",
"Action": [
"sns:Publish"
],
"Resource": [
"*"
]
}
]
}
```
------
La policy include quattro dichiarazioni che consentono a `WooferLambdaRole` di eseguire le seguenti operazioni:
+ Eseguire una funzione Lambda (`publishNewBark`). Questa funzione viene creata in una fase successiva di questo tutorial.
+ Accedi ad Amazon CloudWatch Logs. La funzione Lambda scrive la diagnostica nei CloudWatch registri in fase di esecuzione.
+ Leggere i dati dal flusso di DynamoDB per `BarkTable`.
+ Pubblicare i messaggi su Amazon SNS.
1. Immetti il seguente comando per collegare la policy a `WooferLambdaRole`.
```
aws iam put-role-policy --role-name WooferLambdaRole \
--policy-name WooferLambdaRolePolicy \
--policy-document file://role-policy.json
```
## Fase 3: creazione di un argomento Amazon SNS
In questa fase, viene creato un argomento Amazon SNS (`wooferTopic`) e viene registrato un indirizzo e-mail su di esso. La funzione Lambda utilizza questo argomento per pubblicare nuovi bark degli utenti di Woofer.
1. Immettere il seguente comando per creare un nuovo argomento Amazon SNS.
```
aws sns create-topic --name wooferTopic
```
1. Immetti il comando seguente per sottoscrivere un indirizzo e-mail a `wooferTopic`. Sostituisci `region` e `accountID` con la regione e l'ID account AWS e sostituisci `example@example.com` con un indirizzo e-mail valido.
```
aws sns subscribe \
--topic-arn arn:aws:sns:region:accountID:wooferTopic \
--protocol email \
--notification-endpoint example@example.com
```
1. Amazon SNS invia un messaggio di conferma al tuo indirizzo e-mail. Scegli il link **Confirm subscription (Conferma sottoscrizione)** per completare la procedura di sottoscrizione.
## Fase 4: creazione e test di una funzione Lambda
In questo passaggio, crei una AWS Lambda funzione (`publishNewBark`) da cui elaborare i record di flusso. `BarkTable`
La funzione `publishNewBark` elabora solo gli eventi di flusso che corrispondono a nuovi item in `BarkTable`. La funzione legge i dati di questo evento, quindi richiama Amazon SNS perché li pubblichi.
1. Crea un file denominato `publishNewBark.js` con i seguenti contenuti. Sostituisci `region` e `accountID` con la tua AWS regione e l'ID dell'account.
```
'use strict';
var AWS = require("aws-sdk");
var sns = new AWS.SNS();
exports.handler = (event, context, callback) => {
event.Records.forEach((record) => {
console.log('Stream record: ', JSON.stringify(record, null, 2));
if (record.eventName == 'INSERT') {
var who = JSON.stringify(record.dynamodb.NewImage.Username.S);
var when = JSON.stringify(record.dynamodb.NewImage.Timestamp.S);
var what = JSON.stringify(record.dynamodb.NewImage.Message.S);
var params = {
Subject: 'A new bark from ' + who,
Message: 'Woofer user ' + who + ' barked the following at ' + when + ':\n\n ' + what,
TopicArn: 'arn:aws:sns:region:accountID:wooferTopic'
};
sns.publish(params, function(err, data) {
if (err) {
console.error("Unable to send message. Error JSON:", JSON.stringify(err, null, 2));
} else {
console.log("Results from sending message: ", JSON.stringify(data, null, 2));
}
});
}
});
callback(null, `Successfully processed ${event.Records.length} records.`);
};
```
1. Crea un file zip che contenga `publishNewBark.js`. Per fare ciò, se disponi di una utility a riga di comando zip, puoi immettere il comando seguente.
```
zip publishNewBark.zip publishNewBark.js
```
1. Quando si crea la funzione Lambda, si specifica l'Amazon Resource Name (ARN) per `WooferLambdaRole` che hai creato in [Fase 2: creazione di un ruolo di esecuzione Lambda](#Streams.Lambda.Tutorial.CreateRole). Immetti il comando seguente per recuperare questo ARN.
```
aws iam get-role --role-name WooferLambdaRole
```
Nell'output, cerca l'ARN di `WooferLambdaRole`.
```
...
"Arn": "arn:aws:iam::region:role/service-role/WooferLambdaRole"
...
```
Immetti il seguente comando per creare la funzione Lambda. Sostituire *roleARN* con l'ARN per. `WooferLambdaRole`
```
aws lambda create-function \
--region region \
--function-name publishNewBark \
--zip-file fileb://publishNewBark.zip \
--role roleARN \
--handler publishNewBark.handler \
--timeout 5 \
--runtime nodejs16.x
```
1. Ora esegui il test di `publishNewBark` per verificare che funziona. Per fare ciò, fornire un input simile a un record reale da DynamoDB Streams.
Crea un file denominato `payload.json` con i seguenti contenuti. Sostituisci `region` e `accountID` con la Regione AWS e l’ID account.
```
{
"Records": [
{
"eventID": "7de3041dd709b024af6f29e4fa13d34c",
"eventName": "INSERT",
"eventVersion": "1.1",
"eventSource": "aws:dynamodb",
"awsRegion": "region",
"dynamodb": {
"ApproximateCreationDateTime": 1479499740,
"Keys": {
"Timestamp": {
"S": "2016-11-18:12:09:36"
},
"Username": {
"S": "John Doe"
}
},
"NewImage": {
"Timestamp": {
"S": "2016-11-18:12:09:36"
},
"Message": {
"S": "This is a bark from the Woofer social network"
},
"Username": {
"S": "John Doe"
}
},
"SequenceNumber": "13021600000000001596893679",
"SizeBytes": 112,
"StreamViewType": "NEW_IMAGE"
},
"eventSourceARN": "arn:aws:dynamodb:region:account ID:table/BarkTable/stream/2016-11-16T20:42:48.104"
}
]
}
```
Immetti il comando seguente per eseguire il test della funzione `publishNewBark`.
```
aws lambda invoke --function-name publishNewBark --payload file://payload.json --cli-binary-format raw-in-base64-out output.txt
```
Se il test viene superato, viene visualizzato il seguente output.
```
{
"StatusCode": 200,
"ExecutedVersion": "$LATEST"
}
```
Inoltre, il file `output.txt` conterrà il testo seguente.
```
"Successfully processed 1 records."
```
Entro pochi minuti riceverai anche un nuovo messaggio e-mail.
**Nota**
AWS Lambda scrive informazioni diagnostiche su Amazon CloudWatch Logs. Se si verificano errori nella funzione Lambda, è possibile utilizzare queste informazioni diagnostiche per la risoluzione dei problemi:
Apri la CloudWatch console all'indirizzo [https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/).
Nel riquadro di navigazione scegli **Logs (Log)**.
Scegli il seguente gruppo di log: `/aws/lambda/publishNewBark`
Scegli il flusso di log più recente per visualizzare l'output e gli errori della funzione.
## Fase 5: creazione e test di un trigger
In [Fase 4: creazione e test di una funzione Lambda](#Streams.Lambda.Tutorial.LambdaFunction), è stato eseguito il test della funzione Lambda per verificarne la corretta esecuzione. In questa fase, è possibile creare un *trigger* associando la funzione Lambda (`publishNewBark`) a un'origine eventi (il flusso `BarkTable`).
1. Quando crei il trigger, è necessario specificare l'ARN del flusso `BarkTable`. Immetti il comando seguente per recuperare questo ARN.
```
aws dynamodb describe-table --table-name BarkTable
```
Nell'output, cerca `LatestStreamArn`.
```
...
"LatestStreamArn": "arn:aws:dynamodb:region:accountID:table/BarkTable/stream/timestamp
...
```
1. Immetti il comando seguente per creare il trigger. Sostituisci `streamARN` con l'ARN del flusso effettivo.
```
aws lambda create-event-source-mapping \
--region region \
--function-name publishNewBark \
--event-source streamARN \
--batch-size 1 \
--starting-position TRIM_HORIZON
```
1. Esegui il test del trigger. Immetti il comando seguente per aggiungere un elemento a `BarkTable`.
```
aws dynamodb put-item \
--table-name BarkTable \
--item Username={S="Jane Doe"},Timestamp={S="2016-11-18:14:32:17"},Message={S="Testing...1...2...3"}
```
Dovresti ricevere un nuovo messaggio e-mail entro pochi minuti.
1. Aprire la console DynamoDB e aggiungere altri elementi a `BarkTable`. È necessario specificare i valori degli attributi `Username` e `Timestamp`. (Sebbene non sia obbligatorio, si dovrebbe inoltre specificare un valore per `Message`) Dovresti ricevere un nuovo messaggio e-mail per ogni item aggiunto a `BarkTable`.
La funzione Lambda elabora solo i nuovi elementi aggiunti a `BarkTable`. Se aggiorni o elimini un item della tabella, la funzione non esegue alcuna azione.
**Nota**
AWS Lambda scrive informazioni diagnostiche su Amazon CloudWatch Logs. Se si verificano errori nella funzione Lambda, è possibile utilizzare queste informazioni diagnostiche per la risoluzione dei problemi:
Apri la CloudWatch console all'indirizzo [https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/).
Nel riquadro di navigazione scegli **Logs (Log)**.
Scegli il seguente gruppo di log: `/aws/lambda/publishNewBark`
Scegli il flusso di log più recente per visualizzare l'output e gli errori della funzione.
# Tutorial n. 2: utilizzo di filtri per elaborare alcuni eventi con DynamoDB e Lambda
In questo tutorial, creerai un AWS Lambda trigger per elaborare solo alcuni eventi in un flusso da una tabella DynamoDB.
**Topics**
+ [Mettere tutto insieme - CloudFormation](#Streams.Lambda.Tutorial2.Cloudformation)
+ [Mettere tutto insieme - CDK](#Streams.Lambda.Tutorial2.CDK)
Con il [filtro eventi Lambda](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html) è possibile utilizzare espressioni di filtro per controllare quali eventi Lambda invia alla funzione per l'elaborazione. Puoi configurare fino a 5 diversi filtri per i flussi DynamoDB. Se si utilizzano finestre di batch, Lambda applica i criteri di filtro a ogni nuovo evento per stabilire se aggiungerlo al batch corrente.
I filtri vengono applicati tramite strutture chiamate `FilterCriteria`. I 3 attributi principali di `FilterCriteria` sono `metadata properties`, `data properties` e `filter patterns`.
Ecco una struttura di esempio di un evento di flussi DynamoDB:
```
{
"eventID": "c9fbe7d0261a5163fcb6940593e41797",
"eventName": "INSERT",
"eventVersion": "1.1",
"eventSource": "aws:dynamodb",
"awsRegion": "us-east-2",
"dynamodb": {
"ApproximateCreationDateTime": 1664559083.0,
"Keys": {
"SK": { "S": "PRODUCT#CHOCOLATE#DARK#1000" },
"PK": { "S": "COMPANY#1000" }
},
"NewImage": {
"quantity": { "N": "50" },
"company_id": { "S": "1000" },
"fabric": { "S": "Florida Chocolates" },
"price": { "N": "15" },
"stores": { "N": "5" },
"product_id": { "S": "1000" },
"SK": { "S": "PRODUCT#CHOCOLATE#DARK#1000" },
"PK": { "S": "COMPANY#1000" },
"state": { "S": "FL" },
"type": { "S": "" }
},
"SequenceNumber": "700000000000888747038",
"SizeBytes": 174,
"StreamViewType": "NEW_AND_OLD_IMAGES"
},
"eventSourceARN": "arn:aws:dynamodb:us-east-2:111122223333:table/chocolate-table-StreamsSampleDDBTable-LUOI6UXQY7J1/stream/2022-09-30T17:05:53.209"
}
```
`metadata properties` sono i campi dell'oggetto evento. Nel caso di flussi DynamoDB, `metadata properties` sono campi come `dynamodb` o `eventName`.
`data properties` sono i campi del corpo dell'evento. Per filtrare su `data properties`, bisogna assicurarsi di contenerli in `FilterCriteria` all'interno della chiave appropriata. Per le origini eventi Dynamo DB, la chiave dati è `NewImage` o `OldImage`.
Infine, le regole di filtro definiranno l'espressione del filtro che si desidera applicare a una proprietà specifica. Ecco alcuni esempi:
| Operatore di confronto | Esempio | Sintassi delle regole (parziale) |
| --- | --- | --- |
| Null | Il tipo di prodotto è null | `{ "product_type": { "S": null } } ` |
| Empty | Il nome del prodotto è vuoto | `{ "product_name": { "S": [ ""] } } ` |
| Equals | Lo stato equivale a Florida | `{ "state": { "S": ["FL"] } } ` |
| And | Lo stato del prodotto equivale a Florida e la categoria di prodotto è Chocolate | `{ "state": { "S": ["FL"] } , "category": { "S": [ "CHOCOLATE"] } } ` |
| Or | Lo stato del prodotto è Florida o California | `{ "state": { "S": ["FL","CA"] } } ` |
| Not | Lo stato del prodotto non è Florida | `{"state": {"S": [{"anything-but": ["FL"]}]}}` |
| Exists | Esiste il prodotto artigianale | `{"homemade": {"S": [{"exists": true}]}}` |
| Does not exist | Il prodotto "homemade" non esiste | `{"homemade": {"S": [{"exists": false}]}}` |
| Begins with | PK inizia con COMPANY | `{"PK": {"S": [{"prefix": "COMPANY"}]}}` |
Per una funzione Lambda è possibile specificare fino a 5 modelli di filtro eventi. Si noti che ognuno di questi 5 eventi verrà valutato come un OR logico. Quindi, se configuri due filtri denominati `Filter_One` e `Filter_Two`, la funzione Lambda eseguirà `Filter_One` OR `Filter_Two`.
**Nota**
Nella pagina di [filtraggio degli eventi Lambda](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html) ci sono alcune opzioni per filtrare e confrontare valori numerici, tuttavia nel caso degli eventi di filtro DynamoDB ciò non si applica perché i numeri in DynamoDB vengono memorizzati come stringhe. Ad esempio ` "quantity": { "N": "50" }`, sappiamo che è un numero a causa della proprietà `"N"`.
## Mettere tutto insieme - CloudFormation
Per mostrare in pratica la funzionalità di filtraggio degli eventi, ecco un CloudFormation modello di esempio. Questo modello genererà una tabella DynamoDB semplice con una chiave di partizione PK e una chiave di ordinamento SK con flussi Amazon DynamoDB abilitati. Creerà una funzione Lambda e un semplice ruolo di esecuzione Lambda che consentirà di scrivere registri su Amazon Cloudwatch e leggere gli eventi dai flussi Amazon DynamoDB. Aggiungerà anche la mappatura dell'origine eventi tra flussi DynamoDB e la funzione Lambda, in modo che la funzione possa essere eseguita ogni volta che c'è un evento nei flussi Amazon DynamoDB.
```
AWSTemplateFormatVersion: "2010-09-09"
Description: Sample application that presents AWS Lambda event source filtering
with Amazon DynamoDB Streams.
Resources:
StreamsSampleDDBTable:
Type: AWS::DynamoDB::Table
Properties:
AttributeDefinitions:
- AttributeName: "PK"
AttributeType: "S"
- AttributeName: "SK"
AttributeType: "S"
KeySchema:
- AttributeName: "PK"
KeyType: "HASH"
- AttributeName: "SK"
KeyType: "RANGE"
StreamSpecification:
StreamViewType: "NEW_AND_OLD_IMAGES"
ProvisionedThroughput:
ReadCapacityUnits: 5
WriteCapacityUnits: 5
LambdaExecutionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: "2012-10-17",
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
Action:
- sts:AssumeRole
Path: "/"
Policies:
- PolicyName: root
PolicyDocument:
Version: "2012-10-17",
Statement:
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: arn:aws:logs:*:*:*
- Effect: Allow
Action:
- dynamodb:DescribeStream
- dynamodb:GetRecords
- dynamodb:GetShardIterator
- dynamodb:ListStreams
Resource: !GetAtt StreamsSampleDDBTable.StreamArn
EventSourceDDBTableStream:
Type: AWS::Lambda::EventSourceMapping
Properties:
BatchSize: 1
Enabled: True
EventSourceArn: !GetAtt StreamsSampleDDBTable.StreamArn
FunctionName: !GetAtt ProcessEventLambda.Arn
StartingPosition: LATEST
ProcessEventLambda:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.7
Timeout: 300
Handler: index.handler
Role: !GetAtt LambdaExecutionRole.Arn
Code:
ZipFile: |
import logging
LOGGER = logging.getLogger()
LOGGER.setLevel(logging.INFO)
def handler(event, context):
LOGGER.info('Received Event: %s', event)
for rec in event['Records']:
LOGGER.info('Record: %s', rec)
Outputs:
StreamsSampleDDBTable:
Description: DynamoDB Table ARN created for this example
Value: !GetAtt StreamsSampleDDBTable.Arn
StreamARN:
Description: DynamoDB Table ARN created for this example
Value: !GetAtt StreamsSampleDDBTable.StreamArn
```
Dopo aver distribuito questo modello di CloudFormation, puoi inserire il seguente elemento Amazon DynamoDB:
```
{
"PK": "COMPANY#1000",
"SK": "PRODUCT#CHOCOLATE#DARK",
"company_id": "1000",
"type": "",
"state": "FL",
"stores": 5,
"price": 15,
"quantity": 50,
"fabric": "Florida Chocolates"
}
```
Grazie alla semplice funzione lambda inclusa in linea in questo modello di formazione cloud, vedrai gli eventi nei gruppi di CloudWatch log di Amazon per la funzione lambda come segue:
```
{
"eventID": "c9fbe7d0261a5163fcb6940593e41797",
"eventName": "INSERT",
"eventVersion": "1.1",
"eventSource": "aws:dynamodb",
"awsRegion": "us-east-2",
"dynamodb": {
"ApproximateCreationDateTime": 1664559083.0,
"Keys": {
"SK": { "S": "PRODUCT#CHOCOLATE#DARK#1000" },
"PK": { "S": "COMPANY#1000" }
},
"NewImage": {
"quantity": { "N": "50" },
"company_id": { "S": "1000" },
"fabric": { "S": "Florida Chocolates" },
"price": { "N": "15" },
"stores": { "N": "5" },
"product_id": { "S": "1000" },
"SK": { "S": "PRODUCT#CHOCOLATE#DARK#1000" },
"PK": { "S": "COMPANY#1000" },
"state": { "S": "FL" },
"type": { "S": "" }
},
"SequenceNumber": "700000000000888747038",
"SizeBytes": 174,
"StreamViewType": "NEW_AND_OLD_IMAGES"
},
"eventSourceARN": "arn:aws:dynamodb:us-east-2:111122223333:table/chocolate-table-StreamsSampleDDBTable-LUOI6UXQY7J1/stream/2022-09-30T17:05:53.209"
}
```
**Esempi di filtri**
+ **Solo prodotti che corrispondono a un determinato stato**
Questo esempio modifica il CloudFormation modello per includere un filtro per abbinare tutti i prodotti provenienti dalla Florida, con l'abbreviazione «FL».
```
EventSourceDDBTableStream:
Type: AWS::Lambda::EventSourceMapping
Properties:
BatchSize: 1
Enabled: True
FilterCriteria:
Filters:
- Pattern: '{ "dynamodb": { "NewImage": { "state": { "S": ["FL"] } } } }'
EventSourceArn: !GetAtt StreamsSampleDDBTable.StreamArn
FunctionName: !GetAtt ProcessEventLambda.Arn
StartingPosition: LATEST
```
Dopo aver ridistribuito lo stack, puoi aggiungere il seguente elemento DynamoDB alla tabella. Si noti che non verrà visualizzato nei registri delle funzioni Lambda, poiché il prodotto in questo esempio proviene dalla California.
```
{
"PK": "COMPANY#1000",
"SK": "PRODUCT#CHOCOLATE#DARK#1000",
"company_id": "1000",
"fabric": "Florida Chocolates",
"price": 15,
"product_id": "1000",
"quantity": 50,
"state": "CA",
"stores": 5,
"type": ""
}
```
+ **Solo gli elementi che iniziano con alcuni valori in PK e SK**
Questo esempio modifica il CloudFormation modello per includere la seguente condizione:
```
EventSourceDDBTableStream:
Type: AWS::Lambda::EventSourceMapping
Properties:
BatchSize: 1
Enabled: True
FilterCriteria:
Filters:
- Pattern: '{"dynamodb": {"Keys": {"PK": { "S": [{ "prefix": "COMPANY" }] },"SK": { "S": [{ "prefix": "PRODUCT" }] }}}}'
EventSourceArn: !GetAtt StreamsSampleDDBTable.StreamArn
FunctionName: !GetAtt ProcessEventLambda.Arn
StartingPosition: LATEST
```
Si noti che la condizione AND richiede che la condizione sia all'interno del modello, dove le chiavi PK e SK sono nella stessa espressione separate da una virgola.
O inizia con alcuni valori su PK e SK o proviene da un determinato stato.
Questo esempio modifica il CloudFormation modello per includere le seguenti condizioni:
```
EventSourceDDBTableStream:
Type: AWS::Lambda::EventSourceMapping
Properties:
BatchSize: 1
Enabled: True
FilterCriteria:
Filters:
- Pattern: '{"dynamodb": {"Keys": {"PK": { "S": [{ "prefix": "COMPANY" }] },"SK": { "S": [{ "prefix": "PRODUCT" }] }}}}'
- Pattern: '{ "dynamodb": { "NewImage": { "state": { "S": ["FL"] } } } }'
EventSourceArn: !GetAtt StreamsSampleDDBTable.StreamArn
FunctionName: !GetAtt ProcessEventLambda.Arn
StartingPosition: LATEST
```
Si noti che la condizione OR viene aggiunta introducendo nuovi modelli nella sezione del filtro.
## Mettere tutto insieme - CDK
Il seguente modello di formazione del progetto CDK di esempio illustra la funzionalità di filtro degli eventi. Prima di lavorare con questo progetto CDK è necessario [installare i prerequisiti](https://docs.aws.amazon.com/cdk/v2/guide/work-with.html), inclusa l'[esecuzione degli script di preparazione](https://docs.aws.amazon.com/cdk/v2/guide/work-with-cdk-python.html).
**Creazione di un progetto CDK**
Per prima cosa crea un nuovo AWS CDK progetto, invocandolo `cdk init` in una directory vuota.
```
mkdir ddb_filters
cd ddb_filters
cdk init app --language python
```
Il comando `cdk init` utilizza il nome della cartella del progetto per denominare vari elementi del progetto, tra cui classi, sottocartelle e file. Tutti i trattini nel nome della cartella vengono convertiti in caratteri di sottolineatura. Altrimenti il nome dovrebbe seguire il formato di un identificatore Python. Ad esempio, non dovrebbe iniziare con un numero o contenere spazi.
Per lavorare con il nuovo progetto, attivare il suo ambiente virtuale. Ciò consente di installare le dipendenze del progetto localmente nella cartella del progetto, anziché globalmente.
```
source .venv/bin/activate
python -m pip install -r requirements.txt
```
**Nota**
Potresti riconoscerlo come il Mac/Linux comando per attivare un ambiente virtuale. I modelli Python includono un file batch, `source.bat`, che consente di utilizzare lo stesso comando su Windows. Funziona anche il comando Windows tradizionale `.venv\Scripts\activate.bat`. Se hai inizializzato il tuo AWS CDK progetto utilizzando AWS CDK Toolkit v1.70.0 o precedente, il tuo ambiente virtuale si trova invece nella directory. `.env` `.venv`
**Infrastruttura di base**
Apri il file `./ddb_filters/ddb_filters_stack.py` con l'editor di testo preferito. Questo file è stato generato automaticamente al momento della creazione del AWS CDK progetto.
Quindi, aggiungi le funzioni `_create_ddb_table` e `_set_ddb_trigger_function`. Queste funzioni creeranno una tabella DynamoDB con chiave di partizione PK e una chiave di ordinamento SK in modalità di assegnazione in modalità on-demand, con flussi Amazon DynamoDB abilitato per impostazione predefinita per mostrare immagini nuove e vecchie.
La funzione Lambda verrà archiviata nella cartella `lambda` nel file `app.py`. Questo file verrà creato in seguito. Comprenderà una variabile di ambiente `APP_TABLE_NAME`, che sarà il nome della tabella Amazon DynamoDB creata da questo stack. Nella stessa funzione concederemo alla funzione Lambda le autorizzazioni di lettura del flusso. Infine, verrà effettuata la sottoscrizione a flussi DynamoDB come origine degli eventi per la funzione Lambda.
Alla fine del file nel metodo `__init__`, richiamerai i rispettivi costrutti per inizializzarli nello stack. Per progetti più grandi che richiedono componenti e servizi aggiuntivi, potrebbe essere meglio definire questi costrutti al di fuori dello stack di base.
```
import os
import json
import aws_cdk as cdk
from aws_cdk import (
Stack,
aws_lambda as _lambda,
aws_dynamodb as dynamodb,
)
from constructs import Construct
class DdbFiltersStack(Stack):
def _create_ddb_table(self):
dynamodb_table = dynamodb.Table(
self,
"AppTable",
partition_key=dynamodb.Attribute(
name="PK", type=dynamodb.AttributeType.STRING
),
sort_key=dynamodb.Attribute(
name="SK", type=dynamodb.AttributeType.STRING),
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
removal_policy=cdk.RemovalPolicy.DESTROY,
)
cdk.CfnOutput(self, "AppTableName", value=dynamodb_table.table_name)
return dynamodb_table
def _set_ddb_trigger_function(self, ddb_table):
events_lambda = _lambda.Function(
self,
"LambdaHandler",
runtime=_lambda.Runtime.PYTHON_3_9,
code=_lambda.Code.from_asset("lambda"),
handler="app.handler",
environment={
"APP_TABLE_NAME": ddb_table.table_name,
},
)
ddb_table.grant_stream_read(events_lambda)
event_subscription = _lambda.CfnEventSourceMapping(
scope=self,
id="companyInsertsOnlyEventSourceMapping",
function_name=events_lambda.function_name,
event_source_arn=ddb_table.table_stream_arn,
maximum_batching_window_in_seconds=1,
starting_position="LATEST",
batch_size=1,
)
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
ddb_table = self._create_ddb_table()
self._set_ddb_trigger_function(ddb_table)
```
Ora creeremo una funzione lambda molto semplice che stamperà i log in Amazon. CloudWatch Per farlo, crea una nuova cartella denominata `lambda`.
```
mkdir lambda
touch app.py
```
Usando l'editor di testo preferito, aggiungi il seguente contenuto al file `app.py`:
```
import logging
LOGGER = logging.getLogger()
LOGGER.setLevel(logging.INFO)
def handler(event, context):
LOGGER.info('Received Event: %s', event)
for rec in event['Records']:
LOGGER.info('Record: %s', rec)
```
Assicurati di essere nella cartella `/ddb_filters/`, digita il seguente comando per creare l'applicazione di esempio:
```
cdk deploy
```
A un certo punto ti verrà chiesto di confermare se desideri implementare la soluzione. Accetta le modifiche digitando `Y`.
```
├───┼──────────────────────────────┼────────────────────────────────────────────────────────────────────────────────┤
│ + │ ${LambdaHandler/ServiceRole} │ arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole │
└───┴──────────────────────────────┴────────────────────────────────────────────────────────────────────────────────┘
Do you wish to deploy these changes (y/n)? y
...
✨ Deployment time: 67.73s
Outputs:
DdbFiltersStack.AppTableName = DdbFiltersStack-AppTable815C50BC-1M1W7209V5YPP
Stack ARN:
arn:aws:cloudformation:us-east-2:111122223333:stack/DdbFiltersStack/66873140-40f3-11ed-8e93-0a74f296a8f6
```
Una volta implementate le modifiche, apri la AWS console e aggiungi un elemento alla tabella.
```
{
"PK": "COMPANY#1000",
"SK": "PRODUCT#CHOCOLATE#DARK",
"company_id": "1000",
"type": "",
"state": "FL",
"stores": 5,
"price": 15,
"quantity": 50,
"fabric": "Florida Chocolates"
}
```
I CloudWatch log dovrebbero ora contenere tutte le informazioni di questa voce.
**Esempi di filtri**
+ **Solo prodotti che corrispondono a un determinato stato**
Apri il file `ddb_filters/ddb_filters/ddb_filters_stack.py` e modificalo per includere il filtro che corrisponde a tutti i prodotti equivalenti a "FL". Questo può essere modificato appena sotto `event_subscription` nella riga 45.
```
event_subscription.add_property_override(
property_path="FilterCriteria",
value={
"Filters": [
{
"Pattern": json.dumps(
{"dynamodb": {"NewImage": {"state": {"S": ["FL"]}}}}
)
},
]
},
)
```
+ **Solo gli elementi che iniziano con alcuni valori in PK e SK**
Modifica lo script Python per includere la seguente condizione:
```
event_subscription.add_property_override(
property_path="FilterCriteria",
value={
"Filters": [
{
"Pattern": json.dumps(
{
{
"dynamodb": {
"Keys": {
"PK": {"S": [{"prefix": "COMPANY"}]},
"SK": {"S": [{"prefix": "PRODUCT"}]},
}
}
}
}
)
},
]
},
```
+ **O inizia con alcuni valori su PK e SK o proviene da un determinato stato.**
Modifica lo script Python per includere le seguenti condizioni:
```
event_subscription.add_property_override(
property_path="FilterCriteria",
value={
"Filters": [
{
"Pattern": json.dumps(
{
{
"dynamodb": {
"Keys": {
"PK": {"S": [{"prefix": "COMPANY"}]},
"SK": {"S": [{"prefix": "PRODUCT"}]},
}
}
}
}
)
},
{
"Pattern": json.dumps(
{"dynamodb": {"NewImage": {"state": {"S": ["FL"]}}}}
)
},
]
},
)
```
Si noti che la condizione OR viene aggiunta aggiungendo altri elementi all'array Filters.
**Pulizia**
Individua lo stack di filtri nella base della tua directory di lavoro ed esegui `cdk destroy`. Ti verrà chiesto di confermare l'eliminazione della risorsa:
```
cdk destroy
Are you sure you want to delete: DdbFiltersStack (y/n)? y
```
# Best practice per l’utilizzo dei flussi DynamoDB con Lambda
Una AWS Lambda funzione viene eseguita all'interno di un *contenitore*, un ambiente di esecuzione isolato da altre funzioni. Quando esegui una funzione per la prima volta, AWS Lambda crea un nuovo contenitore e inizia a eseguire il codice della funzione.
Una funzione Lambda dispone di un *gestore* che viene eseguito a ogni richiamo. Il gestore contiene la logica di business principale della funzione. Ad esempio, la funzione Lambda illustrata in [Fase 4: creazione e test di una funzione Lambda](Streams.Lambda.Tutorial.md#Streams.Lambda.Tutorial.LambdaFunction) dispone di un gestore in grado di elaborare record in un flusso DynamoDB.
Puoi anche fornire un codice di inizializzazione che venga eseguito una sola volta, dopo la creazione del contenitore, ma prima che il gestore venga AWS Lambda eseguito per la prima volta. La funzione Lambda mostrata in [Fase 4: creazione e test di una funzione Lambda](Streams.Lambda.Tutorial.md#Streams.Lambda.Tutorial.LambdaFunction) ha un codice di inizializzazione che importa l'SDK per JavaScript Node.js e crea un client per Amazon SNS. Questi oggetti dovrebbero essere definiti una sola volta, esternamente al gestore.
Dopo l'esecuzione della funzione, AWS Lambda potrebbe scegliere di riutilizzare il contenitore per le successive chiamate della funzione. In questo caso, il gestore della funzione potrebbe essere in grado di utilizzare nuovamente le risorse definite nel codice di inizializzazione. (Non puoi controllare per quanto tempo AWS Lambda conserva il container, né se questo verrà nuovamente utilizzato o meno).
Per l' AWS Lambda utilizzo dei trigger DynamoDB, consigliamo quanto segue:
+ AWS i client di servizio devono essere istanziati nel codice di inizializzazione, non nel gestore. Ciò consente di AWS Lambda riutilizzare le connessioni esistenti, per tutta la durata del contenitore.
+ In generale, non è necessario gestire in modo esplicito le connessioni o implementare il pool di connessioni perché lo AWS Lambda gestisce per te.
Un consumer Lambda per un flusso DynamoDB non garantisce una consegna esatta di una sola volta e può portare a duplicati occasionali. Assicurati che il codice della funzione Lambda sia idempotente per evitare che si verifichino problemi imprevisti dovuti all’elaborazione duplicata.
Per ulteriori informazioni, consulta [Best practice per l'utilizzo delle AWS Lambda funzioni](https://docs.aws.amazon.com/lambda/latest/dg/best-practices.html) nella *AWS Lambda Developer* Guide.