Como aproveitar uma arquitetura orientada a eventos com MongoDB e Databricks

Francesco Baldissera9 min read • Published Jul 13, 2023 • Updated Jul 13, 2023

Spark Framework de agregação Python JavaScript MongoDB

Avalie esse Tutorial

Siga este tutorial para obter uma visão detalhada de como aproveitar o MongoDB Atlas App Services, além dos recursos de construção e implantação de modelos do Databricks para alimentar estratégias baseadas em dados com dados de eventos em tempo real. Vamos começar!

Noções básicas

Vamos usar um cluster MongoDB Atlas M10 como serviço de backend para a solução. Se você ainda não estiver familiarizado com o MongoDB Atlas, poderá acompanhar o cursoIntrodução ao MongoDB para começar com os fundamentos básicos da configuração e do gerenciamento de clusters.

Coleta de dados

A solução é baseada em dados que imitam uma collection de ingestão dearquitetura orientada por eventos da vitrine de um site de e-commerce. Vamos usar um conjunto de dados sintético para representar o que receberemos em nosso banco de dados na cloud a partir de um fluxo de eventos Kafka. A fonte de dados pode ser encontrada em Kaggle.

Os dados estão em formato tabular. Quando convertido em um objeto adequado para o MongoDB, ficará assim:

Trecho de código

A arquitetura orientada a eventos é muito simples. Ele é composto por apenas quatro eventos diferentes que um usuário pode realizar no site de comércio eletrônico:

event_type	Descrição
"visualização"	Um cliente visualiza um produto na página de detalhes do produto.
"carrinho"	Um cliente adiciona um produto ao carro.
"remove_from_cart"	Um cliente remove um produto do carrinho.
"Compra"	Um cliente conclui uma transação de um produto específico.

Os dados no conjunto de dados Kaggle são feitos de 4.6 milhões de documentos, que armazenaremos em um banco de dados chamado "ecom_events" e na coleção "cosmetics". Esta coleção representa todos os eventos que acontecem em uma loja de múltiplas categorias durante o mês de novembro 2019.

Escolhemos esta data especificamente porque ela conterá o comportamento correspondente às promoções da Black Friday, portanto, certamente mostrará as mudanças de preço e, portanto, será mais interessante avaliar a elasticidade de preço dos produtos durante esse período.

Dados agregados no MongoDB

Usando o poderoso MongoDB Atlas Aggregation Pipeline, você pode moldar seus dados da maneira que precisar. Vamos moldar os eventos em uma visão agregada que nos fornecerá um "purchase log " para que possamos ter preços históricos e quantidades totais vendidas por produto. Dessa forma, podemos alimentar um modelo de regressão linear para obter o melhor ajuste possível de uma linha que representa a relação entre preço e unidades vendidas.

Abaixo, você verá as diferentes etapas do aggregation pipeline:

