@phpeterle Blog Na Woovi rodamos alguns jobs Apache Flink em produção, todos lendo mudanças do MongoDB (CDC, change data capture). Um indexa documentos no Elasticsearch, outros alimentam analytics, e o mais recente publica eventos de domínio no Kafka. Esse último substituiu um processo Node.js que fazia o mesmo trabalho com change streams, e a migração me obrigou a pensar com cuidado no que o Flink entrega e no que ele cobra.
O processo Node funcionou bem por muito tempo. Os problemas começaram com o crescimento da empresa e, em especial, depois de um pico no volume de eventos: um único processo, sem controle de carga e sem como escalar para mais de uma réplica, passou a estourar memória e a perder eventos justamente nos horários de mais movimento. Foi a necessidade de escala e de disponibilidade que nos levou ao Flink.
Este post é a parte conceitual dessa conversa. A história da migração, com os incidentes e os números, está em Do Node.js ao Flink.
O que é o Flink
Flink é um motor de processamento de streams com estado. Você descreve o job como um grafo de operadores (ler de uma fonte, transformar, agrupar por chave, escrever num destino) e o Flink distribui esse grafo entre processos. O JobManager coordena e os TaskManagers executam as tarefas em slots.
Cada operador pode rodar com paralelismo N. Um keyBy garante que todos os registros de uma mesma chave caiam na mesma instância do operador, na ordem em que chegaram, e é isso que permite escalar sem perder a ordem por entidade.
Ele roda na JVM. A API principal é a DataStream API, em Java, e por cima dela existem a Table API, o Flink SQL e o PyFlink. Fontes e destinos vêm de conectores: Kafka, sistemas de arquivos, Elasticsearch, JDBC e, pelo projeto Flink CDC, bancos como MySQL, Postgres e MongoDB.
flowchart LR
S[Fonte<br/>Kafka, CDC...] --> M[map / filter]
M --> K{keyBy}
K --> A1[operador<br/>instância 1]
K --> A2[operador<br/>instância 2]
A1 --> D[Destino<br/>Kafka, ES...]
A2 --> D
CP[(Checkpoint<br/>S3)] -.snapshot.- S
CP -.snapshot.- A1
CP -.snapshot.- A2
As ideias que importam
Estado e checkpoint
Quase todo job útil guarda alguma coisa entre um registro e outro: uma contagem por chave, uma janela de tempo aberta, a posição de leitura da fonte. No Flink isso é estado, e quem cuida dele é o framework. Você não precisa manter essa posição num Redis à parte e torcer para salvar na hora certa.
De tempos em tempos o Flink tira um checkpoint. Ele injeta barreiras no fluxo a partir das fontes, e cada operador salva o seu estado quando a barreira passa por ele. O resultado é um snapshot consistente do job inteiro, com a posição das fontes e o estado de cada operador, gravado num storage durável (no nosso caso, S3). Se uma tarefa falha, o job volta do último checkpoint e relê a fonte a partir da posição que foi salva junto.
O estado vive num state backend. O hashmap mantém tudo no heap da JVM e é rápido; o RocksDB guarda em disco local e aguenta estados maiores que a memória. Os savepoints são checkpoints disparados à mão, usados para atualizar o código do job sem perder a posição.
Backpressure
Entre operadores, o Flink usa controle de fluxo por crédito: um operador só envia dados quando o seguinte tem buffer livre. Se o destino fica lento, a lentidão sobe pelo grafo até a fonte, que passa a ler mais devagar. O sistema desacelera, e o trabalho pendente fica na fonte (no oplog, no tópico do Kafka) em vez de se acumular na memória do processo.
Tempo de evento
Para agregações por janela, o Flink separa o momento em que o evento aconteceu do momento em que ele chegou, e usa watermarks para decidir quando uma janela pode fechar. No CDC isso quase não aparece, mas é um dos principais motivos para usar Flink em analytics de streaming.
Escala e disponibilidade
Como ele escala
Um job Flink é dividido em subtarefas. Se um operador tem paralelismo 8, existem 8 cópias dele, espalhadas pelos slots dos TaskManagers, cada uma processando uma fatia dos dados. Para aguentar mais carga, você aumenta o paralelismo e adiciona TaskManagers. O código do job não muda.
Quando há estado por chave, o Flink divide o espaço de chaves em key groups, e cada subtarefa fica responsável por um conjunto deles. Ao mudar o paralelismo, o job para com um savepoint e volta com o novo número de subtarefas, e o Flink redistribui os key groups (e o estado de cada um) entre elas. Cada chave continua indo sempre para o mesmo lugar, então a ordem por entidade se mantém. O operator do Kubernetes tem ainda um autoscaler que ajusta o paralelismo de acordo com a carga.
A escala da leitura depende da fonte. Uma fonte Kafka distribui as partições do tópico entre as subtarefas, e mais partições permitem mais paralelismo. Um change stream do MongoDB não se divide, então ali a escala vem de outro jeito: cada coleção vira um pipeline independente, e o que vem depois da leitura (montagem do evento, lookups, escrita) pode rodar com paralelismo maior.
Escalar não se resume ao Flink. O que ele publica no Kafka escala do lado dos consumidores: com o tópico particionado pela chave da entidade, dá para colocar uma réplica de consumidor por partição no mesmo consumer group. Isso era impossível com o processo Node lendo direto do change stream, que não tem consumer group e processaria tudo em dobro com uma segunda réplica.
Como ele se mantém disponível
A disponibilidade do Flink vem da recuperação. Ele não roda duas cópias ativas do mesmo job. Quando algo falha, ele volta ao último estado consistente e continua dali.
Se um TaskManager cai, o JobManager percebe, reagenda as tarefas afetadas em outros slots e restaura o estado a partir do último checkpoint. Com failover por região, só as tarefas ligadas à parte que falhou reiniciam; num job com várias fontes independentes, as outras seguem rodando. A estratégia de restart decide quantas vezes e com que intervalo ele tenta.
O próprio JobManager também pode cair. Com alta disponibilidade ligada (no Kubernetes, usando ConfigMaps para eleição de líder e metadados), um novo JobManager assume, encontra o último checkpoint registrado e retoma o job. Como os checkpoints ficam num storage durável como o S3, perder um pod ou um nó inteiro não leva o estado junto.
Upgrades seguem a mesma lógica: o job para com um savepoint, a nova versão sobe a partir dele e continua da mesma posição na fonte.
Esse modelo tem custo. Durante a recuperação o job fica parado por alguns segundos ou mais, e tudo o que foi processado depois do último checkpoint é processado de novo. Quem consome a saída precisa tolerar esse atraso e essas duplicatas, e o intervalo de checkpoint vira uma troca entre custo de snapshot e quanto trabalho se refaz a cada falha. No nosso caso, o Kafka entre o Flink e os consumidores absorve as pausas, e o id determinístico do evento resolve as duplicatas.
Vantagens
Garantia de entrega explícita
O checkpoint amarra a posição da fonte ao estado e à saída. O KafkaSink em modo AT_LEAST_ONCE só deixa o checkpoint completar depois de dar flush no que enviou. Então nenhum registro lido se perde: no pior caso, depois de uma falha, os registros desde o último checkpoint são enviados de novo. Em modo EXACTLY_ONCE, o sink escreve dentro de uma transação do Kafka e só faz commit quando o checkpoint fecha.
Num consumidor escrito à mão, essa garantia depende de alguém lembrar de salvar a posição depois do efeito, e é fácil errar. No processo Node que substituímos, a posição era salva antes de o handler rodar, e qualquer crash entre uma coisa e outra perdia o evento sem deixar log.
Backpressure de ponta a ponta
Em Node, um EventEmitter não espera a Promise de um listener async. Cada evento dispara um handler novo, e sob carga isso vira concorrência ilimitada, depois falta de memória, depois perda. No Flink, um destino lento faz a fonte ler mais devagar. A carga extra aparece como atraso, que dá para medir e resolver.
Escala e isolamento
Paralelismo é configuração, e com keyBy na chave certa você escala sem perder a ordem por entidade. Para jobs com várias fontes independentes, o Flink também tem ferramentas de isolamento que fazem diferença:
- slot sharing groups, que colocam os pipelines pesados em slots próprios;
- failover por região, em que a falha de um pipeline reinicia só as tarefas ligadas a ele;
- estratégias de restart como
exponential-delay, que tenta de novo com espera crescente em vez de desistir depois de N falhas.
No nosso job de eventos, cada coleção do MongoDB é um pipeline separado. Se a fonte de uma coleção dá problema, só ela reinicia.
Eficiência
Na nossa medição, o job Flink de eventos usa cerca de 10 vezes menos CPU em média que o processo Node que ele substituiu. A comparação não é exata: parte do trabalho saiu do processo antigo e foi para consumidores fora do Flink, e o processo antigo fazia uma leitura extra do documento a cada update. Mesmo assim, a diferença de ordem de grandeza ficou clara.
Ecossistema
Conectores que já participam do checkpoint, métricas por operador, interface web com o grafo de backpressure e um operator para Kubernetes. Num consumidor próprio, cada uma dessas coisas seria código para escrever e manter.
Desvantagens
Operar Flink dá trabalho
Um job Flink é um sistema distribuído com estado. Você passa a operar JobManager, TaskManagers, checkpoints no S3, o operator do Kubernetes e savepoints a cada upgrade.
O modo de falha mais perigoso é silencioso. Um job em FAILED não derruba nenhuma API, ele só para de produzir. Já tivemos o job de indexação parado por quase dois dias depois de uma manutenção do banco, e um ambiente de staging uma semana parado sem ninguém notar. Em outra ocasião, os deploys de dois jobs pararam de chegar ao operator porque um recurso ficou preso em Terminating, enquanto a pipeline de CI continuava verde.
Se o Flink está no caminho crítico, ele precisa de alerta para o estado do job, a idade do último checkpoint e o atraso da fonte, e de plantão no mesmo nível de qualquer serviço de produção.
Memória reservada
A JVM reserva memória de forma fixa. Nosso job de eventos ocupa pouco mais de 2 GiB mesmo com pouca carga. Para jobs pequenos, isso pesa mais que a CPU economizada.
Outra linguagem
Se o resto da empresa escreve TypeScript, qualquer lógica de domínio que precise rodar dentro do job tem que ser portada para Java, ou caber em SQL. No nosso caso, o evento de cobrança precisava carregar o BR Code do Pix. Isso virou cerca de 1.400 linhas de Java, mais testes de paridade gerados a partir do código TypeScript, e agora toda mudança de um lado precisa ser acompanhada do outro.
Exactly-once custa latência
O exactly-once do Flink vale para o estado interno do job. Para o Kafka, ele depende de transações, e um consumidor read_committed só enxerga os registros quando a transação fecha, no próximo checkpoint. Com checkpoint a cada 60 segundos, isso pode dar quase um minuto de atraso, o que não serve para uma tela que mostra “pagamento confirmado”.
Por isso usamos at-least-once. O preço é que duplicata passa a fazer parte do contrato: o evento precisa de um id determinístico, e todo consumidor com efeito externo precisa deduplicar.
O paralelismo para na fonte
O Flink escala até onde a fonte deixa. Um change stream do MongoDB não se divide, então cada coleção é lida por uma única tarefa. E se a leitura da fonte fica lenta, o job inteiro herda o atraso. Nossos jobs de CDC leem de um único nó do replica set e ficam dezenas de minutos atrás enquanto o backup noturno do banco roda. O problema não está no Flink, mas é nele que o atraso aparece.
Curva de aprendizado
Estado por chave, serialização de tipos, watermarks, compatibilidade de estado entre versões do job, tuning de memória da JVM e de checkpoint. Para quem vem de serviços Node.js sem estado, é bastante coisa nova de uma vez.
Quando eu usaria
Flink compensa quando perder um registro é problema de negócio e você precisa de uma garantia de entrega que dê para explicar, quando o volume é alto ou oscila muito e o backpressure faz diferença, ou quando há estado de verdade (agregações, joins, janelas). Em todos esses casos, o time precisa aceitar operar mais uma peça com plantão.
Para volume pequeno, sem estado, em que perder um evento de vez em quando é aceitável, um consumidor simples na linguagem do time sai mais barato. E se o problema é só publicar eventos de negócio que o seu próprio código escreve, uma outbox no caminho de escrita resolve sem CDC. Nós não fomos por esse caminho porque as coleções principais têm dezenas de pontos de escrita, entre código, scripts e migrations, e o CDC pega todos eles sem depender de cada um lembrar de gravar na outbox.