DynamoDB Streams e vida útil

Você pode fazer backup ou processar os itens excluídos por vida útil (TTL) habilitando o Amazon DynamoDB Streams na tabela e processando os registros de fluxos dos itens expirados. Para ter mais informações, consulte Ler e processar um fluxo.

O registro de fluxos contém um campo de identidade do usuário Records[<index>].userIdentity.

Os itens excluídos pelo processo de vida útil após a expiração têm os seguintes campos:

O JSON a seguir mostra a parte relevante de um único registro de fluxos.

"Records": [
    {
        ...

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

        ...

    }
]

Usar o DynamoDB Streams e o Lambda para arquivar itens excluídos do TTL

Combinar a vida útil (TTL) do DynamoDB, o DynamoDB Streams e o AWS Lambda pode ajudar a simplificar os dados de arquivo, reduzir os custos de armazenamento do DynamoDB e diminuir a complexidade do código. O uso do Lambda como consumidor de fluxo oferece muitas vantagens, principalmente a redução de custos em comparação com outros consumidores, como a Kinesis Client Library (KCL). Você não é cobrado por chamadas de API GetRecords no fluxo do DynamoDB ao usar o Lambda para consumir eventos, e o Lambda pode fornecer filtragem de eventos por meio da identificação de padrões JSON em um evento de fluxo. Com a filtragem de conteúdo de padrão de evento, é possível definir até cinco filtros diferentes para controlar quais eventos são enviados ao Lambda para processamento. Isso ajuda a reduzir as invocações de suas funções do Lambda, simplifica o código e diminui o custo geral.

Embora o DynamoDB Streams contenha todas as modificações de dados, como as ações Create, Modify e Remove, isso pode resultar em invocações indesejadas da função do Lambda de arquivo. Por exemplo, digamos que você tenha uma tabela com 2 milhões de modificações de dados por hora fluindo para o fluxo, mas menos de 5% delas são exclusões de itens que expirarão no processo de TTL e precisam ser arquivadas. Com filtros de origem de eventos do Lambda, a função do Lambda invocará apenas 100 mil vezes por hora. O resultado da filtragem de eventos é que você é cobrado apenas pelas invocações necessárias, e não pelos 2 milhões de invocações que você teria sem a filtragem de eventos.

A filtragem de eventos é aplicada ao mapeamento da origem de eventos do Lambda, que é um recurso que lê com base em um evento escolhido (o fluxo do DynamoDB) e invoca uma função do Lambda. No diagrama a seguir, você pode ver como um item excluído por vida útil é consumido por uma função do Lambda usando fluxos e filtros de eventos.

Padrão de filtro de eventos da vida útil do DynamoDB

A adição do JSON a seguir aos critérios de filtro do mapeamento de origem de eventos permite que você invoque sua função do Lambda somente para itens excluídos por TTL:

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

Crie um mapeamento da origem de eventos no AWS Lambda.

Use os trechos de código a seguir para criar um mapeamento de origem de eventos filtrado que você possa conectar ao fluxo do DynamoDB de uma tabela. Cada bloco de código inclui o padrão de filtro de eventos.

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