Introdução 

A centralização de dados e a gestão eficiente das comunicações com clientes representam desafios críticos em ambientes empresariais modernos. Com a expansão de múltiplos canais de comunicação, tais como CRM, Marketing digital e outras plataformas de engajamento, torna-se imperativo manter a consistência e relevância das interações com os clientes. A eficácia das estratégias de comunicação está diretamente ligada à capacidade de consolidar informações provenientes de diversas fontes e de garantir o cumprimento das regras de supressão de clientes, prevenindo contatos excessivos ou inadequados. 

Objetivo

O objetivo da Tabela de Quarentena de Clientes é criar uma solução centralizada que permita: 

• consolidar dados de diferentes canais de comunicação em uma única base de dados no Databricks. 

• monitorar e controlar a frequência de comunicações com cada cliente de acordo com regras de negócio definidas. 

• implementar um mecanismo de quarentena para evitar comunicações excessivas e desnecessárias com os clientes. 

Embora exista um Data Lake centralizado, ele pode não contemplar todos os canais de comunicação em sua totalidade devido a constantes mudanças e expansões. Diante desse contexto, desenvolvemos a Tabela de Quarentena utilizando tecnologias de Big Data, como PySpark, e, com a infraestrutura de processamento distribuído do Databricks, conseguimos manipular grandes volumes de dados de maneira eficiente, possibilitando a aplicação de regras complexas de negócio. 

Estrutura do projeto 

Para organizar o projeto de forma eficiente, utilizaremos uma estrutura de pastas dentro do Workspace Databricks, similar à desenvolvimento de software, separando o código-fonte das demais interfaces de configurações, apis, entre outros, permitindo uma melhor gestão e manutenção do código. Definimos o nome da aplicação de ‘Quarentena Global’. 

Descrição das pastas 

• analytics: contém consultas SQL utilizadas para análise de dados e geração de insights. 

• docs: documentação, guias técnicos e atualizações. 

• tests: scripts de validação e verificação da integridade dos dados. 

• prd: contém o código de produção organizado em subpastas:
  – src: código-fonte da solução.
  – config: arquivos de configuração e parâmetros (libs, functions, variáveis etc).
  – etl: scripts de ETL responsáveis pela extração, transformação e carregamento dos dados.
  – main: script principal responsável por orquestrar o processo de ETL. 

Origem dos dados 

Para as campanhas de relacionamento com o cliente, são gravados dados de atuação em arquivos Excel (xlsx) no Sharepoint. Para as campanhas do Marketing, existe uma estrutura de pastas que podem ser acessadas via Protocolo de Transferência Seguro (SFTP) do Salesforce Marketing Cloud com arquivos csv.  

Além disso, temos também uma base de suspeitas de fraude que são registradas no Data Lake em um bucket S3 da AWS em parquet, além de um arquivo json com uma lista de restritos. Por fim, campanhas de recomendação de produtos e ofertas que já estão disponíveis no Delta Lake do Unity Catalog. 

As regras de negócio estabelecem períodos de quarentena específicos para cada canal: 

• CRM = 30 dias 

• Marketing = 7 dias 

• Recomendação = 15 dias 

• Restritos = 90 dias 

• Ofertas = 5 dias 

• Fraude = Até a exclusão dos registros na base de origem 

Trabalhamos com múltiplos formatos de arquivo: xlsx, csv, json e parquet. Vamos então proceder com a implementação. 

Desenvolvimento

1. Leitura de bibliotecas e configuração

Para esta abordagem, utilizamos um notebook na pasta config para carregar as bibliotecas necessárias e configurar as variáveis de ambiente.  

2. Funções utilizadas

Definimos funções específicas para conectar e extrair dados de diversas fontes, como Sharepoint e SFTP. Essas funções são essenciais para a etapa de extração de dados e estão armazenadas no notebook na pasta config, juntamente com códigos de parâmetros, bibliotecas e variáveis. 

3. Extração de dados

Na pasta etl, utilizaremos as funções definidas para extrair dados dos diferentes sistemas de origem e gravamos nos diretórios de volumes do Databricks. Em conformidade com as melhores práticas de segurança da informação e LGPD, os dados sensíveis das credenciais das APIs são ocultadas como variáveis de ambiente utilizando Databricks Secrets (dbutils.secrets). 

Após extração dos arquivos, fazemos a leitura de todos eles em seus diferentes formatos, transformando-os em Spark Data frames para serem trabalhados a seguir. Note que utilizamos uma função do Pandas para ler o arquivo Excel antes de transformá-lo. 

4. Transformação dos dados

Realizamos a limpeza, padronização e enriquecimento dos dados extraídos para garantir a consistência e a qualidade deles. Implementamos as regras de quarentena específicas para cada fonte de dados. É importante que todos os Dataframes tenham as mesmas colunas com os mesmos tipos de dados para posteriormente ser feita a união destes com seus respectivos filtros de periodicidade da quarentena.

