DynamoDB Streams und Time to Live (TTL)

Sie können Elemente, die durch die Gültigkeitsdauer (TTL) gelöscht wurden, sichern oder anderweitig verarbeiten, indem Sie Amazon DynamoDB Streams in der Tabelle aktivieren und die Streams-Datensätze der abgelaufenen Elemente verarbeiten. Weitere Informationen finden Sie unter Lesen und Verarbeiten eines Streams.

Der Stream-Datensatz enthält ein Benutzeridentitätsfeld Records[<index>].userIdentity.

Die Elemente, die nach Ablauf durch den Gültigkeitsdauer-Prozess gelöscht wurden, haben die folgenden Felder:

• Records[<index>].userIdentity.type

  "Service"

• Records[<index>].userIdentity.principalId

  "dynamodb.amazonaws.com"

Anmerkung

Wenn Sie TTL in einer globalen Tabelle verwenden, wird das Feld für die Region, in der die TTL ausgeführt wurde, festgelegt. userIdentity Dieses Feld wird in anderen Regionen nicht festgelegt, wenn der Löschvorgang repliziert wird.

Der folgende JSON-Ausdruck zeigt den relevanten Teil eines einzelnen Stream-Datensatzes.

"Records": [
    {
        ...

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

        ...

    }
]

Verwenden von DynamoDB Streams und Lambda zum Archivieren von TTL-gelöschten Elementen

Die Kombination von DynamoDB Time to Live (TTL), DynamoDB Streams und AWS -Lambda kann dazu beitragen, die Archivierung von Daten zu vereinfachen, die DynamoDB-Speicherkosten zu senken und die Codekomplexität zu reduzieren. Die Verwendung von Lambda als Stream-Konsument bietet viele Vorteile, insbesondere geringere Kosten im Vergleich zu anderen Konsumenten wie der Kinesis Client Library (KCL). GetRecords-API-Aufrufe in Ihrem DynamoDB-Stream werden Ihnen nicht in Rechnung gestellt, wenn Sie Lambda zum Verarbeiten von Ereignissen verwenden, und Lambda kann eine Ereignisfilterung bieten, indem es JSON-Muster in einem Stream-Ereignis identifiziert. Bei der Inhaltsfilterung nach Ereignismustern können Sie bis zu fünf verschiedene Filter definieren, um zu steuern, welche Ereignisse zur Verarbeitung an Lambda gesendet werden. Dies hilft, die Zahl der Aufrufe Ihrer Lambda-Funktionen zu reduzieren, den Code zu vereinfachen und die Gesamtkosten zu senken.

Zwar enthält DynamoDB Streams alle Datenänderungen, beispielsweise Create-, Modify- und Remove-Aktionen, dies kann jedoch zu unerwünschten Aufrufen Ihrer Lambda-Archivfunktion führen. Angenommen, Sie verfügen über eine Tabelle mit 2 Millionen Datenänderungen pro Stunde, die in den Stream fließen. Allerdings handelt es sich bei weniger als 5 Prozent dieser Änderungen um Elementlöschungen, die während des TTL-Prozesses ablaufen und archiviert werden müssen. Bei Verwendung von Lambda-Ereignisquellenfiltern wird die Lambda-Funktion nur 100 000-mal pro Stunde aufgerufen. Infolge der Ereignisfilterung werden Ihnen anstelle der 2 Millionen Aufrufe, die Sie ohne Ereignisfilterung hätten, nur die erforderlichen Aufrufe in Rechnung gestellt.

Die Ereignisfilterung wird auf die Lambda-Ereignisquellenzuweisung, angewendet. Hierbei handelt es sich um eine Ressource, die aus einem ausgewählten Ereignis – dem DynamoDB-Stream – Daten liest und eine Lambda-Funktion aufruft. Im folgenden Diagramm ist zu sehen, wie ein durch Time to Live gelöschtes Element von einer Lambda-Funktion unter Verwendung von Streams und Ereignisfiltern verarbeitet wird.

Ereignisfiltermuster für DynamoDB Time to Live

Durch Hinzufügen der folgenden JSON zu den Filterkriterien für Ihre Ereignisquellenzuweisung kann Ihre Lambda-Funktion nur für TTL-gelöschte Elemente aufgerufen werden:

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

AWS Lambda Erstellen Sie eine Zuordnung der Ereignisquellen

Verwenden Sie die folgenden Codeausschnitte, um eine gefilterte Ereignisquellenzuweisung zu erstellen, die Sie mit dem DynamoDB-Stream einer Tabelle verbinden können. Jeder Codeblock enthält das Ereignisfiltermuster.

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"]
   }
}