Correspondência: Estamos interessados apenas em comprar eventos, então executamos uma etapa de correspondência para a chave event_type com o valor 'compra'.
Trecho de código{ '$match': { 'event_type': 'purchase' } }
Grupo: Estamos interessados em saber quantas vezes um determinado produto foi comprado em um dia e a que preço. Portanto, agrupamos por todas as chaves relevantes, enquanto também realizamos uma transformação de tipo de dados para o "event_time ", e calculamos um novo campo, "total_sales ", para obter vendas totais diárias a um preço específico.
Trecho de código{ '$group': { '_id': { 'event_time': { '$dateToString': { 'format': '%Y-%m-%d', 'date': '$event_time' } }, 'product_id': '$product_id', 'price': '$price', 'brand': '$brand', 'category_code': '$category_code' }, 'total_sales': { '$sum': 1 } } }
Projeto: Em seguida, executamos uma etapa do projeto para eliminar o aninhamento de objetos resultante após a fase de grupos. (Confira o MongoDB Compass Aggregation Pipeline Builder, pois você poderá ver o resultado de cada um dos estágios adicionados ao seufunil!)
Trecho de código{ '$project': { 'total_sales': 1, 'event_time': '$_id.event_time', 'product_id': '$_id.product_id', 'price': '$_id.price', 'brand': '$_id.brand', 'category_code': '$_id.category_code', '_id': 0 } }
Group, Sort e Project: Precisamos de apenas um objeto que tenha o histórico de vendas de um produto durante o tempo, uma espécie de registro de dados de série temporal que computa agregados ao longo do tempo. Observe como também executaremos uma transformação de dados no estágio "$project" para obter a "receita" gerada por esse produto naquele dia específico. Para conseguir isso, precisamos agrupar, classificar e projetar dessa forma:
Trecho de código{ '$group': { '_id': '$product_id', 'sales_history': { '$push': '$$ROOT' } } }, { '$sort': { 'sales_history': -1 } }, { '$project': { 'product_id': '$_id', 'event_time': '$sales_history.event_time', 'price': '$sales_history.price', 'brand': '$sales_history.brand', 'category_code': '$sales_history.category_code', 'total_sales': '$sales_history.total_sales', 'revenue': { '$map': { 'input': '$sales_history', 'as': 'item', 'in': { '$multiply': [ '$$item.price', '$$item.total_sales' ] } } } } }
Saída: A última etapa do pipeline é enviar nossos objetos com formato adequado para uma nova coleção chamada "purchase_log". Essa coleção servirá como base para alimentar nosso modelo, e o pipeline de agregação será a linha de base de um trigger mais adiante para automatizar a geração desse registro sempre que houver uma compra, mas, nesse caso, usaremos um estágio $merge.
Trecho de código{ '$out': 'purchase_log' }

Com esse pipeline de agregação, estamos efetivamente transformando nossos dados no registro de compras necessário para entender o histórico de vendas pelo preço de cada produto e começar a construir nosso painel para leads de categoria para entender as vendas de produtos e usar esses dados para calcular a elasticidade-preço da demanda de cada um deles.

Camada de inteligência: Criando seu modelo e implementando-o em um endpoint da Databricks

O objetivo desta etapa é calcular a flexibilidade de preço da demanda de cada produto em tempo real. Usando o Databricks, você pode iniciar facilmente um cluster e anexar seu Notebookde construção de modelo a ele.

Em seu Notebook, você pode importar dados MongoDB usando o MongoDB Connector para Spark, e você também pode aproveitar a biblioteca demódulos Python personalizada MlFlow para escrever seus scripts Python, como este abaixo:

Trecho de código

Além disso, você pode registrar os experimentos e depois registrá-los como modelos para que possam ser servidos como endpoints na interface do usuário:

Registro do modelo como experimento diretamente do Notebook:

Trecho de código

Na página do modelo, você pode clicar em “deploy model” e obterá um URL de endpoint.

Depois de testar seu modelo de endpoint, é hora de orquestrar seu aplicativo para obter análises em tempo real.

Orquestrando seu aplicativo

Para este desafio, usaremos Triggers e Funções do MongoDB para garantir que agregaremos os dados apenas do último produto comprado sempre que houver um evento de compra e recalcularemos sua elasticidade de preço passando seu log de compra em uma pós-chamada HTTP para o Ponto de extremidade do databricks.

Agregação de dados após cada compra

Primeiro, você precisará configurar um fluxo de eventos que possa capturar as alterações no comportamento do consumidor e as alterações de preço em tempo real, para que ele agregue e atualize seus dados purchase_log.

Ao aproveitar oAtlas App Services, você pode criar aplicativos orientados a eventos e integrar serviços na cloud. Portanto, para este caso de uso, gostaríamos de configurar um trigger que "listen" para qualquer novo evento "purchase" na collection de cosméticos, como você pode ver nas capturas de tela abaixo. Para começar a usar os Serviços de Aplicativos, você pode conferir a documentação.

Depois de clicar em “Add Trigger,”, você pode configurá-lo para ser executado somente quando houver uma nova inserção na coleção:

Role a página para baixo, você também pode configurar a função que será acionada:

Essas funções podem ser definidas (e testadas) no editor de funções. A função que estamos usando simplesmente recupera dados da coleção de cosméticos, executa algum processamento de dados nas informações e salva o resultado em uma nova coleção.

Trecho de código