Todas as fontes têm a coluna id_cliente como chave primária, além de informações de email, telefone, assunto e data de envio. Será necessário adicionar colunas informando a diferença de dias entre a data do processamento e a data do envio assim como, inferir a data da saída prevista, baseado na regra de negócio. Para isso, utilizaremos as funções datediff e date_add da lib do pyspark já importadas no notebook de bibliotecas. Além de renomear algumas colunas que tem títulos diferentes em algumas bases. 

Podemos também automatizar o processo, desde que seja possível estabelecer um padrão na regra de negócio. Suponhamos que as novas ações futuras já estejam no Delta Lake Databricks como tabelas no Unity Catalog e que sejam padronizadas com um período de quarentena de 40 dias.  

Em seguida, padronizamos as colunas e aplicamos o filtro de 40 dias em uma consulta SQL utilizando spark.sql dentro de uma função Python chamada padronizacao_quarenta_dias, que recebe uma tabela como parâmetro: 

Criamos uma variável chamada ‘nova_acao’ que recebe essa função com uma tabela da nova ação como parâmetro, e então, seguimos com a junção das ações. 

Note que, após a transformação dos dados, é possível unir todos os dataframes com dados distintos e filtrar a quantidade de dias de acordo com a regra de negócio e acrescentar uma coluna referente a data da carga dos dados (aqui setado como ‘2024-05-28’). Os clientes que ultrapassam a quantidade de dias estipulados, saem automaticamente da quarentena e ficam disponíveis para contato novamente, assim o histórico pode ser consultado através das datas de cargas passadas. 

5. Carregamento dos dados

Criaremos uma visão temporária associada ao dataframe consolidate_data utilizando a função createOrReplaceTempView (esta será automaticamente removida quando a sessão Spark encerrar). Carregamos os dados da tempview na tabela ‘quarentena_global’ que será criada no Unity Catalog. Com o Spark SQL, podemos executar um código DDL (Data Definition Language) para fazer a criação da tabela caso não exista no catálogo. Em seguida, os dados consolidados da tempview são inseridos na tabela usando as cláusulas SQL CREATE TABLE IF NOT EXIST e INSERT INTO. A tabela será particionada por ‘data_carga’ e ‘canal’, com o ‘id_cliente’ e ‘canal’ como chaves primárias. 

Por fim, a base é armazenada no Delta Lake como tabela gerenciada (Managed Table) em uma database que fica dentro de um catálogo, seguindo os níveis hierárquicos dos objetos de dados do Unity Catalog:  

6. Execução e agendamento

Por fim, temos o notebook principal na pasta main responsável por executar o pipeline e rodar todos os códigos dos notebooks anteriores na ordem já tratada até o momento. Utilizaremos o comando %run informando o caminho das pastas onde estão cada notebook em ordem de execução. Também é possível agendar a execução deste código principal na aba ‘Schedule’, com várias opções de data, horários, alertas e outras configurações adicionais, transformando essa execução em um job, podendo ser monitorado no fluxo de trabalho ‘Workflows’ conforme ilustrado na imagem abaixo: 

Tabela final – integração com o Unity Catalog 

No Unity Catalog, a tabela fica disponível para consultas, seja por meio do SQL Editor ou diretamente em um notebook, podendo salvar essas consultas na pasta analytics e utilizá-las em dashboards ou outras análises. Permite-se gerenciar, auditar e governar os dados em larga escala dentro do ambiente Databricks. Além de que é possível compartilhar e conectar esses dados com outros parceiros e serviços dentro do ecossistema Databricks, facilitando a colaboração e integração com diferentes áreas de negócios e parceiros externos. 

Antes de realizar novas campanhas e ações de engajamento, é possível acessar facilmente a tabela para verificar os clientes que ainda estão em quarentena, garantindo que estes não sejam contatados prematuramente. Isso assegura o respeito às regras de quarentena e melhora a eficiência das campanhas ao evitar interações indesejadas com os clientes. 

Visualização da tabela final: 

Conclusão 

A implementação da Quarentena de Clientes utilizando os poderosos recursos de processamento distribuído e transformação de dados complexos do Spark no Databricks proporciona uma visão centralizada e consolidada das interações dos clientes através de múltiplos canais de comunicação. Este processo assegura a consistência das ações enviadas, respeitando a privacidade e preferências dos clientes, e evitando comunicações excessivas. Com o uso de tecnologias de Big Data, implementamos uma solução escalável e eficiente para a gestão das comunicações com clientes, atendendo as regras do negócio assim como às exigências de conformidade e segurança da informação. 

A utilização de Delta Lake proporciona transações ACID (Atomicidade, Consistência, Isolamento e Durabilidade) e versionamento de dados, assegurando a integridade e consistência dos dados armazenados. O Unity Catalog facilita o gerenciamento, auditoria e governança dos dados em larga escala e permite que a tabela de quarentena seja compartilhada com outros parceiros da Databricks, facilitando a colaboração interdepartamental e com terceiros. Além disso, a tabela pode ser consultada antes de qualquer nova campanha ou ação, garantindo que os clientes em quarentena não sejam contatados, otimizando a eficácia das estratégias de comunicação e respeitando as políticas de quarentena estabelecidas. 

*As opiniões aqui colocadas refletem a minha opinião pessoal e não necessariamente a opinião da Compass UOL.