🖕🏽 💿 👨🏽‍🎤 Delta: plataforma de enriquecimento e sincronização de dados 😇 🕴🏻 🙍🏻

Antecipando o lançamento de um novo fluxo no curso de Data Engineer, preparamos uma tradução de material interessante.

Visão geral

Falaremos sobre um padrão bastante popular pelo qual os aplicativos usam vários repositórios de dados, nos quais cada repositório é usado para seus próprios propósitos, por exemplo, armazenar a forma canônica de dados (MySQL, etc.), fornecer recursos avançados de pesquisa (ElasticSearch etc.) .), armazenamento em cache (Memcached, etc.) e outros. Normalmente, ao usar vários armazenamentos de dados, um deles funciona como armazenamento principal e o outro como armazenamento derivado. O único problema é como sincronizar esses armazenamentos de dados.

Examinamos vários padrões diferentes que tentavam resolver o problema de sincronizar vários repositórios, como entrada dupla, transações distribuídas etc. No entanto, essas abordagens têm limitações significativas em termos de uso, confiabilidade e manutenção na vida real. Além da sincronização de dados, alguns aplicativos também precisam enriquecer os dados chamando serviços externos.

Para resolver esses problemas, o Delta foi desenvolvido. A Delta é uma plataforma consistente e orientada a eventos para sincronizar e enriquecer dados.

Soluções existentes

Entrada dupla

Para sincronizar dois armazenamentos de dados, você pode usar a gravação dupla, que grava em um armazenamento e, em seguida, grava imediatamente em outro. O primeiro registro pode ser repetido e o segundo pode ser interrompido se o primeiro falhar após esgotar o número de tentativas. No entanto, dois repositórios de dados podem parar de sincronizar se a gravação no segundo repositório falhar. Esse problema geralmente é resolvido com a criação de um procedimento de recuperação que pode periodicamente transferir novamente os dados do primeiro armazenamento para o segundo ou fazer isso apenas se forem encontradas diferenças nos dados.

Problemas:

A execução de um procedimento de recuperação é um trabalho específico que não pode ser reutilizado. Além disso, os dados entre armazenamentos permanecem fora de sincronia até que o procedimento de recuperação seja concluído. A solução é complicada se mais de dois repositórios de dados forem usados. Por fim, o procedimento de recuperação pode adicionar estresse à fonte de dados original.

Tabela de log de alterações

Quando ocorrem alterações em um conjunto de tabelas (por exemplo, inserindo, atualizando e excluindo registros), os registros de alterações são adicionados à tabela de log como parte da mesma transação. Outro encadeamento ou processo constantemente solicita eventos da tabela de log e os grava em um ou mais armazenamentos de dados, quando se torna necessário excluir eventos da tabela de log após confirmar o registro de todos os armazenamentos.

Problemas:

Esse padrão deve ser implementado como uma biblioteca e, idealmente, sem alterar o código do aplicativo que o utiliza. Em um ambiente poliglota, a implementação dessa biblioteca deve existir em qualquer idioma necessário, mas é muito difícil garantir a coordenação das funções e do comportamento entre os idiomas.

Outro problema está na obtenção de alterações de esquema em sistemas que não suportam alterações de esquema transacional [1] [2], como o MySQL. Portanto, um modelo para fazer uma alteração (por exemplo, alterar um esquema) e gravá-lo na tabela de log de alterações nem sempre funcionará.

Transações Distribuídas

As transações distribuídas podem ser usadas para dividir uma transação entre vários armazenamentos de dados heterogêneos, para que a operação seja confirmada em todos os armazenamentos utilizados ou não confirmada em nenhum deles.

Problemas:

As transações distribuídas são um grande problema para data warehouses heterogêneos. Por sua natureza, eles só podem confiar no menor denominador comum dos sistemas envolvidos. Por exemplo, transações XA bloqueiam a execução se ocorrer uma falha durante o processo de preparação. Além disso, o XA não fornece detecção de conflito e não suporta esquemas otimistas de gerenciamento de simultaneidade. Além disso, alguns sistemas como o ElasticSearch não suportam XA ou qualquer outro modelo de transação heterogêneo. Assim, garantir a atomicidade da gravação em várias tecnologias de armazenamento de dados continua sendo uma tarefa muito difícil para aplicativos [3].

Delta

O Delta foi projetado para lidar com as limitações das soluções de sincronização de dados existentes e também enriquece os dados em tempo real. Nosso objetivo era abstrair todos esses pontos complexos dos desenvolvedores de aplicativos, para que eles pudessem se concentrar totalmente na implementação da funcionalidade do negócio. A seguir, descreveremos a "Pesquisa de filme", o caso de uso real do Delta da Netflix.

