Orquestrando uploads de arquivos dependentes com o AWS Step Functions

O Amazon S3 é um serviço de armazenamento de objetos que muitos clientes usam para armazenamento de arquivos. Com o uso das notificações de eventos do Amazon S3 ou do Amazon EventBridge, os clientes podem criar cargas de trabalho (workloads) com arquitetura orientada a eventos (EDA). Essa arquitetura responde aos eventos produzidos quando ocorrem alterações nos objetos nos buckets do S3.

O EDA envolve comunicação assíncrona entre os componentes do sistema. Isso serve para desacoplar os componentes, permitindo que cada componente seja autônomo.

Alguns cenários podem introduzir acoplamento na arquitetura devido à dependência entre eventos. Este blog apresenta um exemplo comum desse acoplamento e como ele pode ser tratado usando o AWS Step Functions.

Visão geral

Neste exemplo, uma organização tem duas equipes autônomas distribuídas, a equipe de vendas e a equipe de armazém. Cada equipe é responsável por carregar um arquivo de dados mensal em um bucket do S3 para que ele possa ser processado.

Os arquivos geram eventos quando são carregados, iniciando os processos posteriores. O processamento do arquivo do armazém (warehouse) limpa os dados e os une aos dados da equipe de envio (shipping). O processamento do arquivo de vendas correlaciona os dados com os dados combinados de armazém  e envio. Isso permite que os analistas realizem previsões e obtenham outros insights.

Para que essa correlação ocorra, o arquivo do armazém deve ser processado antes do arquivo de vendas. Como as duas equipes são autônomas, não há coordenação entre as equipes. Isso significa que os arquivos podem ser carregados a qualquer momento, sem garantia de que o arquivo do armazém (warehouse) seja processado antes do arquivo de vendas.

Para cenários como esses, o padrão Aggregator pode ser usado. O padrão coleta e armazena os eventos e aciona um novo evento com base nos eventos combinados. No cenário descrito, os eventos combinados são o arquivo armazém processado e o arquivo vendas carregado.

Os requisitos do padrão agregador são:

1. Correlação — Uma forma de agrupar os eventos relacionados. Isso é preenchido por um identificador exclusivo no nome do arquivo.
2. Agregador de eventos — Um armazenamento (storage) persistente para os eventos.
3. Verificação de conclusão e trigger — Uma condição quando a combinação dos eventos são recebidos e uma forma de publicar o evento resultante.

Visão geral da arquitetura

A arquitetura usa os seguintes serviços da AWS:

• Amazon DynamoDB como agregador de eventos.
• Step Functions para orquestrar o fluxo de trabalho.
• AWS Lambda para analisar o nome do arquivo e extrair o identificador de correlação.
• AWS Serverless Application Model (AWS SAM) para infraestrutura como código e implantação.

1. Upload de arquivo: as equipes de vendas e armazém carregam seus respectivos arquivos para o S3.
2. EventBridge: O evento ObjectCreated é enviado para o EventBridge, onde há uma regra com um destino do fluxo de trabalho principal (workflow).
3. State Machine principal: essa state machine orquestra as operações do agregador e o processamento dos arquivos. Ele encapsula os fluxos de trabalho de cada arquivo para separar a lógica do agregador da lógica do fluxo de trabalho dos arquivos.
4. Analisador e correlação de arquivos: a lógica de negócios para identificar o arquivo e seu tipo é executada nessa função do Lambda.
5. Armazenamento com estado: uma tabela do DynamoDB armazena informações sobre o arquivo, como nome, tipo e status de processamento. A state machine lê e grava na tabela do DynamoDB. Os tokens de tarefas também são armazenados nessa tabela.
6. Processamento de arquivos: dependendo do tipo de arquivo e de quaisquer condições prévias, as state machines correspondentes ao tipo de arquivo são executadas. Essas state machines contêm a lógica para processar o arquivo específico.
7. Token de tarefa e retorno de chamada: O token de tarefa é gerado quando o arquivo dependente tenta ser processado antes do arquivo independente. O padrão “Esperar por um retorno de chamada” do Step Functions continua a execução do arquivo dependente após o processamento do arquivo independente.

