Adoção do Flink SQL

• A equipe Azar Matching Dev Team, da Hyperconnect, escolheu implementar o processamento de streaming com SQL em vez de código de aplicação para dividir um app legado monolítico em Flink que consumia 96 CPUs
• Dividir em vários apps Flink melhora o isolamento, mas aumenta a carga operacional; por isso, concluíram que o Flink SQL era mais adequado em termos de produtividade e eficiência operacional, aproveitando o entendimento interno do Flink pela equipe
• O Flink SQL oferecia suporte a Checkpoint/Savepoint, HA do JobManager, redistribuição de TaskManager, janelas, joins, event time, watermark, além de UDFs e conectores customizados, o que o tornou mais adequado ao contexto da equipe do que ksqlDB e Spark Structured Streaming
• O ambiente de produção foi montado como um cluster Flink em modo Session sobre Kubernetes, e o deploy de queries e a interrupção de jobs foram tratados em modelo GitOps com a API do Flink SQL Gateway e GitHub Actions
• Depois de cerca de 1 ano de operação estável e com expansão em andamento, a equipe ainda considera incômodos o redeploy de queries e alterações na configuração do cluster, e planeja melhorias com base no padrão GitOps Controller

---

Contexto: por que substituir um app legado pesado de streaming

• A Azar Matching Dev Team gerenciava vários apps baseados em Flink, e entre eles havia um app legado pesado que consumia 96 CPUs
• Esse app tinha uma estrutura monolítica, reunindo em um só lugar funções como múltiplos joins de eventos de matching, publicação condicional de eventos e armazenamento de flags no Redis
• Depois de uma mudança nos nós de execução por conta de um trabalho de infraestrutura em toda a empresa, o app deixou de funcionar corretamente, e não era fácil resolver rapidamente só com tuning simples
• Como a funcionalidade crítica de join de eventos já havia sido implementada em um novo app Flink de outro projeto, era necessário substituir a parte posterior ao join, responsável pela publicação condicional de eventos e execução da lógica

Comparação das abordagens de substituição

• Implementar tudo em um único app Flink reduz o número de itens a gerenciar, mas aumenta a chance de voltar a ter um app gigante, além de fazer com que a falha de uma parte possa afetar outras funções
• Dividir em vários apps Flink permite gerenciar cada app de forma independente, mas, à medida que o número de apps cresce, também aumentam a carga com cluster, recursos e deploy
• Com Flink SQL, a lógica pode ser definida em queries para desenvolvimento rápido e com apenas um cluster para administrar, mas lógicas complexas são difíceis de expressar só com SQL, além de exigir experiência operacional com cluster
• Como a equipe já tinha um bom entendimento da implementação interna do Flink, concluiu que o Flink SQL tinha vantagens em produtividade e eficiência operacional

Por que escolheram Flink SQL

• O Flink SQL permite implementar apps de processamento de streaming de eventos em SQL, sem escrever diretamente código de aplicação
• Em termos de alta disponibilidade (HA), o Flink suporta processamento com estado e permite salvar e restaurar o estado do job em intervalos regulares ou em momentos específicos por meio de Checkpoint e Savepoint
  • O JobManager pode ser configurado em modo HA com líder e standby
  • Se parte dos TaskManagers falhar, os trabalhos desses TaskManagers podem ser redistribuídos para outros TaskManagers conforme a estratégia de retry do job
• Só com a sintaxe SQL já é possível cobrir os principais recursos de processamento de streaming
  • SELECT transforma o formato dos dados e WHERE filtra registros
  • JOIN combina vários streams e UNION pode unir streams
  • Há suporte a janelas como tumbling, hopping (sliding) e session window
  • Event time e watermark permitem definir a tolerância para dados atrasados
• UDFs e conectores customizados ampliam os recursos além do que existe por padrão
  • A maior parte do legado existente seguia padrões usando comandos Redis SET ou INCR, e como não havia um Redis Connector oficial do Flink, a equipe escreveu e usou seu próprio Redis Connector
  • Na época, não existia função built-in para calcular a interseção de arrays, então isso foi implementado como UDF e usado nas queries

Comparação com ksqlDB e Spark Structured Streaming

• O ksqlDB já fazia parte da plataforma Confluent usada internamente para Kafka, e também havia casos de uso dentro da empresa
• Ainda assim, a equipe considerou que seu comportamento de HA em processamento stateful era ineficiente
  • Em failover de operações stateful, todo o changelog com o histórico das mudanças de estado precisa ser reproduzido, o que pode tornar o failover demorado
  • O modelo de manter réplicas do stream de processamento e atualizar continuamente o changelog no state interno faz com que as mesmas operações também sejam executadas nas réplicas, podendo dobrar o consumo de recursos
  • Mais detalhes em Configuring ksqlDB for High Availability | Confluent Developer