A Netflix faz uso extensivo da arquitetura de microsserviço e cada microsserviço normalmente atende a um tipo de dados. As principais informações sobre o filme são retiradas em um microsserviço chamado Movie Service, e os dados relacionados, como informações sobre produtores, atores, fornecedores e outros, são gerenciados por vários outros microsserviços (ou seja, Deal Service, Talent Service e Vendor Service).
Os usuários comerciais do Netflix Studios geralmente precisam pesquisar por vários critérios para filmes, e é por isso que é muito importante que eles possam pesquisar todos os dados relacionados aos filmes.

Antes da Delta, a equipe de busca de filmes precisava recuperar dados de vários microsserviços antes de indexar os dados do filme. Além disso, a equipe teve que desenvolver um sistema que atualizasse periodicamente o índice de pesquisa, solicitando alterações de outros microsserviços, mesmo que não houvesse alterações. Esse sistema rapidamente ficou cheio de complexidade e ficou difícil de manter.

Figura 1. Sistema de votação antes da Delta
Depois de usar o Delta, o sistema foi simplificado para um sistema orientado a eventos, conforme mostrado na figura a seguir. Os eventos do CDC (Change-Data-Capture) são enviados aos tópicos Keystone Kafka usando o Delta-Connector. Um aplicativo Delta criado usando o Delta Stream Processing Framework (baseado no Flink) recebe eventos CDC do tópico, os enriquece, invocando outros microsserviços e, finalmente, passa os dados enriquecidos para o índice de pesquisa no Elasticsearch. Todo o processo ocorre quase em tempo real, ou seja, assim que as alterações são registradas no data warehouse, os índices de pesquisa são atualizados.

Figura 2. Pipeline de dados usando Delta
Nas seções a seguir, descrevemos o trabalho do Delta-Connector, que se conecta ao repositório e publica eventos do CDC no nível de transporte, que é uma infraestrutura de transmissão de dados em tempo real que direciona eventos do CDC para tópicos do Kafka. E no final, falaremos sobre a estrutura de processamento de fluxo Delta que os desenvolvedores de aplicativos podem usar para a lógica de processamento e enriquecimento.

CDC (captura de dados alterados)

Desenvolvemos um serviço CDC chamado Delta-Connector, que pode capturar alterações confirmadas do armazenamento de dados em tempo real e gravá-las no fluxo. Alterações em tempo real são obtidas do log de transações e despejos de armazenamento. Os despejos são usados porque os logs de transações geralmente não armazenam todo o histórico de alterações. As alterações geralmente são serializadas como eventos Delta, para que o destinatário não precise se preocupar com a origem da mudança.

O Delta-Connector suporta vários recursos adicionais, como:

Capacidade de gravar saída personalizada após Kafka.
A capacidade de ativar despejos manuais a qualquer momento para todas as tabelas, uma tabela específica ou para determinadas chaves primárias.
Os despejos podem ser coletados por pedaços, portanto, não há necessidade de iniciar tudo de novo no caso de uma falha.
Não há necessidade de colocar bloqueios nas tabelas, o que é muito importante para que o tráfego de gravação no banco de dados nunca seja bloqueado pelo nosso serviço.
Alta disponibilidade devido a backups nas zonas de disponibilidade da AWS.

Atualmente, oferecemos suporte ao MySQL e Postgres, incluindo implantações no AWS RDS e Aurora. Também apoiamos Cassandra (multi-master). Você pode aprender mais sobre o Delta-Connector neste blog .

Kafka e nível de transporte

A camada Delta Event Transport é construída no serviço de mensagens da plataforma Keystone .

Portanto, historicamente, a publicação na Netflix foi otimizada para maior disponibilidade, em vez de longevidade (consulte o artigo anterior ). Um compromisso foi a inconsistência potencial dos dados do broker em vários cenários de fronteira. Por exemplo, a eleição de líder impuro é responsável por garantir que o destinatário potencialmente duplique ou perca eventos.

Com a Delta, queríamos obter garantias mais fortes de durabilidade, a fim de garantir a entrega de eventos CDC para armazenamentos de derivativos. Para fazer isso, propusemos um cluster Kafka especialmente projetado como um objeto de primeira classe. Você pode observar algumas configurações do broker na tabela abaixo:

Nos clusters Keystone Kafka, a eleição de líder impuro geralmente é ativada para garantir a disponibilidade do editor. Isso pode resultar na perda de mensagens se uma réplica não sincronizada for selecionada como líder. Para o novo cluster Kafka altamente confiável, a opção de eleição de líder impuro é desabilitada para evitar a perda de mensagens.