Passo a passo

Você precisará dos seguintes pré-requisitos:

• AWS CLI e AWS SAM CLI instaladas.
• Uma conta da AWS.
• Permissões suficientes para gerenciar os recursos da AWS.
• Git instalado.

Para implantar o exemplo, siga as instruções no repositório do GitHub.

Este passo a passo mostra o que acontece se o arquivo dependente (arquivo de vendas) for carregado antes do arquivo independente (arquivo de armazém ).

1. O fluxo de trabalho começa com o upload do arquivo de vendas para o bucket dedicado do Sales S3. O exemplo usa buckets S3 separados para os dois arquivos, pois pressupõe que as equipes de vendas e armazém sejam distribuídas e autônomas. Você pode encontrar arquivos de amostra no repositório de código.2. O upload do arquivo para o S3 envia um evento para o EventBridge, no qual a state machine do agregador atua. O padrão de evento usado na regra do EventBridge é:

   {
     "detail-type": ["Object Created"],
     "source": ["aws.s3"],
     "detail": {
       "bucket": {
         "name": ["sales-mfu-eda-09092023", "warehouse-mfu-eda-09092023"]
       },
       "reason": ["PutObject"]
     }
   }

   JSON

   3. A state machine do agregador começa invocando a função Lambda do analisador de arquivos. Essa função analisa o tipo de arquivo e usa o identificador para correlacionar os arquivos. Neste exemplo, o nome do arquivo contém o tipo de arquivo e o identificador de correlação (o ano_mês). Para usar outras formas de representar o tipo de arquivo e o identificador de correlação, você pode modificar essa função para analisar essas informações.
   

4. A próxima etapa na state machine insere um registro do evento na tabela do agregador de eventos do DynamoDB. A tabela tem uma chave primária composta com o identificador de correlação como chave de partição e o tipo de arquivo como chave de classificação. O status de processamento do arquivo é monitorado para fornecer feedback sobre o estado do fluxo de trabalho.

5. Com base no tipo de arquivo, a state machine determina qual ramificação seguir. No exemplo, a ramificação de vendas é executada. A state machine tenta obter o status do arquivo armazém  (dependente) do DynamoDB usando o identificador de correlação. Usando o resultado dessa consulta, a state machine determina se o arquivo armazém  correspondente já foi processado.

6. Como o arquivo Warehouse ainda não foi processado, o padrão de integração WaitForTaskToken é usado. A state machine espera nessa etapa e cria um token de tarefa, que os serviços externos usam para acionar a state machine para continuar sua execução. O registro de vendas na tabela do DynamoDB é atualizado com o token de tarefa.

7. Navegue até o console do S3 e faça o upload do arquivo de exemplo do Warehouse no bucket do armazém  (warehouse) S3. Isso invoca uma nova instância do fluxo de trabalho do Step Functions, que flui pela outra ramificação após a etapa de escolha do tipo de arquivo. Nessa ramificação, a state machine do Warehouse é executada e o status de processamento do arquivo é atualizado no DynamoDB.

Quando o status do arquivo do Warehouse é alterado para “Concluído”, a state machine do armazém  verifica se há um arquivo de vendas pendente no DynamoDB. Se houver, ele recupera o token da tarefa e chama o método sendTaskSuccess. Isso aciona a state machine de vendas, que está em um estado de espera para continuar. A state machine de vendas é iniciada e o status do processamento é atualizado.

Conclusão

Esta postagem mostra como lidar com dependências de arquivos em arquiteturas orientadas a eventos. Você pode personalizar o exemplo fornecido no repositório de código para seu próprio caso de uso.

Essa solução é específica para dependências de arquivos em arquiteturas orientadas por eventos. Para obter mais informações sobre como resolver dependências e agregadores de eventos, leia a postagem do blog: Mudando para arquiteturas orientadas a eventos com agregadores de eventos Serverless.

Para saber mais sobre arquiteturas orientadas a eventos, visite a seção de arquitetura orientada a eventos no Serverless Land.