Processar mensagens do Amazon MSK com o Lambda
nota
Se você deseja enviar dados para um destino que não seja uma função do Lambda ou enriquecer os dados antes de enviá-los, consulte Amazon EventBridge Pipes (Pipes do Amazon EventBridge).
Tópicos
- Adicionar o Amazon MSK como uma origem de evento
- Parâmetros de configuração do Amazon MSK
- Criar mapeamentos da origem do evento entre contas
- Usar um cluster do Amazon MSK como uma origem de eventos
- Posições iniciais de sondagem e fluxo
- Métricas do Amazon CloudWatch
- Auto scaling da origem do evento do Amazon MSK
Adicionar o Amazon MSK como uma origem de evento
Para criar um mapeamento de origem de evento, adicione o Amazon MSK como um acionador de função do Lambda usando o console do Lambda, um AWS SDK
Esta seção descreve como criar um mapeamento de origens de eventos usando o console do Lambda e oAWS CLI.
Pré-requisitos
-
Um cluster do Amazon MSK e um tópico do Kafka. Para obter mais informações, consulte Conceitos básicos do uso do Amazon MSK no Guia do desenvolvedor do Amazon Managed Streaming for Apache Kafka.
-
Uma função de execução com permissão para acessar os recursos da AWS que seu cluster do MSK usa.
ID de grupo de consumidores personalizável
Ao configurar o Kafka como uma origem de eventos, você pode especificar um ID de grupo de consumidores. Esse ID de grupo de consumidores é um identificador existente para o grupo de consumidores do Kafka no qual você deseja que a função do Lambda ingresse. Você pode usar esse recurso para migrar facilmente qualquer configuração de processamento em andamento de registros do Kafka de outros consumidores para o Lambda.
Se você especificar um ID de grupo de consumidores e houver outros pesquisadores ativos dentro desse grupo de consumidores, o Kafka distribuirá mensagens entre todos os consumidores. Em outras palavras, o Lambda não receberá todas as mensagens para o tópico do Kafka. Se você quiser que o Lambda gerencie todas as mensagens do tópico, desative todos os outros pesquisadores desse grupo de consumidores.
Além disso, se você especificar um ID de grupo de consumidores e o Kafka encontrar um grupo de consumidores válido já existente com o mesmo ID, o Lambda ignorará o parâmetro StartingPosition
no mapeamento de origem de eventos. Em vez disso, o Lambda começará a processar registros de acordo com o deslocamento confirmado do grupo de consumidores. Se você especificar um ID de grupo de consumidores e o Kafka não conseguir encontrar um grupo de consumidores existente, o Lambda configurará a origem de eventos com a StartingPosition
especificada.
O ID do grupo de consumidores que você especificar deverá ser exclusivo entre todas as origens de eventos do Kafka. Após criar um mapeamento de origem de eventos do Kafka com o ID do grupo de consumidores especificado, você não poderá atualizar esse valor.
Adicionar um gatilho do Amazon MSK (console)
Siga estas etapas para adicionar seu cluster do Amazon MSK e um tópico do Kafka como um acionador para sua função do Lambda.
Para adicionar um acionador do Amazon MSK à sua função do Lambda (console)
-
Abra a página Functions
(Funções) no console do Lambda. -
Escolha o nome da função do Lambda.
-
Em Visão geral da função, escolha Adicionar gatilho.
-
Em Trigger configuration (Configuração do acionador), faça o seguinte:
-
Selecione o tipo de acionador MSK.
-
Em Cluster do MSK, selecione seu cluster.
-
Em Batch size (Tamanho do lote), insira o número máximo de mensagens a serem recebidas em um único lote.
-
Em Batch window (Janela de lote), insira o tempo máximo em segundos usado pelo Lambda para coletar os registros antes de invocar a função.
-
Em Topic name (Nome do tópico), insira um nome para o tópico do Kafka.
-
(Opcional) em Consumer group ID (ID do grupo de consumidores), insira o ID de um grupo de consumidores do Kafka no qual ingressar.
-
(Opcional) Em Posição inicial, escolha Mais recente para começar a realizar a leitura do fluxo a partir do registro mais recente, Horizonte de corte para começar no registro mais antigo disponível ou No carimbo de data e hora para especificar um carimbo de data e hora para começar a realizar a leitura.
-
(Opcional) Em Authentication (Autenticação), escolha a chave secreta para autenticar com os agentes no cluster do MSK.
-
Para criar o gatilho em um estado desativado para teste (recomendado), desmarque Ativar gatilho. Ou, para habilitar o gatilho imediatamente, selecione Ativar gatilho.
-
-
Para criar o acionador, selecione Add (Adicionar).
Adicionando um gatilho do Amazon MSK (AWS CLI)
Use os comandos de exemplo a seguir da AWS CLI para criar e visualizar um acionador do Amazon MSK para sua função do Lambda.
Criar um trigger usando a AWS CLI
exemplo : criar um mapeamento da origem de eventos para o cluster que usa autenticação do IAM
O exemplo a seguir usa o comando create-event-source-mappingmy-kafka-function
em um tópico do Kafka chamado AWSKafkaTopic
. A posição inicial do tópico está definida como LATEST
. Quando o cluster usa a autenticação baseada em perfil do IAM, você não precisa de um objeto SourceAccessConfiguration. Exemplo:
aws lambda create-event-source-mapping \ --event-source-arn
arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2
\ --topicsAWSKafkaTopic
\ --starting-positionLATEST
\ --function-namemy-kafka-function
exemplo : criar um mapeamento da origem de eventos para o cluster que usa autenticação SASL/SCRAM
Se o cluster usar autenticação SASL/SCRAM, você precisa incluir um objeto SourceAccessConfiguration que especifique SASL_SCRAM_512_AUTH
e um ARN secreto do Secrets Manager.
aws lambda create-event-source-mapping \ --event-source-arn
arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2
\ --topicsAWSKafkaTopic
\ --starting-positionLATEST
\ --function-namemy-kafka-function
--source-access-configurations'[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
exemplo : criar um mapeamento da origem de eventos para o cluster que usa autenticação mTLS
Se o cluster usar autenticação mTLS, você precisa incluir um objeto SourceAccessConfiguration que especifique CLIENT_CERTIFICATE_TLS_AUTH
e um ARN secreto do Secrets Manager.
aws lambda create-event-source-mapping \ --event-source-arn
arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2
\ --topicsAWSKafkaTopic
\ --starting-positionLATEST
\ --function-namemy-kafka-function
--source-access-configurations'[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
Para obter mais informações, consulte o .CreateEventSourceMappingDocumentação de referência da API.
Visualizar o status usando a AWS CLI
O exemplo a seguir usa o comando get-event-source-mapping
aws lambda get-event-source-mapping \ --uuid
6d9bce8e-836b-442c-8070-74e77903c815
Parâmetros de configuração do Amazon MSK
Todos os tipos de origem de evento Lambda compartilham o mesmoCreateEventSourceMappingeUpdateEventSourceMappingOperações de API do. No entanto, apenas alguns dos parâmetros se aplicam ao Amazon MSK.
Parâmetro | Obrigatório | Padrão | Observações |
---|---|---|---|
AmazonManagedKafkaEventSourceConfig |
N |
Contém o campo ConsumerGroupId, que assume, por padrão, um valor exclusivo. |
Pode definir apenas em Criar |
BatchSize |
N |
100 |
Máximo: 10.000. |
Habilitado |
N |
Habilitado |
nenhuma |
EventSourceArn |
S |
N/D |
Pode definir apenas em Criar |
FunctionName |
S |
N/D |
nenhuma |
FilterCriteria |
N |
N/D |
|
MaximumBatchingWindowInSeconds |
N |
500 ms |
|
SourceAccessConfigurations |
N |
Sem credenciais do |
Credenciais de autenticação SASL/SCRAM ou CLIENT_CERTIFICATE_TLS_AUTH (MutualTLS) para a fonte do evento |
StartingPosition |
S |
N/D |
AT_TIMESTAMP, TRIM_HORIZON, ou LATEST Pode definir apenas em Criar |
StartingPositionTimestamp |
N |
N/D |
Obrigatório se StartingPosition estiver definido como AT_TIMESTAMP |
Tópicos |
S |
N/D |
Nome do tópico do Kafka Pode definir apenas em Criar |
Criar mapeamentos da origem do evento entre contas
Você pode usar a conectividade privada com várias VPCs para conectar uma função do Lambda a um cluster DO MSK provisionado em outra Conta da AWS. A conectividade com várias VPCs usa AWS PrivateLink, que mantém todo o tráfego na rede da AWS.
nota
Você não pode criar mapeamentos da origem do evento entre contas para clusters do MSK sem servidor.
Para criar um mapeamento da origem de eventos entre contas, primeiro você precisa configurar a conectividade com várias VPCs para o cluster do MSK. Ao criar o mapeamento da origem de eventos, use o ARN da conexão VPC gerenciada, em vez do ARN do cluster, conforme mostrado nos exemplos a seguir. A operação CreateEventSourceMapping também difere dependendo do tipo de autenticação usado pelo cluster do MSK.
exemplo : criar um mapeamento da origem de eventos entre contas para o cluster que usa autenticação do IAM
Quando o cluster usa a autenticação baseada em perfil do IAM, você não precisa de um objeto SourceAccessConfiguration. Exemplo:
aws lambda create-event-source-mapping \ --event-source-arn arn:aws:kafka:
us-east-1:111122223333
:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7
\ --topics AWSKafkaTopic \ --starting-position LATEST \ --function-name my-kafka-function
exemplo : criar um mapeamento da origem de eventos entre contas para o cluster que usa autenticação SASL/SCRAM
Se o cluster usar autenticação SASL/SCRAM, você precisa incluir um objeto SourceAccessConfiguration que especifique SASL_SCRAM_512_AUTH
e um ARN secreto do Secrets Manager.
Há duas maneiras de usar segredos para mapeamentos da origem do evento entre contas do Amazon MSK com autenticação SASL/SCRAM:
-
Crie um segredo na conta da função do Lambda e sincronize-o com o segredo do cluster. Crie uma rotação para manter os dois segredos sincronizados. Essa opção permite controlar o segredo pela conta da função.
-
Use o segredo associado ao cluster do MSK. Esse segredo deve permitir o acesso entre contas à conta da função do Lambda. Para obter mais informações, consulte Permissões para segredos do AWS Secrets Manager para usuários em uma conta diferente.
aws lambda create-event-source-mapping \ --event-source-arn arn:aws:kafka:
us-east-1:111122223333
:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7
\ --topics AWSKafkaTopic \ --starting-position LATEST \ --function-name my-kafka-function \ --source-access-configurations'[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'
exemplo : criar um mapeamento da origem de eventos entre contas para o cluster que usa autenticação mTLS
Se o cluster usar autenticação mTLS, você precisa incluir um objeto SourceAccessConfiguration que especifique CLIENT_CERTIFICATE_TLS_AUTH
e um ARN secreto do Secrets Manager. O segredo pode ser armazenado na conta do cluster ou na conta da função do Lambda.
aws lambda create-event-source-mapping \ --event-source-arn arn:aws:kafka:
us-east-1:111122223333
:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7
\ --topics AWSKafkaTopic \ --starting-position LATEST \ --function-name my-kafka-function \ --source-access-configurations'[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'
Usar um cluster do Amazon MSK como uma origem de eventos
Quando você adiciona seu cluster do Apache Kafka ou do Amazon MSK como um gatilho para a função do Lambda, o cluster é usado como uma origem de eventos.
O Lambda lê os dados de eventos dos tópicos do Kafka que você especifica como Topics
em uma solicitação CreateEventSourceMapping com base na StartingPosition
especificada. Após o processamento bem-sucedido, seu tópico do Kafka é confirmado no cluster do Kafka.
Se você especificar StartingPosition
como LATEST
, o Lambda começará a ler da mensagem mais recente em cada partição pertencente ao tópico. Como pode haver algum atraso após a configuração do acionador antes de o Lambda começar a ler as mensagens, o Lambda não lerá nenhuma mensagem produzida durante a janela.
O Lambda lê as mensagens sequencialmente para cada partição de tópico do Kafka. Uma única carga do Lambda pode conter mensagens de várias partições. Quando mais registros estiverem disponíveis, o Lambda continuará processando-os em lotes, com base no valor de BatchSize
especificado na solicitação CreateEventSourceMapping, até que a função atinja o tópico.
Depois que o Lambda processa cada lote, ele confirma os deslocamentos das mensagens nesse lote. Se sua função retorna um erro para qualquer uma das mensagens em um lote, o Lambda tenta novamente todo o lote de mensagens até que o processamento seja bem-sucedido ou as mensagens expiram. É possível enviar registros que apresentaram falha em todas as tentativas a um destino em caso de falha para processamento posterior.
nota
Embora as funções do Lambda normalmente tenham um limite máximo de tempo de 15 minutos, os mapeamentos da origem dos eventos para o Amazon MSK, o Apache Kafka autogerenciado, o Amazon DocumentDB e o Amazon MQ para ActiveMQ e RabbitMQ são compatíveis somente com funções com limites máximos de tempo limite de 14 minutos. Essa restrição garante que o mapeamento da origem do evento possa solucionar adequadamente os erros de função e repetições.
Posições iniciais de sondagem e fluxo
Esteja ciente de que a sondagem do fluxo durante a criação e as atualizações do mapeamento da origem do evento é, finalmente, consistente.
-
Durante a criação do mapeamento da origem do evento, pode levar alguns minutos para a sondagem de eventos do fluxo iniciar.
-
Durante as atualizações do mapeamento da origem do evento, pode levar alguns minutos para interromper e reiniciar a sondagem de eventos do fluxo.
Esse comportamento significa que, se você especificar LATEST
como posição inicial do fluxo, o mapeamento da origem do evento poderá perder eventos durante a criação ou as atualizações. Para garantir que nenhum evento seja perdido, especifique a posição inicial do fluxo como TRIM_HORIZON
ou AT_TIMESTAMP
.
Métricas do Amazon CloudWatch
O Lambda emite a métrica OffsetLag
enquanto sua função processa registros. O valor dessa métrica é a diferença de deslocamento entre o último registro gravado no tópico da origem de eventos do Kafka e o último registro que o grupo de consumidores da função processou. Você pode usar OffsetLag
para estimar a latência entre o momento em que um registro é adicionado e o momento em que o grupo de consumidores o processa.
Uma tendência crescente em OffsetLag
pode indicar problemas com pesquisadores no grupo de consumidores da função. Para ter mais informações, consulte Visualizar métricas das funções do Lambda.
Auto scaling da origem do evento do Amazon MSK
Quando você cria inicialmente uma fonte de eventos do Amazon MSK, o Lambda aloca um consumidor para processar todas as partições no tópico do Kafka. Cada consumidor conta com vários processadores em execução em paralelo para lidar com um aumento de workloads. O Lambda aumenta ou reduz a escala na vertical automaticamente do número de consumidores com base na workload. Para preservar a ordenação de mensagens em cada partição, o número máximo de consumidores é um consumidor por partição no tópico.
A cada um minuto, o Lambda avalia o atraso de compensação do consumidor de todas as partições do tópico. Se o atraso for muito alto, a partição está recebendo mensagens mais rápido do que o Lambda pode processá-las. Se necessário, o Lambda adiciona ou remove os consumidores do tópico. O processo de escalabilidade de adicionar ou remover consumidores ocorre em até três minutos após a avaliação.
Se a função do Lambda de destino sofrer um controle de utilização, o Lambda reduzirá o número de consumidores. Essa ação reduz a workload na função, reduzindo o número de mensagens que os consumidores podem recuperar e enviar para a função.
Para monitorar o throughput do tópico do Kafka, veja a Métrica de atraso de deslocamento que o Lambda emite enquanto sua função processa registros.
Para verificar quantas invocações de função ocorrem em paralelo, você também pode monitorar ométricas de simultaneidadePara sua função.