Usando a ingestão de streaming com o Amazon SageMaker Feature Store para tomar decisões baseadas em ML em tempo quase real

Por Paul Hargis, Arunprasath Shankar, Megan Leoni e Mark Roy

As empresas estão cada vez mais usando o aprendizado de máquina (ML) para tomar decisões em tempo quase real, como colocar um anúncio, atribuir um motorista, recomendar um produto ou até mesmo precificar dinamicamente produtos e serviços. Os modelos de ML fazem previsões com um conjunto de dados de entrada conhecidos como features, e os cientistas de dados gastam facilmente mais de 60% do seu tempo projetando e construindo essas features. Além disso, previsões altamente precisas dependem do acesso em tempo hábil aos valores de features que mudam rapidamente ao longo do tempo, adicionando ainda mais complexidade ao trabalho de construção de uma solução altamente disponível e precisa. Por exemplo, um modelo para um aplicativo de compartilhamento de viagem pode escolher o melhor preço para uma viagem do aeroporto, mas somente se souber o número de pedidos de viagem recebidos nos últimos 10 minutos e o número de passageiros projetados para pousar nos próximos 10 minutos. Um modelo de roteamento em um aplicativo de call center pode escolher o melhor agente disponível para uma chamada recebida, mas só é eficaz se souber os cliques mais recentes da sessão da Web do cliente.

Embora o valor comercial das previsões de ML em tempo quase real seja enorme, a arquitetura necessária para entregá-las de forma confiável, segura e com bom desempenho é complicada. As soluções precisam de atualizações de alto rendimento e recuperação de baixa latência dos valores de features mais recentes em milissegundos, algo que a maioria dos cientistas de dados não está preparada para fornecer. Como resultado, algumas empresas gastaram milhões de dólares inventando sua própria infraestrutura proprietária para gerenciamento de features. Outras empresas limitaram suas aplicações de ML a padrões mais simples, como a pontuação em lote, até que os fornecedores de ML forneçam soluções mais abrangentes para armazenamento de features on-line.

Para enfrentar esses desafios, o Amazon SageMaker Feature Store oferece um repositório central totalmente gerenciado para features de ML, facilitando o armazenamento e recuperação de features com segurança, sem precisar criar e manter sua própria infraestrutura. O Amazon SageMaker Feature Store permite definir grupos de features, usar ingestão de lotes e ingestão de streaming, recuperar os valores de features mais recentes com latência de um dígito de milissegundo para previsões on-line altamente precisas e extrair conjuntos de dados corretos em um ponto específico no tempo para treinamento. Em vez de criar e manter esses recursos de infraestrutura, você obtém um serviço totalmente gerenciado que se expande à medida que seus dados crescem, possibilita o compartilhamento de features entre equipes e permite que seus cientistas de dados se concentrem na criação de modelos de ML voltados para casos de uso que diferenciem o seu negócio. As equipes agora podem oferecer features robustas uma vez e reutilizá-las muitas vezes em uma variedade de modelos que podem ser construídos por diferentes equipes.

Este post apresenta um exemplo completo de como você pode combinar a engenharia de features de streaming com o Amazon SageMaker Feature Store para tomar decisões baseadas em ML em tempo quase real. Mostramos um caso de uso de detecção de fraude de cartão de crédito que atualiza features agregadas de um streaming de transações e usa recuperação de features de baixa latência para ajudar a detectar transações fraudulentas. Teste você mesmo visitando nosso repositório de código.

Caso de uso: fraude de cartão de crédito

Números de cartão de crédito roubados podem ser comprados em massa na dark web de vazamentos anteriores ou hacks de organizações que armazenam esses dados confidenciais. Os fraudadores compram essas listas de cartões e tentam fazer tantas transações quanto possível com os números roubados até que o cartão seja bloqueado. Esses ataques de fraude geralmente acontecem em um curto período de tempo, e isso pode ser facilmente detectado em transações históricas porque a velocidade das transações durante o ataque difere significativamente do padrão de gastos usual do titular do cartão.

A tabela a seguir mostra uma sequência de transações de um cartão de crédito onde o titular do cartão primeiro tem um padrão de gastos genuíno e, em seguida, experimenta um ataque de fraude a partir de 4 de Novembro.

