DynamoDB Streams e Time to Live

Puoi eseguire il backup o altrimenti elaborare gli elementi eliminati da Time to Live (TTL) abilitando Amazon DynamoDB Streams sulla tabella ed elaborando i record dei flussi degli articoli scaduti. Per ulteriori informazioni, consulta Lettura ed elaborazione di un flusso.

Il record Streams contiene un campo di identità utente Records[<index>].userIdentity.

Gli elementi eliminati dal processo Time to Live (TTL, Time to Live) dopo la scadenza hanno i seguenti campi:

• Records[<index>].userIdentity.type

  "Service"

• Records[<index>].userIdentity.principalId

  "dynamodb.amazonaws.com"

Nota

Quando si utilizza TTL in una tabella globale, il campo verrà impostato nella regione in cui TTL è stata eseguita. userIdentity Questo campo non verrà impostato in altre regioni quando l'eliminazione viene replicata.

Di seguito JSON viene mostrata la parte rilevante di un singolo record di stream.

"Records": [
    {
        ...

        "userIdentity": {
            "type": "Service",
            "principalId": "dynamodb.amazonaws.com"
        }

        ...

    }
]

Utilizzo di DynamoDB Streams e Lambda per archiviare gli elementi eliminati TTL

La combinazione di DynamoDB Time to Live TTL (), DynamoDB Streams e Lambda può contribuire a semplificare l'archiviazione dei dati, ridurre i costi di storage DynamoDB AWS e ridurre la complessità del codice. L'utilizzo di Lambda come utente di streaming offre molti vantaggi, in particolare la riduzione dei costi rispetto ad altri consumatori come Kinesis Client Library (). KCL Non ti vengono addebitati costi per GetRecords API le chiamate sul tuo flusso DynamoDB quando usi Lambda per consumare eventi e Lambda può fornire il filtraggio degli eventi identificando i modelli in un evento di flusso. JSON Attraverso il filtraggio dei contenuti del modello di eventi, è possibile definire fino a cinque filtri diversi per controllare quali eventi vengono inviati a Lambda per l'elaborazione. Ciò aiuta a ridurre le invocazioni delle funzioni Lambda, semplifica il codice e riduce i costi complessivi.

Sebbene DynamoDB Streams contenga tutte le modifiche ai dati, ad esempio Create, Modify e le azioni Remove, questo può causare invocazioni indesiderate della funzione Lambda di archivio. Ad esempio, supponiamo di avere una tabella con 2 milioni di modifiche ai dati all'ora che fluiscono nel flusso, ma meno del 5% di queste sono eliminazioni di elementi che scadono durante il processo e devono essere archiviate. TTL Con Filtri di origine evento Lambda, la funzione Lambda richiamerà solo 100.000 volte all'ora. Il risultato con il filtraggio degli eventi è che ti vengono addebitati solo le invocazioni necessarie anziché i 2 milioni di invocazioni che avresti avuto senza filtraggio degli eventi.

Il filtraggio degli eventi viene applicato alla mappatura di origine evento Lambda, una risorsa che legge da un evento scelto, ovvero il flusso DynamoDB, e invoca una funzione Lambda. Nel diagramma seguente, puoi vedere come un elemento eliminato Time to Live viene consumato da una funzione Lambda utilizzando flussi e filtri eventi.

Modello di filtro evento DynamoDB Time to Live

L'aggiunta di quanto segue JSON ai criteri del filtro di mappatura dell'origine degli eventi consente l'invocazione della funzione Lambda solo per gli elementi eliminati: TTL

{
    "Filters": [
        {
            "Pattern": { "userIdentity": { "type": ["Service"], "principalId": ["dynamodb.amazonaws.com"] } }
        }
    ]
}

Crea una mappatura delle sorgenti degli eventi AWS Lambda

Utilizza i seguenti frammenti di codice per creare una mappatura dell'origine eventi filtrata che è possibile connettere al flusso DynamoDB di una tabella. Ciascun blocco di codice include il modello di filtro eventi.

AWS CLI

aws lambda create-event-source-mapping \
--event-source-arn 'arn:aws:dynamodb:eu-west-1:012345678910:table/test/stream/2021-12-10T00:00:00.000' \
--batch-size 10 \
--enabled \
--function-name test_func \
--starting-position LATEST \
--filter-criteria '{"Filters": [{"Pattern": "{\"userIdentity\":{\"type\":[\"Service\"],\"principalId\":[\"dynamodb.amazonaws.com\"]}}"}]}'

Java

LambdaClient client = LambdaClient.builder()
        .region(Region.EU_WEST_1)
        .build();

Filter userIdentity = Filter.builder()
        .pattern("{\"userIdentity\":{\"type\":[\"Service\"],\"principalId\":[\"dynamodb.amazonaws.com\"]}}")
        .build();

FilterCriteria filterCriteria = FilterCriteria.builder()
        .filters(userIdentity)
        .build();

CreateEventSourceMappingRequest mappingRequest = CreateEventSourceMappingRequest.builder()
        .eventSourceArn("arn:aws:dynamodb:eu-west-1:012345678910:table/test/stream/2021-12-10T00:00:00.000")
        .batchSize(10)
        .enabled(Boolean.TRUE)
        .functionName("test_func")
        .startingPosition("LATEST")
        .filterCriteria(filterCriteria)
        .build();

try{
    CreateEventSourceMappingResponse eventSourceMappingResponse = client.createEventSourceMapping(mappingRequest);
    System.out.println("The mapping ARN is "+eventSourceMappingResponse.eventSourceArn());

}catch (ServiceException e){
    System.out.println(e.getMessage());
}

Node

const client = new LambdaClient({ region: "eu-west-1" });

const input = {
    EventSourceArn: "arn:aws:dynamodb:eu-west-1:012345678910:table/test/stream/2021-12-10T00:00:00.000",
    BatchSize: 10,
    Enabled: true,
    FunctionName: "test_func",
    StartingPosition: "LATEST",
    FilterCriteria: { "Filters": [{ "Pattern": "{\"userIdentity\":{\"type\":[\"Service\"],\"principalId\":[\"dynamodb.amazonaws.com\"]}}" }] }
}

const command = new CreateEventSourceMappingCommand(input);

try {
    const results = await client.send(command);
    console.log(results);
} catch (err) {
    console.error(err);
}

Python

session = boto3.session.Session(region_name = 'eu-west-1')
client = session.client('lambda')

try:
    response = client.create_event_source_mapping(
        EventSourceArn='arn:aws:dynamodb:eu-west-1:012345678910:table/test/stream/2021-12-10T00:00:00.000',
        BatchSize=10,
        Enabled=True,
        FunctionName='test_func',
        StartingPosition='LATEST',
        FilterCriteria={
            'Filters': [
                {
                    'Pattern': "{\"userIdentity\":{\"type\":[\"Service\"],\"principalId\":[\"dynamodb.amazonaws.com\"]}}"
                },
            ]
        }
    )
    print(response)
except Exception as e:
    print(e)

JSON

{
  "userIdentity": {
     "type": ["Service"],
     "principalId": ["dynamodb.amazonaws.com"]
   }
}