exports = async function() {
  const collection = context.services.get("mongodb-atlas").db('ecom_events').collection('cosmetics');
  
  // Retrieving the last purchase event document
  let lastItemArr = [];
  
  try {
    lastItemArr = await collection.find({event_type: 'purchase'}, { product_id: 1 }).sort({ _id: -1 }).limit(1).toArray();
  } 
  catch (error) {
    console.error('An error occurred during find execution:', error);
  }
  console.log(JSON.stringify(lastItemArr));
  
  // Defining the product_id of the last purchase event document 
  var lastProductId = lastItemArr.length > 0 ? lastItemArr[0].product_id : null; 
  console.log(JSON.stringify(lastProductId));
  console.log(typeof lastProductId);
  if (!lastProductId) {
    return null; 
  }
  
  // Filtering the collection to get only the documents that match the same product_id as the last purchase event
  let lastColl = [];
    lastColl = await collection.find({"product_id": lastProductId}).toArray();
  console.log(JSON.stringify(lastColl));
  
  
  // Defining the aggregation pipeline for modeling a purchase log triggered by the purchase events.
  const agg = [
    {
      '$match': {
        'event_type': 'purchase',
        'product_id': lastProductId
      }
    }, {
      '$group': {
        '_id': {
          'event_time': '$event_time', 
          'product_id': '$product_id', 
          'price': '$price', 
          'brand': '$brand', 
          'category_code': '$category_code'
        }, 
        'total_sales': {
          '$sum': 1
        }
      }
    }, {
      '$project': {
        'total_sales': 1, 
        'event_time': '$_id.event_time', 
        'product_id': '$_id.product_id', 
        'price': '$_id.price', 
        'brand': '$_id.brand', 
        'category_code': '$_id.category_code', 
        '_id': 0
      }
    }, {
      '$group': {
        '_id': '$product_id', 
        'sales_history': {
          '$push': '$$ROOT'
        }
      }
    }, {
      '$sort': {
        'sales_history': -1
      }
    }, {
    '$project': {
      'product_id': '$_id', 
      'event_time': '$sales_history.event_time', 
      'price': '$sales_history.price', 
      'brand': '$sales_history.brand', 
      'category_code': '$sales_history.category_code', 
      'total_sales': '$sales_history.total_sales', 
      'revenue': {
        '$map': {
          'input': '$sales_history', 
          'as': 'item', 
          'in': {
            '$multiply': [
              '$$item.price', '$$item.total_sales'
            ]
          }
        }
      }
    }
  }
, {
    '$merge': {
     'into': 'purchase_log',
     'on': '_id',
     'whenMatched': 'merge',
     'whenNotMatched': 'insert'
      }
    }
  ];
  
  // Running the aggregation
    const purchaseLog = await collection.aggregate(agg);
    const log = await purchaseLog.toArray();
    return log;
};

A função acima destina-se a moldar os dados do último itemproduct_id adquirido no preference_log histórico necessário para calcular a flexibilidade preço. Como você pode ver no código abaixo, o resultado cria um documento com preço histórico e dados totais de compra:

Trecho de código

Observe como implementamos o estágio$merge para não substituir a coleção anterior e apenas inserir os dados correspondentes ao último item comprado.

Cálculo da flexibilidade de preço

A próxima etapa é processar o fluxo de eventos e calcular a elasticidade de preço da demanda para cada produto. Para isso, você pode configurar um trigger para que, toda vez que houver uma inserção ou substituição na collection "purchase_log ", façamos uma solicitação pós-HTTP para recuperar a flexibilidade preço.

O gatilho executará uma função como a abaixo:

Trecho de código