Para este post, treinamos um modelo de ML para identificar esse tipo de comportamento baseado em features que descrevem o padrão de gastos de um cartão individual, como o número de transações ou o valor médio da transação desse cartão em uma determinada janela de tempo. Este modelo protege os titulares de cartões contra fraudes no momento da venda, detectando e bloqueando transações suspeitas antes que o pagamento seja concluído. O modelo faz previsões em um contexto de baixa latência e em tempo real e conta com o recebimento de cálculos de features atualizadas, para que ele possa responder a um ataque de fraude em curso. Em um cenário real, as features relacionadas aos padrões de gastos do titular do cartão apenas fazem parte do conjunto de features do modelo, e podemos incluir informações sobre o comerciante, o titular do cartão, o dispositivo usado para efetuar o pagamento e quaisquer outros dados que possam ser relevantes para detectar fraudes.

Como nosso caso de uso depende do perfil dos padrões de gastos de um cartão individual, é crucial que possamos identificar cartões de crédito em um fluxo de transações. A maioria dos conjuntos de dados de detecção de fraudes disponíveis publicamente não fornece essas informações, portanto, usamos a biblioteca Python Faker para gerar um conjunto de transações que abrangem um período de 5 meses. Este conjunto de dados contém 5,4 milhões de transações distribuídas por 10.000 números de cartões de crédito únicos (e falsos) e é intencionalmente desbalanceado para corresponder à realidade da fraude de cartão de crédito (apenas 0,25% das transações são fraudulentas). Variamos o número de transações por dia por cartão, bem como os valores das transações. Veja nosso repositório de código para obter mais detalhes.

Visão geral da solução

Queremos que nosso modelo de detecção de fraudes classifique as transações com cartão de crédito percebendo um aumento de transações recentes que difere significativamente do padrão de gastos usual do titular do cartão. Parece simples, mas como construímos isso?

O diagrama a seguir mostra a arquitetura geral da solução. Entendemos que esse mesmo padrão funcionará bem para uma variedade de casos de uso de agregação de streaming. Basicamente, o padrão envolve os seguintes cinco componentes. Nós entraremos em mais detalhes sobre estes componentes nas seções subsequentes:

1. Repositório de features — Usamos o Amazon SageMaker Feature Store para fornecer um repositório de features com escritas de alto rendimento e leituras seguras de baixa latência, usando valores de features que estão organizados em vários grupos de features.
2. Ingestão em lote — A ingestão em lote recebe transações de cartão de crédito históricas rotuladas e cria as features agregadas e proporções necessárias para treinar o modelo de detecção de fraudes. Usamos um trabalho de processamento do Amazon SageMaker e o contêiner integrado de Spark para calcular contagens semanais agregadas e valores médios por transação. Em seguida, ingerimos essas features no repositório de features para uso em inferências on-line.
3. Treinamento e implantação de modelos — Esse é um aspecto direto da solução. Usamos o Amazon SageMaker para treinar um modelo usando o algoritmo integrado XGBoost nas features agregadas criadas a partir de transações históricas. O modelo é implantado em um endpoint SageMaker, no qual ele lida com solicitações de detecção de fraudes em transações em tempo real.
4. Ingestão de streaming — Uma aplicação do Amazon Kinesis Data Analytics calcula as features agregadas de um streaming de transações e uma função do AWS Lambda atualiza o repositório de features on-line.
5. Previsões de streaming — Por fim, fazemos previsões de fraudes em um streaming de transações, usando o AWS Lambda para extrair features agregadas do repositório de features on-line. Usamos os dados mais recentes das features para calcular as taxas de transação e, em seguida, chamar o endpoint de detecção de fraudes.

Repositório de features

Os modelos de ML dependem de features bem projetadas provenientes de uma variedade de fontes de dados, com transformações tão simples quanto cálculos ou tão complicadas quanto um pipeline de várias etapas, que leva horas de tempo de computação e codificação complexa. O Amazon SageMaker Feature Store permite a reutilização dessas features entre equipes e modelos, o que melhora a produtividade dos cientistas de dados, acelera o time-to-market e garante a consistência das entradas do modelo.