• O Spark Structured Streaming é um motor de streaming baseado no engine do Spark SQL
  • Havia casos de uso internos e era possível criar UDFs e Custom Sinks
  • Seu ecossistema é maior e mais consolidado que o do Flink
• Como o Spark opera em micro-batches, pode haver latência por registro, o que pode ser uma desvantagem em relação ao Flink quando o processamento em tempo real é importante
• A equipe quase não tinha experiência com Spark e ainda precisaria implementar Custom Sink, então a escolha pelo Spark não pareceu simples

Montagem do ambiente de cluster

• Localmente, é possível baixar os binários na página oficial do Flink e iniciar um cluster com {FLINK_HOME}/bin/start-cluster.sh
• Ao executar {FLINK_HOME}/bin/sql-client.sh, abre-se a CLI do Flink SQL, onde é possível submeter queries de teste como SELECT 1;
• Depois de submeter uma query, é possível verificar na UI web do Flink que ela foi convertida em Job e executada
• Com o lançamento do Flink SQL Gateway no fim de 2022, passou a ser possível submeter queries via HTTP

Arquitetura operacional baseada em Kubernetes

• Como a maioria dos serviços internos roda sobre Kubernetes, o cluster Flink SQL também foi montado sobre Kubernetes
• Os apps Flink existentes eram todos implantados e operados em Application mode
  • Nesse modelo, cada aplicação sobe seu próprio cluster
  • No Kubernetes, cada app roda com seus próprios Pods de JobManager e TaskManager
  • Isso traz vantagens em independência e isolamento entre apps, além de facilitar configuração por job e gestão de dependências
• Como o Flink SQL submete Jobs a um cluster já em execução, foi necessário subir JobManager e TaskManager em Session mode
• O cluster foi configurado com base no guia Stand Alone Cluster on Kubernetes
• Para o ambiente HA, foi usada como referência a configuração de High-Availability with Standalone Kubernetes, usando S3 em high-availability.storageDir
• A abordagem Native Kubernetes foi considerada inadequada para a infraestrutura interna, porque sobe o cluster com shell scripts fornecidos pela plataforma, enquanto o ambiente da empresa define e faz deploy das configurações de deployment diretamente

Configuração de HA e integração com S3

• Para HA e integração com S3, usaram a seguinte configuração em config.yaml

high-availability.type: kubernetes
high-availability.storageDir: s3://{s3-path-for-flinksql-recovery}
kubernetes.cluster-id: {cluster-id}
kubernetes.namespace: {k8s-namespace}


# namespace 내의 service account 를 통해 Kubernetes cluster 에 접근할 수 있도록 권한을 부여하는 작업이 필요할 수 있습니다.
kubernetes.service-account: {k8s-service-account-for-flinksql}

• Em ambiente HA, são iniciados dois pods de JobManager, e cada um precisa ter um endereço diferente para que a eleição de líder e outros mecanismos funcionem corretamente
• Os argumentos de execução do contêiner do JobManager foram configurados assim

args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]

• Com isso, informações como o pod de JobManager atualmente eleito como líder e o Job ID em execução ficam armazenadas no Kubernetes ConfigMap para uso no mecanismo de HA

Deploy de queries em modelo GitOps

• O Flink ainda não oferece diretamente uma UI web ou ferramenta dedicada para Flink SQL
• Um caso de integração com Hue foi avaliado como PoC, mas na época havia problemas de compatibilidade de versão com o Flink SQL Gateway, o que exigiria desenvolvimento adicional e também bastante tempo para montar o ambiente
• Como o padrão GitOps já era bastante usado internamente, a equipe implementou GitHub Actions para fazer deploy de queries e interromper Jobs
• Dentro do repositório, criaram pastas por Job e reuniram nelas os arquivos SQL com as queries a executar
• O GitHub Actions recebe o nome da pasta e identifica o arquivo SQL do qual deve extrair as queries
• A implementação faz chamadas à REST API do Flink SQL Gateway e foi escrita em Python por ser simples e fácil de testar

Casos operacionais e resposta a incidentes