Também aumentamos o fator de replicação de 2 para 3 e o mínimo de réplicas de sincronizaçãode 1 a 2. Os editores que estão gravando neste cluster exigem respostas de todos os outros, garantindo que 2 de 3 réplicas tenham as mensagens mais atualizadas enviadas pelo editor.

Quando a instância do broker sai, a nova instância substitui a antiga. No entanto, o novo intermediário precisará acompanhar réplicas não sincronizadas, o que pode levar várias horas. Para reduzir o tempo de recuperação desse cenário, começamos a usar a Amazon Elastic Block Store em vez de discos de corretores locais. Quando uma nova instância substitui uma instância concluída do broker, ela anexa o volume EBS que a instância concluída tinha e começa a acompanhar as novas mensagens. Esse processo reduz o tempo para eliminar o backlog de várias horas para vários minutos, já que a nova instância não precisa mais ser replicada de um estado vazio. Em geral, os ciclos de vida separados do armazenamento e do intermediário reduzem significativamente o efeito da alteração do intermediário.

Para aumentar ainda mais a garantia da entrega de dados, usamos um sistema de rastreamento de mensagens para detectar qualquer perda de mensagens em condições extremas (por exemplo, sincronização do relógio no líder da seção).

Estrutura de processamento de fluxo

O nível de processamento na Delta é baseado na plataforma Netflix SPaaS, que permite a integração do Apache Flink com o ecossistema da Netflix. A plataforma fornece uma interface com o usuário que controla a implantação de tarefas Flink e a orquestração de clusters Flink na parte superior de nossa plataforma de gerenciamento de contêineres Titus. A interface também gerencia as configurações de tarefas e permite que os usuários façam alterações dinamicamente na configuração sem precisar recompilar as tarefas do Flink.

A Delta fornece uma estrutura de processamento de fluxo para dados baseados em Flink e SPaaS, que usa anotaçõesDSL (Domain Specific Language) para abstrair detalhes técnicos. Por exemplo, para determinar a etapa pela qual os eventos serão enriquecidos chamando serviços externos, os usuários precisam escrever a próxima DSL e a estrutura criará um modelo baseado nela que o Flink executará.

Figura 3. Exemplo de enriquecimento de DSL no Delta

A estrutura de processamento não apenas reduz a curva de aprendizado, mas também fornece funções gerais de processamento de fluxo, como desduplicação, esquematização, bem como flexibilidade e tolerância a falhas para resolver problemas comuns no trabalho.

O Delta Stream Processing Framework consiste em dois módulos principais, o módulo DSL e API e o módulo Runtime. O módulo DSL e API fornece as APIs DSL e UDF (Função Definida pelo Usuário) para que os usuários possam gravar sua própria lógica de processamento (como filtragem ou transformações). O módulo Runtime fornece uma implementação do analisador DSL, que constrói uma representação interna das etapas de processamento nos modelos DAG. O componente Execution interpreta os modelos DAG para inicializar as instruções reais do Flink e, finalmente, iniciar o aplicativo Flink. A arquitetura da estrutura é ilustrada na figura a seguir.

Figura 4. Arquitetura da Delta Stream Processing Framework

Essa abordagem tem várias vantagens:

- Flink SPaaS.
, - (UDF).
Delta , , .

A Delta está em produção há mais de um ano e desempenha um papel fundamental em muitos aplicativos do Netflix Studio. Ajudou as equipes a implementar casos de uso, como indexação de pesquisa, armazenamento de dados e fluxos de trabalho orientados a eventos. A seguir, é apresentada uma visão geral da arquitetura de alto nível da plataforma Delta.

Figura 5. A arquitetura de alto nível da Delta.

Agradecimentos

Gostaríamos de agradecer às seguintes pessoas que contribuíram para a criação e desenvolvimento da Delta na Netflix: Allen Wang, Charles Zhao, Jaebin Yoon, Josh Snyder, Kasturi Chatterjee, Mark Cho, Olof Johansson, Piyush Goyal, Prashanth Ramdas, Raghuram Onti Srinivasan, Sandeep Gupta , Steven Wu, Tharanga Gamaethige, Yun Wang e Zhenzhong Xu.

Fontes

dev.mysql.com/doc/refman/5.7/en/implicit-commit.html
dev.mysql.com/doc/refman/5.7/en/cannot-roll-back.html
Martin Kleppmann, Alastair R. Beresford, Boerge Svingen: Online event processing. Commun. ACM 62(5): 43–49 (2019). DOI: doi.org/10.1145/3312527

: «Data Build Tool Amazon Redshift».

Delta: plataforma de enriquecimento e sincronização de dados