Cada feature dentro do SageMaker Feature Store é organizada em um agrupamento lógico chamado grupo de features. Você decide quais grupos de features você precisa para seus modelos. Cada um pode ter dezenas, centenas ou até milhares de features. Os grupos de features são gerenciados e dimensionados de forma independente, mas todos estão disponíveis para pesquisa e descoberta em equipes de cientistas de dados responsáveis por muitos modelos de ML e casos de uso independentes.

Os modelos de ML geralmente exigem features de vários grupos de features. Um aspecto chave de um grupo de features é a frequência com que os valores das features precisam ser atualizados ou materializados para treinamento ou inferência. Você pode atualizar algumas features por hora, por dia ou por semana, e um subconjunto de features deve ser transmitido para o repositório de features em tempo quase real. Transmitir todas as atualizações de features adicionaria complexidade desnecessária e poderia até diminuir a qualidade das distribuições de dados, por não dar a você a chance de remover outliers.

Em nosso caso de uso, criamos um grupo de features chamado cc-agg-batch-fg para features agregadas de cartão de crédito atualizados em lote, e um chamado cc-agg-fg para features de streaming. O grupo de features de lote é atualizado todas as noites e fornece features agregadas olhando para trás em uma janela de tempo de uma semana. Recalcular agregações de uma semana em um streaming de transações não oferece sinais significativos e seria um desperdício de recursos.

Por outro lado, nosso grupo de features cc-agg-fg deve ser atualizado por streaming, porque ele oferece as últimas contagens de transações e os valores médios por transação olhando para trás em uma janela de tempo de 10 minutos. Sem a agregação de streaming, não conseguiríamos identificar o padrão típico de ataque de fraude de uma rápida sequência de compras.

Ao isolar features que são recalculadas todas as noites, podemos melhorar a taxa de transferência de ingestão para nossas features de streaming. A separação nos permite otimizar a ingestão de cada grupo de forma independente. Ao projetar para seus casos de uso, tenha em mente que modelos que exigem features de um grande número de grupos de features podem querer fazer várias recuperações do repositório de features em paralelo para evitar adicionar latência excessiva a um fluxo de trabalho de previsão em tempo real.

Os grupos de features para o nosso caso de uso são vistos no diagrama a seguir.

Cada grupo de features deve ter uma feature usada como identificador de registro (para este post, o número do cartão de crédito). O identificador de registro atua como uma chave primária para o grupo de features, permitindo pesquisas rápidas, bem como junções entre grupos de features. Uma feature de hora de evento também é necessária, o que permite que o repositório de features rastreie o histórico de valores de features ao longo do tempo. Isso se torna importante quando se olha para trás para o estado das features em um ponto específico no tempo.

Em cada grupo de features, rastreamos o número de transações por cada cartão de crédito único e seu valor médio de transação. A única diferença entre nossos dois grupos é a janela de tempo usada para agregação. Usamos uma janela de 10 minutos para agregação de streaming e uma janela de 1 semana para agregação em lote.

Com o Amazon SageMaker Feature Store, você tem a flexibilidade de criar grupos de features que estejam somente off-line, somente on-line ou ambos, on-line e off-line. O repositório online fornece escritas de alto rendimento e recuperações de baixa latência de valores de features, ideais para inferência on-line. O repositório off-line é fornecido usando o Amazon S3, oferecendo às empresas um repositório altamente escalável, com um histórico completo de valores de features, particionado por grupo de features. O repositório off-line é ideal para casos de uso de treinamento e pontuação em lote.

Quando você habilita um grupo de features para fornecer repositórios on-line e off-line, o SageMaker sincroniza automaticamente os valores das features em um repositório off-line, anexando continuamente os valores mais recentes para fornecer um histórico completo de valores ao longo do tempo. Outro benefício dos grupos de features que estão on-line e off-line é ajudar a evitar o problema de assimetria entre treinamento e inferência. O SageMaker permite alimentar ambos, treinamento e inferência, com os mesmos valores de features transformados, garantindo consistência para gerar previsões mais precisas. O foco em nosso post é demonstrar o streaming de features on-line, por isso implementamos grupos de features somente on-line.

Ingestão em lote

Para materializar nossas features de lote, criamos um pipeline de features que é executado todas as noites como um trabalho de processamento do Amazon SageMaker. O trabalho tem duas responsabilidades: produzir o conjunto de dados para treinar nosso modelo e preencher o grupo de features em lote com os valores mais atualizados para features agregadas de 1 semana, conforme mostrado no diagrama a seguir:

Cada transação histórica usada no conjunto de treinamento é enriquecida com features agregadas para o cartão de crédito específico envolvido na transação. Olhamos para trás sobre duas janelas de tempo deslizantes separadas: 1 semana atrás e os 10 minutos anteriores. As features reais usadas para treinar o modelo incluem as seguintes proporções desses valores agregados:

• amt_ratio1 = avg_amt_last_10m/avg_amt_last_1w
• amt_ratio2 = transaction_amount/avg_amt_last_1w
• count_ratio = num_trans_last_10m/num_trans_last_1w

Por exemplo, a terceira proporção é a contagem de transações dos 10 minutos anteriores dividida pela contagem de transações da última semana. Nosso modelo de ML pode aprender padrões de atividade normal versus atividade fraudulenta a partir desses índices, em vez de depender de contagens brutas e valores de transação. Os padrões de gastos em cartões diferentes variam muito, portanto, as proporções normalizadas fornecem um sinal melhor para o modelo do que os próprios valores agregados.

Você pode estar se perguntando por que nosso trabalho em lote está computando features dos últimos 10 minutos. Isso não é relevante apenas para inferência online? Precisamos da janela de 10 minutos sobre transações históricas para criar um conjunto de dados de treinamento preciso. Isso é fundamental para garantir a consistência com a janela de streaming de 10 minutos que será usada em tempo real para suportar inferência on-line.

O conjunto de dados de treinamento resultante do trabalho de processamento pode ser salvo diretamente como um CSV para treinamento de modelo, ou pode ser ingerido em massa em um grupo de features off-line que pode ser usado para outros modelos e por outras equipes de ciência de dados para lidar com uma grande variedade de outros casos de uso. Por exemplo, podemos criar e preencher um grupo de features chamado cc-transactions-fg. Nosso trabalho de treinamento pode então obter um conjunto de dados de treinamento específico com base nas necessidades de nosso modelo, selecionando intervalos de datas específicos e um subconjunto de features de interesse. Essa abordagem permite que várias equipes reutilizem grupos de features e mantenham menos pipelines de features, gerando economias significativas de custos e melhorias de produtividade ao longo do tempo. Este notebook de exemplo demonstra o padrão de como usar o SageMaker Feature Store como um repositório central do qual os cientistas de dados podem extrair conjuntos de dados de treinamento.

Além de criar um conjunto de dados de treinamento, usamos a API PutRecord para colocar as features agregadas de 1 semana no repositório de features on-line todas as noites. O código a seguir demonstra a colocação de um registro em um grupo de features on-line dado valores de features específicos, incluindo um identificador de registro e uma hora de evento:

record = [{'FeatureName': 'cc_num',

              'ValueAsString': str(cc_num)},

             {'FeatureName':'avg_amt_last_1w',

              'ValueAsString': str(avg_amt_last_1w)},

             {'FeatureName':'num_trans_last_1w',

              'ValueAsString': str(num_trans_last_1w)}]

event_time_feature = {

                 'FeatureName': 'trans_time',

                 'ValueAsString': str(int(round(time.time())))}

record.append(event_time_feature)

response = feature_store_client.put_record(

    FeatureGroupName=’cc-agg-batch-fg’, Record=record)

Os engenheiros de ML geralmente constroem uma versão separada do código de engenharia de features para features on-line com base no código original escrito por cientistas de dados para treinamento de modelos. Isso pode oferecer o desempenho desejado, mas é uma etapa de desenvolvimento extra e apresenta mais chances de assimetria entre treinamento e inferência. No nosso caso de uso, mostramos como usar o SQL para agregações pode permitir que um cientista de dados forneça o mesmo código para lote e streaming.

Ingestão de streaming

O Amazon SageMaker Feature Store oferece recuperação de um dígito de milissegundo de features pré-calculadas e também pode desempenhar um papel eficaz nas soluções que exigem ingestão de streaming. Nosso caso de uso demonstra ambos. A agregação semanal é tratada como um grupo de features pré-calculado, materializado todas as noites como mostrado anteriormente. Agora vamos mergulhar em como calculamos features agregadas em uma janela de 10 minutos e os ingerimos no repositório de features para posterior inferência on-line.