exports = async function(changeEvent) {
  
  // Defining a variable for the full document of the last purchase log in the collection
  const { fullDocument } = changeEvent; 
  console.log("Received doc: " + fullDocument.product_id);
  
  // Defining the collection to get
  const collection = context.services.get("mongodb-atlas").db("ecom_events").collection("purchase_log");
  console.log("It passed test 1");
  
  // Fail proofing
  if (!fullDocument) {
    throw new Error('Error: could not get fullDocument from context');
  }
  console.log("It passed test 2");
  
  if (!collection) {
    throw new Error('Error: could not get collection from context');
  }

console.log("It passed test 3");
  
  //Defining the connection variables
  const ENDPOINT_URL = "YOUR_ENDPOINT_URL";
  const AUTH_TOKEN = "BASIC_TOKEN";
  
  // Defining data to pass it into Databricks endpoint
  const data = {"inputs": [fullDocument]};
  
  console.log("It passed test 4");
  
  // Fetching data to the endpoint using http.post to get price elasticity of demand
  try {
    const res = await context.http.post({
      "url": ENDPOINT_URL,
      "body": JSON.stringify(data),
      "encodeBodyAsJSON": false,
      "headers": {
        "Authorization": [AUTH_TOKEN],
        "Content-Type": ["application/json"]
      }
      
    });
    
    console.log("It passed test 5");
   
    if (res.statusCode !== 200) {
      throw new Error(`Failed to fetch data. Status code: ${res.statusCode}`);
    }
    
    console.log("It passed test 6");
    
    // Logging response test
    const responseText = await res.body.text();
    console.log("Response body:", responseText);

// Parsing response from endpoint
    const responseBody = JSON.parse(responseText);
    const price_elasticity = responseBody.predictions;
  
    console.log("It passed test 7 with price elasticity: " + price_elasticity);
  
    //Updating price elasticity of demand for specific document on the purchase log collection
  
      await collection.updateOne({"product_id": fullDocument.product_id}, {$push:{"price_elasticity": price_elasticity}} );
      console.log("It updated the product_id " + fullDocument.product_id + "successfully, adding price elasticity " + price_elasticity );  
    } 
      
    catch (err) {
    console.error(err);
    throw err;
  }
};

Visualize os dados com o MongoDB Charts

Por fim, você precisará visualizar os dados para facilitar a compreensão das partes envolvidas na flexibilidade de preço da demanda de cada produto. Você pode usar uma ferramenta de visualização como o MongoDB Charts para criar dashboards e relatórios que mostram a flexibilidade de preço da demanda ao longo do tempo e como ela é impactada por alterações no preço, nas ofertas de produtos e no comportamento do consumidor.

Evoluindo seus aplicativos

A nova variável “price_elasticity” pode ser facilmente passada para a collection que alimenta seu PIMS, permitindo que os desenvolvedores criem outro conjunto de regras com base nesses valores para automatizar uma ferramenta de precificação dinâmica completa.

Ele também pode ser incorporado em seus aplicativos. Digamos que um sistema CMS de comércio eletrônico usado por sua categoria leve ao ajuste manual dos preços de diferentes produtos. Ou, nesse caso, criar regras diferentes com base na elasticidade de preço da demanda para automatizar a definição de preços.

Os mesmos dados podem ser usados como um recurso para prever o total de vendas e criar um ponto de preço recomendado para a receita líquida.

Concluindo, esta estrutura pode ser usada para criar qualquer tipo de caso de uso de análise em tempo real que você possa imaginar em combinação com qualquer um dos diversos casos de uso que você encontrará onde o aprendizado de máquina poderia ser usado como uma fonte de decisão inteligente e automatizada. -processos de fabricação.

Encontre todo o código usado no repositório do GitHub e passe pelo Foro da Comunidade para mais dúvidas, comentários ou feedback!

Avalie esse Tutorial

Relacionado

Podcast

Sugestões de esquema com Julia Openhein - Episódio 59 do podcast

Aug 09, 2024 | 13 min

Tutorial

Build a RESTful API With HapiJS and MongoDB

Sep 11, 2024 | 15 min read

Artigo

Tudo que você sabe sobre MongoDB está errado!

Sep 23, 2022 | 11 min read

Tutorial

Como construir um aplicativo web Go com Gi, MongoDB e AI

Aug 30, 2024 | 11 min read

Sumário

Noções básicas
Coleta de dados
Dados agregados no MongoDB
Camada de inteligência: Criando seu modelo e implementando-o em um endpoint da Databricks
Orquestrando seu aplicativo
Visualize os dados com o MongoDB Charts
Evoluindo seus aplicativos