• Não houve experiência de falha do JobManager, mas, pela configuração de HA, se ele falhar outro JobManager pode ser eleito líder e continuar o trabalho
• Ocasionalmente houve falhas de TaskManager, em sua maioria casos em que o Pod foi reiniciado por políticas de QoS do Kubernetes
• Foi confirmado que, mesmo com falha de alguns TaskManagers, o trabalho continuava após redistribuição para outros TaskManagers
• As falhas de query ocorreram principalmente por entrada de dados anormal ou falta de recursos computacionais
  • Ao ler dados JSON, formatos JSON inválidos podem ser ignorados com a opção json.ignore-parse-errors
  • Ao extrair dados de um path específico com JSON_VALUE, erros por ausência de valor ou tipo diferente podem ser tratados definindo um valor padrão com DEFAULT {VALUE} ON ERROR
  • Quando a CPU do TaskManager ultrapassa 100% ou falta memória, a equipe aumenta os recursos do TaskManager ou o parallelism da query e então faz o redeploy
• Em reinícios de cluster por mudança de configuração ou adição de UDF, alguns Jobs chegaram a falhar
  • A causa geralmente estava em configurações inadequadas de timeout ou retry
  • A equipe ajustou timeout e retry para que o Job não encerrasse as tentativas cedo demais e continuasse tentando até o cluster estabilizar após o reinício

Restrições em mudança de query e restauração de state

• Ao modificar condições de uma query e fazer novo deploy, restaurar o state por savepoint só é viável em alterações muito simples, como mudança de valores em expressões condicionais
• Se as condições de window mudam, o state também muda, o que dificulta manter compatibilidade e pode inviabilizar a restauração com savepoint
• Se o state precisa ser preservado e os requisitos mudam com frequência, pode ser mais adequado escrever a aplicação diretamente

Pontos de monitoramento

• O Flink oferece muitos metrics nativos, então, com a infraestrutura interna de monitoramento e um Metric Reporter adequado, é fácil montar o ambiente de observabilidade
• numRunningJobs indica a quantidade de Jobs atualmente em execução no cluster; se o valor cair de repente e permanecer baixo, é possível concluir que houve Job com falha
• taskmanager.cpu.load e taskmanager.memory.used ajudam a entender o uso de recursos do cluster
• busyTimeMsPerSecond permite verificar quanto cada TaskManager está ocupado por Job
• Ao usar Kafka como source, records-lag-max permite checar rapidamente o atraso dos dados

Exemplo: agregação em janela de eventos de login no Kafka

• O exemplo do apêndice recebe eventos do Kafka e publica no Kafka, a cada 10 segundos, a contagem de eventos de login do último 1 minuto
• Os dados de entrada estão em formato JSON e incluem os campos event_time, event_type e data.user_id
• A query configura pipeline.name, parallelism.default e table.exec.state.ttl
• A tabela de entrada login_event usa o conector Kafka e formato JSON, com json.ignore-parse-errors definido como true
• row_time é gerado a partir de event_time, e a watermark é configurada para tratar eventos que cheguem com até 5 segundos de atraso em relação ao maior event time observado até então
• A tabela de saída windowed_login_count publica o resultado em um tópico Kafka e inclui o campo proc_time AS PROCTIME()
• HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE) cria uma hopping window de 1 minuto com intervalo de 10 segundos, e COUNT(*) agrega o número de registros na janela

Resultados operacionais e melhorias pendentes

• A equipe conseguiu aproveitar sua experiência prévia com Flink para adicionar várias funcionalidades com mais facilidade e rapidez do que antes
• A avaliação foi de que os resultados foram satisfatórios em termos de produtividade e eficiência operacional
• Após a adoção, o sistema funcionou de forma estável por cerca de 1 ano sem praticamente nenhum trabalho operacional relevante
• Atualmente, a operação está sendo expandida gradualmente
• Ainda há incômodo com o redeploy de queries e mudanças na configuração do cluster, e a equipe planeja melhorar o ambiente de deploy de queries com a implementação do padrão GitOps Controller

[flgkselql98]

Sistemas distribuídos como o Flink precisam manter 2 ou 3 racks para garantir HA, e parece que, ao integrá-lo ao Kubernetes, eles conseguiram garantir essa HA. Mas, no fim, ainda seria preciso considerar os recursos dos slave nodes do kube; fico pensando se montaram nós dedicados só para rodar Flink (porque, quando a carga do Flink aumentar, parece provável haver problema de queda de slave node).
Nessa perspectiva, existe alguma vantagem em usar Kubernetes?

Além disso, quando se usa funções de janela no Flink, os dados daquele intervalo ficam mantidos em memória para que o sql join funcione. Olhando pelo ponto de vista de trade-off, será que o Flink é uma boa escolha? Se, com o passar do tempo, esse SQL cada vez maior + o job morrerem, o estrago seria enorme...

Eu também fico pensando que, quando há necessidade de join na data source do topo, em vez de usar Flink, como isso poderia ser tratado descendo para o nível da aplicação.