Você pode executar a ingestão de streaming em um tópico do Apache Kafka ou em um Amazon Kinesis Data Stream, aplicando transformação e agregação de features e enviando o resultado para o repositório de features. Para equipes confortáveis com Java, o Apache Flink é uma estrutura popular para agregação de streaming. No entanto, para cientistas de dados com habilidades Java limitadas, o SQL é uma opção muito mais acessível.

Em nosso caso de uso, escutamos um fluxo de dados de transações com cartão de crédito do Kinesis e usamos uma aplicação SQL simples do Kinesis Data Analytics para criar features agregadas. Uma função do AWS Lambda ingere essas features no repositório de features para uso subsequente no momento da inferência. Estabelecer a aplicação SQL é simples. Você escolhe um fluxo de origem, define uma consulta SQL e identifica um destino (para nosso caso de uso, uma função do Lambda).

Para produzir contagens e valores médios agregados olhando para trás em uma janela de 10 minutos, usamos a seguinte consulta SQL no fluxo de entrada:

SELECT STREAM "cc_num",

               COUNT(*) OVER LAST_10_MINUTES,

               AVG("amount") OVER LAST_10_MINUTES

FROM transactions WINDOW LAST_10_MINUTES AS (PARTITION BY "cc_num" RANGE INTERVAL '10' MINUTE PRECEDING)

Neste exemplo, observe que a linha final tem uma contagem de quatro transações nos últimos 10 minutos a partir do cartão de crédito que termina com 1248, e um valor médio de transação correspondente de US$200,00. A consulta SQL é consistente com a usada para a criação de nosso conjunto de dados de treinamento, ajudando a evitar a assimetria entre treinamento e inferência.

À medida que as transações são transmitidas para a aplicação SQL, a aplicação envia os resultados agregados para nossa função do Lambda, conforme mostrado no diagrama a seguir. A função Lambda usa essas features e preenche o grupo de features cc-agg-fg.

A atualização dos valores das features no repositório de features pela Lambda é feita usando uma chamada simples para a API PutRecord. O trecho de código seguinte é a parte central do código Python para armazenar as features agregadas:

record = [{'FeatureName': 'cc_num', 

           'ValueAsString': str(cc_num)},

          {'FeatureName':'avg_amt_last_10m', 

           'ValueAsString': str(avg_amt_last_10m)},

          {'FeatureName':'num_trans_last_10m', 

           'ValueAsString': str(num_trans_last_10m)},

          {'FeatureName': 'evt_time', 

           'ValueAsString': str(int(round(time.time())))}]

featurestore_runtime.put_record(FeatureGroupName='cc-agg-fg', 

                                Record=record)

Preparamos o registro como uma lista de pares de valores nomeados, incluindo a hora atual como a hora do evento. A API do SageMaker Feature Store garante que esse novo registro siga o esquema que identificamos quando criamos o grupo de features. Caso já exista um registro para essa chave primária, ele agora será substituído no repositório on-line.

Previsões de streaming

Agora que temos a ingestão de streaming mantendo o repositório de features atualizado com os valores de features mais recentes, vamos ver como fazemos previsões de fraudes.

Criamos uma segunda função do Lambda que usa um fluxo de dados do Kinesis como um gatilho. Para cada novo evento de transação, recuperamos features de lote e streaming do SageMaker Feature Store, calculamos as proporções e invocamos o endpoint do modelo SageMaker para fazer a previsão conforme mostrado no diagrama a seguir.

Usamos o código a seguir para recuperar valores de features sob demanda do repositório de features antes de chamar o endpoint do modelo SageMaker:

featurestore_runtime = 

        boto3.client(service_name='sagemaker-featurestore-runtime')

response = featurestore_runtime.get_record(

               FeatureGroupName=feature_group_name,

        RecordIdentifierValueAsString=record_identifier_value)

Finalmente, com o vetor de features de entrada do modelo montado, chamamos o endpoint do modelo para prever se uma transação específica de cartão de crédito é fraudulenta:

sagemaker_runtime = 

    boto3.client(service_name='runtime.sagemaker')

request_body = ','.join(features)

response = sagemaker_runtime.invoke_endpoint(

    EndpointName=ENDPOINT_NAME,

    ContentType='text/csv',

    Body=request_body)

probability = json.loads(response['Body'].read().decode('utf-8'))

No exemplo acima, o modelo retornou com uma probabilidade de 98% de que a transação específica fosse fraudulenta e foi capaz de utilizar features de entrada agregadas em tempo quase real, com base nos 10 minutos mais recentes de transações nesse cartão de crédito.

Vê-lo funcionar de ponta a ponta

Para demonstrar o fluxo de trabalho completo de nossa solução, simplesmente enviamos transações com cartão de crédito para o fluxo de dados do Kinesis. Nossa agregação automatizada de features de streaming assume a partir daí, mantendo uma visão em tempo quase real das contagens de transações e valores na SageMaker Feature Store, com uma janela deslizante dos últimos 10 minutos. Essas features são combinadas com as features agregadas de 1 semana que já foram ingeridas na Feature Store em lote, permitindo-nos fazer previsões de fraude em cada transação.

Enviamos uma única transação a partir de três cartões de crédito diferentes. Em seguida, simulamos um ataque de fraude em um quarto cartão de crédito enviando muitas transações consecutivas em segundos. A saída da nossa função Lambda é mostrada abaixo. Como esperado, as três primeiras transações pontuais são previstas como NOT FRAUD. Das 10 transações fraudulentas, a primeira é prevista como NOT FRAUD, e o resto está corretamente identificado como FRAUD. Observe como as features agregadas são mantidas atualizadas, ajudando a gerar previsões mais precisas.

Conclusão

Mostramos como o Amazon SageMaker Feature Store pode desempenhar um papel fundamental na arquitetura da solução para fluxos de trabalho operacionais críticos que precisam de agregação de streaming e inferência de baixa latência. Com um repositório de features pronto para empresas, você pode usar a ingestão de lotes e a ingestão de streaming para alimentar grupos de features e acessar valores de features sob demanda para executar previsões on-line para gerar valor comercial significativo para o seu negócio. As features de ML agora podem ser compartilhadas em escala entre muitas equipes de cientistas de dados e milhares de modelos de ML, melhorando a consistência dos dados, a precisão do modelo e a produtividade dos cientistas de dados. O Amazon SageMaker Feature Store está disponível agora e você pode experimentar esse exemplo completo. Compartilhe conosco o que você pensa.

Sobre os Autores

Paul Hargis é especialista em AI/ML, arquiteto de soluções na Amazon Web Services (AWS). Antes dessa função, ele era arquiteto líder da Amazon Exports and Expansions ajudando a amazon.com a melhorar a experiência para compradores internacionais. Paul gosta de ajudar os clientes a expandir suas iniciativas de aprendizado de máquina para resolver problemas do mundo real. Ele é casado e tem uma filha que corre em equipes de Cross Country e Track no ensino médio.

Megan Leoni é arquiteta de soluções especialista em AI/ML na AWS, ajudando clientes de toda a Europa, Oriente Médio e África a projetar e implementar soluções de ML. Antes de ingressar na AWS, Megan trabalhou como cientista de dados criando e implantando modelos de detecção de fraudes em tempo real.

Mark Roy é arquiteto principal de aprendizado de máquina na AWS, ajudando os clientes da AWS a projetar e criar soluções de AI/ML. O trabalho de Mark abrange uma ampla variedade de casos de uso de ML, com um interesse principal em visão computacional, aprendizado profundo e dimensionamento de ML em toda a empresa. Ele ajudou empresas em muitos setores, incluindo Seguros, Serviços Financeiros, Mídia e Entretenimento, Saúde, Serviços Públicos e Manufatura. Mark possui 6 certificações da AWS, incluindo a Certificação Specialty ML. Antes de ingressar na AWS, Mark foi arquiteto, desenvolvedor e líder em tecnologia por mais de 25 anos, incluindo 19 anos em serviços financeiros.

Arunprasath Shankar é um arquiteto de soluções especializado em inteligência artificial e aprendizado de máquina (AI/ML) na AWS, ajudando clientes globais a escalar suas soluções de IA de forma eficaz e eficiente na nuvem. Em seu tempo livre, Arun gosta de assistir filmes de ficção científica e ouvir música clássica.

Revisor

Yan Marim é arquiteto de soluções na AWS e tem interesse principalmente em AI/ML. Atualmente apoia clientes na sua jornada de transformação digital na nuvem.