Once Upon a Migration in Engineering

No início de muitas startups, a pressão para entregar rápido leva as equipes de engenharia a construírem MVPs (Produtos Mínimos Viáveis) e lançarem funcionalidades às pressas, visando validar o produto e alcançar product-market fit. Nessa corrida, sistemas e modelos de dados são criados para atender às necessidades imediatas. Com o passar dos anos, porém, é comum que certas decisões iniciais precisem ser reavaliadas à luz do crescimento do produto e de novas demandas (técnicas ou de negócio). É aí que entra em cena uma tarefa frequente em times de engenharia maduros: migrações de sistema e de dados.

Existem diversos tipos de migrações que podemos enfrentar. Podemos precisar migrar a arquitetura (por exemplo, extrair microservices de um monolito legado), reescrever componentes inteiros, ou, foco deste artigo, realizar grandes migrações de dados. Aqui estamos falando de alterar ou mover grandes volumes de dados de um formato, local ou esquema para outro – às vezes envolvendo múltiplos serviços e bancos de dados diferentes – de maneira segura e eficiente.

Cenário: Migração de Dados em Múltiplos Sistemas

Imagine o seguinte cenário real (simplificado para fins didáticos): sua empresa possui um ecossistema com vários serviços e bases de dados, cada um guardando informações sobre o titular de uma conta de formas redundantes. Por exemplo, um serviço de processamento de PIX e boletos armazena dados do pagador/beneficiário; outro sistema de autorização de cartão mantém referência ao titular do cartão; o serviço de extrato registra quem enviou ou recebeu cada transação, etc. Em resumo, diversos sistemas possuem cópias de referência do mesmo dado de identidade.

Agora, devido a uma nova exigência regulatória, vocês precisam modificar essa referência de titular em todos esses sistemas ao mesmo tempo (por exemplo, trocar o identificador usado para o cliente por um novo ID unificado, ou atualizar o formato de um documento). Como proceder? Uma mudança desse tipo exige coordenação cuidadosa: ou todos os serviços adotam a nova referência simultaneamente, ou corremos risco de inconsistências e falhas em cascata.

Neste artigo, vamos explorar as estratégias adotadas em uma migração de dados em larga escala semelhante a essa. Vamos discutir os desafios técnicos envolvidos, as decisões tomadas para minimizar riscos e dores de cabeça, e como utilizamos ferramentas – em especial o Elixir – para processar grandes volumes de dados de forma eficiente.

Desafios em Grandes Migrações de Dados

Realizar migrações de dados volumosas e distribuídas traz uma série of desafios técnicos e operacionais. Antes de entrarmos nas soluções, vale destacar alguns dos principais problemas que precisamos endereçar:

• Consumo de memória e recursos: Processar milhões de registros ou comandos de atualização pode facilmente esgotar memória e CPU se feito de forma ingênua. Ler um arquivo enorme inteiro para a RAM ou carregar um dataset massivo de uma só vez não escala. Precisamos de abordagens streaming (fluxo) que utilizem memória constante, mesmo com arquivos gigantes.

• Manter consistência e ordem: No cenário descrito, pode ser crucial que as atualizações ocorram em ordem e de forma coordenada. Se um sistema é atualizado antes de outro, podemos ter falhas temporárias. Além disso, se estamos executando comandos SQL, pode haver dependências entre eles que exigem ordem sequencial. Precisamos garantir execução segura e ordenada dos comandos.

• Limitações de ferramentas/banco: Algumas ferramentas de acesso a banco de dados têm restrições – por exemplo, o driver Postgrex (usado pelo Ecto no Elixir) não suporta múltiplos comandos SQL na mesma query quando usando o protocolo padrão extendido. Isso significa que não dá pra simplesmente juntar UPDATE X; UPDATE Y; UPDATE Z; em uma única chamada – temos que executar um por vez.

• Tempo de execução e desempenho: Atualizar dados em larga escala pode levar horas se feito de forma ineficiente. Precisamos pensar em formas de otimizar o throughput – seja agrupando operações em lote, paralelizando onde seguro, ou outras otimizações – sem comprometer a segurança.

• Robustez e recuperação de erros: Em um mundo ideal, todas as atualizações ocorreriam sem nenhum erro. No mundo real, porém, é possível que alguns comandos falhem – seja por dados inesperados, conflitos (ex: violação de chave única), deadlocks, ou erros transientes de conexão. Uma migração robusta deve lidar graciosamente com erros: registrar o problema e seguir em frente, ao invés de abortar tudo. Afinal, se 5 registros falharem em 1 milhão, talvez seja aceitável tratar esses 5 manualmente depois, mas queremos que os outros 999.995 sejam migrados com sucesso.

• Log e monitoramento: Executar 100 mil ou 1 milhão de operações gera muito log. Se não tivermos cuidado, podemos inundar nosso sistema de logs, dificultando achar informação útil. Queremos uma forma de consolidar logs – por exemplo, registrar todos os erros num único arquivo ou estrutura – e ter algum feedback de progresso, sem imprimir uma linha de log para cada operação bem-sucedida.

• Orquestração entre serviços: No caso de múltiplos sistemas envolvidos, há ainda o desafio de orquestrar a migração de forma centralizada ou coordenada. Podemos precisar rodar scripts em cada serviço ou banco, sincronizados de alguma maneira. Alternativamente, um serviço orquestrador poderia disparar as migrações individuais e acompanhar seu progresso. De qualquer forma, requer planejamento para que todos cheguem ao estado final consistente.

Tendo estes desafios em mente, vamos agora detalhar como abordá-los. Usaremos como espinha dorsal um caso prático de migração de dados realizado em Elixir, mas ressaltando conceitos gerais que podem ser aplicados em outras linguagens e ferramentas.

Estratégia 1: Streaming de Dados para Uso Mínimo de Memória

O primeiro mandamento de trabalhar com grandes volumes de dados é não carregar tudo de uma vez em memória. Isso vale tanto para leitura quanto para escrita.

Se o seu plano de migração envolve, por exemplo, ler um arquivo de exportação com 500 mil linhas de comandos SQL ou dados, você deve evitar ler o arquivo inteiro com um IO.puts ou similar. Em vez disso, processe em streaming, ou seja, em pedaços.

Lendo arquivos enormes de forma eficiente

Suponha que temos um arquivo .sql gigantesco armazenado em um bucket S3, contendo centenas de milhares de comandos UPDATE que precisamos executar.

Em vez de baixar todo o arquivo e carregá-lo na memória, podemos processar linha a linha em fluxo. No Elixir, a biblioteca ExAws nos ajuda nisso. Por exemplo:

• O método ExAws.S3.download_file/4 pode baixar um objeto do S3 em múltiplas partes (concorrentes), ao invés de uma tacada só. Por padrão ele usa 8 partes paralelas de 1MB cada, o que maximiza throughput sem sobrecarregar memória (8 MB no total por padrão, podendo configurar menos). Também permite baixar direto para arquivo local ou para a memória.

• Se usamos :memory como destino no download_file, podemos então encadear com ExAws.stream!/2 para obter um Stream lazy em Elixir. Esse stream vai nos entregar pedaços (chunks) de bytes conforme vamos consumindo, sem armazenar tudo antecipadamente. Importante: nenhum dado é efetivamente buscado do S3 até começarmos a consumir o stream, ou seja, só puxamos pequenos blocos na medida da necessidade, mantendo o uso de memória praticamente constante.

• Como cada chunk pode ser de até 1MB e não necessariamente termina exatamente no fim de uma linha do arquivo, precisamos juntar as partes e separá-las por linha. Podemos usar Stream.chunk_while/4 para isso: acumulamos bytes até encontrarmos um \n de fim de linha, entregamos cada linha completa, e mantemos o resto para o próximo chunk. A própria documentação do ExAws traz um exemplo de como implementar isso. Basicamente, concatenamos o chunk atual com o “resto” da linha anterior (se houver), separamos as linhas completas utilizando algo como :binary.match para encontrar quebras de linha, e guardamos o pedaço final incompleto para juntar com o próximo chunk. Com essa técnica, mesmo que a leitura do S3 seja em blocos de 1MB, o stream resultante nos fornece exatamente uma linha por vez, já devidamente separada.

Em código Elixir, poderia ficar assim (simplificado):

stream = ExAws.S3.download_file("meu-bucket", "caminho/arquivo.sql", :memory)
         |> ExAws.stream!()
         |> Stream.chunk_while("", fn chunk, acc ->
              data = acc <> chunk
              case :binary.match(data, "\n") do
                {pos, _len} -> 
                  # encontramos pelo menos um \n
                  # separa até o último \n completo
                  last_newline_index = ... # calcular posição do último \n em data
                  {complete_lines, rest} = split_at_position(data, last_newline_index)
                  {:cont, lines_list(complete_lines), rest}
                :nomatch ->
                  # nenhum \n no chunk inteiro, acumula e continua
                  {:cont, [], data}
              end
            end,
            fn acc ->
              # no fim, emite o que sobrou como última linha (se não vazio)
              if acc == "", do: {:cont, []}, else: {:cont, [acc], ""} 
            end)

(Obs: O código acima omite detalhes de implementação da busca do último newline por brevidade; consulte a documentação do ExAws para um exemplo completo.)

Com isso, conseguimos iterar por cada linha do arquivo S3 sem jamais carregar o arquivo inteiro na memória. Apenas um chunk (1MB por padrão, ajustável) e uma linha por vez ficam em RAM. Essa abordagem nos salvou de estourar os limites do pod Kubernetes onde o job rodava – lembre-se, em ambientes K8s geralmente definimos limites de memória, e um processo que tenta alocar vários GB de uma vez será morto pelo OOM Kill.

Dica: Caso seus arquivos no S3 estejam comprimidos (por exemplo .gz), você pode incorporar uma etapa de descompressão on-the-fly no stream. Por exemplo, adicionando |> StreamGzip.gunzip() (da biblioteca :stream_gzip) logo após o ExAws.stream!, conseguimos baixar menos dados (comprimidos) e descompactar em fluxo, sem usar espaço em disco. Isso é extremamente útil para arquivos de migração, que muitas vezes são textos repetitivos e comprimem bem.

Alternativa: stream via arquivo local ou HTTP

A solução acima utiliza as capacidades do ExAws para streaming em memória. Em alguns casos, você pode preferir ou precisar de abordagens alternativas:

• Download para arquivo local + File.stream!: O ExAws.S3.download_file/4 também aceita um path de arquivo local. Podemos baixar o arquivo inteiro diretamente para, digamos, "/tmp/arquivo.sql", sem carregá-lo em memória, já que o download é feito em partes e salvo em disco. Depois, usamos File.stream!("/tmp/arquivo.sql", [], 8192) para ler do disco linha a linha. O lado bom: delega ao sistema de arquivos, que geralmente lida bem com arquivos grandes. O lado ruim: requer espaço em disco (no K8s, talvez um volume temporário), e é uma etapa a mais (download completo antes de processar). Ainda assim, é uma alternativa simples se streaming puro estiver complicado.

• Streaming via HTTP (presigned URL): Caso você não esteja usando Elixir/ExAws, ou queira uma solução independente, outra opção é gerar uma URL temporária (presigned) para o arquivo no S3 e usar um cliente HTTP que suporte streaming de resposta. Por exemplo, em Elixir poderíamos usar o Mint ou HTTPoison para fazer um GET na URL e ir processando o corpo do HTTP em pedaços conforme chega. A própria abordagem discutida acima com HTTPStream no Poeticoding 1 e Poeticoding 2 seguem esse caminho – obtém-se a URL assinada e depois usa um stream HTTP para ler e transformar os dados linha a linha, tudo de forma lazy. Essa técnica independe do ExAws e poderia ser usada em qualquer linguagem com HTTP streaming (em Python requests com stream, Node streams, etc). O importante, de novo, é não montar tudo em memória.

Em suma, seja via biblioteca de cloud (ExAws) ou via HTTP, prefira streams a cargas completas. Assim garantimos uso mínimo de memória e melhoramos até o throughput (pois começamos a processar as primeiras linhas enquanto o restante ainda está baixando, pipelineando as etapas).

Estratégia 2: Execução Sequencial e Controle de Transações

Como mencionado, a stack Ecto/Postgrex tem uma restrição importante: não podemos mandar múltiplas instruções SQL de uma só vez (separadas por ponto e vírgula) num mesmo comando via Repo.query. Se tentarmos juntar sql1; sql2; sql3 em uma única string e chamar Repo.query, receberemos um erro de sintaxe do Postgrex dizendo que não suporta múltiplos comandos no protocolo extendido. Portanto, devemos alimentar nosso banco com um comando por vez. No contexto do Elixir, isso significa que nosso stream de linhas SQL será processado elemento a elemento, cada linha resultando em uma chamada Repo.query(sql) independente.

Isso garante também que a ordem é respeitada: lemos a primeira linha do arquivo, executamos no banco; depois a segunda, executa; e assim por diante. Se a ordem das operações importa (e geralmente importa, especialmente em migrações), estaremos seguros.

Uma implementação simples em Elixir poderia ser:

Stream.with_index(stream_lines, 1)  # junta cada linha com seu número (1,2,3,...)
|> Stream.each(fn {sql_line, line_num} ->
     # Limpando eventuais espaços e ';' do final da linha
     sql = sql_line |> String.trim() |> String.trim_trailing(";")
     case Repo.query(sql, [], log: false) do
       {:ok, _res} -> 
         :ok  # sucesso, não faz nada especial
       {:error, %Postgrex.Error{} = err} ->
         handle_error(line_num, sql, err)  # vamos tratar erros adiante
     end
   end)
|> Stream.run()

No snippet acima, algumas coisas a notar:

• Utilizamos Repo.query/3 ao invés de funções como Repo.query! ou macros de Ecto. O Repo.query nos permite executar SQL “cru” diretamente. Retorna {:ok, result} ou {:error, error} em vez de lançar exceção, o que nos dá controle de tratamento de erro sem parar o fluxo. Passamos log: false para suprimir o log padrão (vamos falar mais sobre logging adiante).

• Removemos ";" do final da linha, se houver, para evitar mandar um ponto-e-vírgula extra que o driver poderia interpretar como comando vazio após o ;.

• A ordem é garantida pelo Stream.each – um comando só inicia depois do anterior finalizar, já que estamos dentro do mesmo processo consumindo sequencialmente

• Reutilização de conexão: Por padrão, cada chamada Repo.query pega uma conexão do pool e devolve após uso. No caso de executarmos 100 mil queries sequencialmente, isso significa pegar e soltar a conexão 100 mil vezes. O overhead disso é relativamente pequeno (afinal cada query em si provavelmente é mais custosa que pegar a conexão), mas podemos otimizar mantendo a mesma conexão para todo o fluxo. Em Elixir, uma forma seria envolver esse processamento dentro de um Repo.checkout (ou DBConnection.run), que “prende” uma conexão do pool para uso exclusivo dentro do bloco fornecido. Assim, evitamos handshake/checkout repetido. Cuidado: não confunda isso com usar uma transação; no Repo.checkout podemos executar comandos independentes com autocommit, mas todos usando o mesmo canal de conexão. Isso pode melhorar um tiquinho a performance e garantir que não teremos espera por conexão do pool no meio.

Exemplo utilizando Repo.checkout:

Repo.checkout(fn ->
  Enum.each(stream_lines, fn sql ->
    case Ecto.Adapters.SQL.query(Repo, sql, [], log: false) do
      {:ok, _} -> :ok
      {:error, err} -> handle_error(..., err)
    end
  end)
end)

No fundo, essa abordagem ou a anterior acabam fazendo a mesma coisa (conexões do Ecto geralmente são persistentes e rápidas). Em muitos casos, a simplicidade de usar apenas o stream com Repo.query é suficiente.

Tamanho da transação: tudo de uma vez ou aos poucos?

Uma pergunta importante ao executar migrações é: devemos envolver todas as operações em uma única transação?. Se tivermos 100 mil comandos de update, podemos pensar: “coloca tudo dentro de Repo.transaction(fn -> ... end) e se algo der errado fazemos rollback geral”. Na prática, isso raramente é uma boa ideia.

Por quê? Uma transação enorme mantém locks e registros no log de transação do banco por muito tempo, consumindo recursos. Se algo falhar no meio, todo o progresso até então é perdido ao dar rollback – o que pode desperdiçar horas de trabalho por causa de um único registro problemático. Além disso, muitos bancos (PostgreSQL incluso) podem ter desempenho pior dentro de uma transação longa, comparado a operações com auto-commit, devido a gerenciamento de MVCC e inchaço de tuplas não confirmadas.

No caso real que enfrentamos, optamos por não ter uma transação englobando tudo. Cada comando se commitava individualmente. Assim, mesmo que um comando ou outro falhasse, as mudanças anteriores já estariam persistidas e não seriam desfeitas. Isso se alinha com a filosofia de continuar apesar de erros.

"E se precisarmos de atomicidade total?" – pode perguntar alguém. Ou seja, ou migra tudo ou nada. Bem, aí é um requisito de negócio bem diferente: em geral, para grandes migrações, costuma-se planejar de forma que possamos reprocessar ou corrigir depois os poucos erros, em vez de exigir atomcidade absoluta (até porque atingir isso pode ser quase impossível sem causar downtime enorme). Mas caso precisasse, uma alternativa seria agrupar em transações menores: por exemplo, comitar a cada 1000 comandos. Assim, a cada mil operações bem-sucedidas, garantimos aquele lote no banco; se uma falhar dentro do lote, podemos rolar só aquele sub-lote. Essa abordagem de chunking transacional reduz a janela de rollback. Contudo, fica a complexidade: como lidar com um erro no meio de um lote? Repetir só ele isoladamente depois? Pode virar um pesadelo gerenciar isso manualmente.

No nosso caso, decidimos: commit por comando = simplicidade e menor risco. O preço a pagar pode ser desempenho (por causa de fsync a cada operação no banco), mas vamos tratar de otimizações mais à frente.

Controlando timeouts e encerramentos

Executar 100k queries sequenciais também exige cuidado com configurações de timeout do Ecto/DB:

• Timeout de query: Por padrão, cada query no Ecto tem timeout ~15 segundos. 99% das nossas operações eram rápidas (ms ou poucos s), mas e se alguma demorar mais (ex: index pesado, ou atualizando milhões de linhas numa tacada)? Para não abortar prematuramente, podemos ajustar o timeout na chamada Repo.query(sql, [], timeout: 120_000) por exemplo, para dar 2 minutos naquele comando específico, ou mesmo :infinity se for algo known slow. Use com parcimônia – timeouts existem para evitar conexões travadas eternamente.

• Mantenha vivo o processo: Se rodamos isso dentro de um job no K8s, certifique-se que o pod aguenta rodar pelo tempo necessário. Os timeouts do Kubernetes (como activeDeadlineSeconds num Job) devem ser configurados conforme a expectativa de duração da migração. E dentro da aplicação Elixir, se estiver sob um Supervisor, talvez aumentar temporariamente o :timeout do GenServer/Task supervisor se necessário, para não ser reiniciado erroneamente.

• Finalização graciosa: Se estamos tratando erros e continuando, idealmente o job deveria terminar com sucesso (código 0) mesmo que tenha havido algumas falhas não-críticas, contanto que fizemos tudo que dava. Podemos sinalizar sucesso, mas garantir que os detalhes dessas falhas estejam logados para acompanhamento. Se preferir marcar o job como “falhou se qualquer erro ocorreu”, também é possível, mas aí perde-se a vantagem de seguir em frente – vai ter que reexecutar tudo.

Agora que já lidamos com leitura e execução básica, vamos falar de otimizações para desempenho e do tratamento robusto de erros.

Estratégia 3: Otimizando Performance com Chunking e Batching

Performance não deve vir antes de corretude em uma migração, mas uma vez assegurados os pilares acima, podemos sim buscar maneiras de acelerar o processo para que a janela de migração seja a menor possível. Duas ideias relacionadas são: chunking (processamento em pedaços) e batching (agrupamento de operações).

Leitura em pedaços e paralelismo de download

Já cobrimos que o ExAws por padrão baixa com 8 chunks simultâneos de 1MB. Isso é bom para throughput, mas se estivermos realmente restritos em memória, poderíamos reduzir a concorrência (exAws permite configurar max_concurrency: 1 e até o tamanho do chunk). Isso deixaria o download totalmente sequencial (1MB por vez) e limitaria o uso de RAM a ~1MB. Em contrapartida, o download total seria um pouco mais lento. É um trade-off: para pods muito limitados talvez seja necessário, mas na maioria dos casos 8 MB de uso (8 threads x 1MB) é aceitável. Monitoramos e vimos que a utilização de memória ficou dentro do orçamento, então mantivemos o padrão que oferecia melhor velocidade.

Agrupar múltiplos inserts/updates em um só comando

Uma otimização poderosa, quando aplicável, é agrupar várias operações similares em um único comando SQL. Por exemplo, se o nosso arquivo SQL fosse composto majoritariamente de milhares de INSERT na mesma tabela, poderíamos combinar vários em um único INSERT ... VALUES (...), (...), ... gigante, reduzindo drasticamente o número de comandos executados. O Ecto oferece a função Repo.insert_all/3 exatamente para inserir múltiplos registros de uma vez.

Há relatos na comunidade de que é muito mais eficiente executar uma query com 10.000 inserts de uma vez do que 10.000 queries separadas. O motivo é claro: latência de rede reduzida (uma ida ao banco ao invés de 10 mil), overhead de parsing do SQL uma vez só, etc. Se você consegue juntar N operações num lote, faça-o.

No nosso caso específico, as operações não eram triviais de agrupar, pois envolviam updates de registros existentes espalhados em tabelas diferentes. Mas se, por exemplo, você está migrando dados de uma tabela antiga para uma nova, poderia ler em chunks de 1000 registros e usar um insert_all na tabela nova para cada chunk, ao invés de inserir um por um.

Com Elixir Streams, podemos facilmente fazer algo como:

stream_data
|> Stream.chunk_every(1000)
|> Stream.each(fn batch ->
     Repo.insert_all(MySchema, batch, on_conflict: :replace_all, conflict_target: ...) 
   end)
|> Stream.run()

No pseudo-código acima, batch seria uma lista de maps ou structes Ecto a inserir. O on_conflict só ilustram que poderíamos até lidar com conflitos se precisarmos atualizar existentes.

Atenção: batching funciona melhor para inserções ou operações homogêneas. Para updates complexos (diferentes tabelas, diferentes WHERE para cada linha), não há um agrupamento SQL nativo fácil. Aí, de fato, seguimos com um por um. Nunca tente simplesmente concatenar vários comandos com ; achando que vai burlar o sistema. Cada statement tem que ser enviado separadamente, a não ser que você opte por usar o protocolo simples do Postgres (não suportado pelo Ecto) ou outra ferramenta externa (por exemplo, rodar um script .sql direto via client psql). No contexto de nossa aplicação Elixir, mantivemos a granularidade de um a um, abrindo mão de agrupar updates.

Chunking de transações e conexões

Mencionamos anteriormente a possibilidade de chunking transacional para melhorar performance: por exemplo, abrir uma transação a cada 1000 operações para diminuir o overhead de commit a cada linha. Optamos por não fazer isso, em parte por simplicidade e em parte porque nosso bottleneck não foi o commit em si, mas sim o tempo de execução total no banco (que envolvia consultas a índices, etc.). No seu caso, se identificar que o flush de disco a cada comando está lentificando (pode acontecer se cada update é muito pequeno e o disco é lento), agrupar alguns em transação pode melhorar. Mas lembre-se do risco: se um falha, aquele bloco inteiro volta. É um equilíbrio delicado.

Um membro da comunidade do Elixir inclusive reforçou que operações em lote não combinam bem com uma transação gigantesca. Ou seja, de nada adianta você juntar inserts em lotes de 10k se depois envolver tudo num transaçãozão de 1 milhão – você perde os ganhos e adiciona riscos. Melhor aplicar lotes + autocommit, ou transações pequenas.

E quanto a paralelizar as operações?

Até agora, assumimos execução estritamente sequencial, o que é normalmente necessário se há dependência de ordem ou se estamos atualizando o mesmo conjunto de dados (para evitar deadlocks e conflitos). Porém, em algumas migrações, dá para paralelizar desde que em partições independentes. Por exemplo, se você tem 10 bancos de dados diferentes para 10 serviços, poderia rodar 10 jobs simultâneos, um por banco, cada um fazendo sua parte. Dentro de um mesmo banco, se as operações forem em tabelas completamente separadas, também poderia rodar threads separadas (mas cuidado extremo com carga no banco!).

De forma geral, preferimos paralelizar as operações dentro de um mesmo banco de dados, mas com controle individual por banco. Cada banco possui uma configuração de máquina específica, com diferentes capacidades de carga, e está vinculado a uma estratégia própria de migração. Todo esse processo foi cuidadosamente controlado e parametrizado de acordo com o cenário.

Avalie no seu contexto: se paralelismo trouxer ganhos e não violar consistência, Elixir torna fácil utilizar Tasks assíncronas ou mesmo a biblioteca Flow/GenStage para processamento paralelo com controle de concorrência. Só tenha em mente a capacidade do banco de dados em aguentar múltiplas operações simultâneas.

Estratégia 4: Tratamento de Erros e Log Consolidado

Chegamos em um ponto crucial: como lidar com erros durante a migração? E, correlacionado, como organizar a saída de logs de forma útil.

Capturando erros sem quebrar o fluxo

A chave para uma migração resiliente é isolar falhas. No pseudo-código que mostramos, usamos Repo.query (não !), para obter um {:error, err} em vez de uma exceção. Assim, nenhum erro de SQL levanta uma exceção não tratada que interromperia todo o processamento. Nós capturamos o erro, logamos, e seguimos.

Exemplos de erros que podem ocorrer em uma linha SQL:

• Violação de integridade (ex: FOREIGN KEY, UNIQUE).
• Erro de sintaxe no comando (às vezes scripts gerados podem ter alguma linha malformada).
• Deadlock detectado no banco (código SQLSTATE 40P01 no Postgres).
• Timeout ou perda de conexão momentânea.

Para muitos desses, repetir a operação não vai adiantar (ex.: sintaxe errada vai continuar errada). Mas para alguns transientes vale a pena tentar novamente. Implementamos uma lógica simples de retry para certos casos:

• Se o erro for de deadlock (err.postgres.code == "40P01"), podemos aguardar um instante e tentar de novo aquela query, um número limitado de vezes. Deadlocks são meio aleatórios e um retry simples pode resolver.

• Se for erro de conexão (por exemplo, DBConnection.ConnectionError indicando que a conexão caiu), o pool do Ecto geralmente já tentará reconectar na próxima query automaticamente. Então, talvez apenas registrar e tentar novamente seja suficiente.

• Se for qualquer outro erro persistente, registramos e não repetimos.

A implementação pode ser algo como:

def exec_sql_with_retry(sql, attempt \\ 1) do
  case Repo.query(sql, [], timeout: 30_000, log: false) do
    {:ok, res} -> :ok
    {:error, %Postgrex.Error{postgres: %{code: code}} = err} 
         when code in ["40P01", "40001"] ->  # 40001 seria serialization failure (ex.: retry do Postgres)
      if attempt <= @max_retries do
        Process.sleep(100 + 50 * attempt)  # espera incremental
        exec_sql_with_retry(sql, attempt + 1)
      else
        log_error(err)
      end
    {:error, err} ->
      log_error(err)
  end
end

No nosso caso específico, a incidência de erros foi baixíssima, então não complicamos muito com retries – registramos e pronto. Mas fica a ideia para quem for implementar.

Logando erros em um só lugar

Em vez de imprimir cada erro na saída padrão (o que iria misturá-los no meio do log geral da aplicação ou do Kubernetes), optamos por consolidar todos os erros em um arquivo de log separado, por arquivo de entrada processado.

Ou seja, se estamos processando arquivo.sql, criamos arquivo.sql.errors.log e vamos adicionando entradas lá dentro sempre que ocorre um erro. Cada entrada inclui pelo menos o número da linha e uma mensagem do erro retornado pelo banco, algo como:

Linha 12345: erro ao executar "UPDATE ...": ERROR: duplicate key value violates unique constraint "idx_user_email"
Linha 67890: erro ao executar "INSERT ...": ERROR: null value in column "id" violates not-null constraint

Isso nos permitiu, ao final, ter um resumo de tudo que falhou em um só lugar, fácil de compartilhar com o time ou analisar depois. Enquanto isso, o job principal podia terminar com sucesso sem poluir os logs de execução normais com milhares de linhas (que dificultariam enxergar o progresso).

Implementação em Elixir usando File:

File.open!("logs/arquivo_sql_erro.log", [:write, :utf8], fn file ->
  Stream.with_index(stream_lines, 1)
  |> Stream.each(fn {sql, line_num} ->
       case Repo.query(String.trim_trailing(sql, ";"), []) do
         {:ok, _} -> :ok
         {:error, err} ->
           IO.puts(file, "Linha #{line_num}: #{err.message}")
       end
     end)
  |> Stream.run()
end)

No snippet, abrimos o arquivo e dentro da função passamos o stream. Cada erro faz um IO.puts no arquivo. Estamos no mesmo processo do stream, então não há condição de corrida no arquivo (se fossem paralelos, teria que sincronizar a escrita). Esse arquivo depois poderia ser anexado como artefato, enviado ao S3, ou lido por alguma rotina de notificação.

Nota: preferimos acumular erros em memória e depois gravar tudo de uma vez? Poderíamos ter usado um Enum.reduce no stream para construir uma lista de erros. Mas imagine se 10 mil linhas falham – segurar 10 mil strings de erro em memória também pesa. Gravar em arquivo conforme ocorrem foi uma abordagem mais simples também, e garante que mesmo que o job pare no meio, os erros até ali já estarão no arquivo parcial.

Silenciando logs verbosos

O Ecto por padrão loga cada query executada (no nível debug do Logger). Se não fizermos nada, rodar 100 mil comandos vai gerar 100 mil linhas de log do tipo "QUERY OK ... etc". Isso é totalmente indesejado aqui. Por isso usamos log: false em cada Repo.query – assim, nenhum log é emitido para queries bem-sucedidas. Apenas nossos próprios logs de erro (ou logs de progresso que quisemos inserir manualmente) aparecem. Essa é uma flag simples e poderosa para controlar isso.

Como resultado, os logs do sistema ficaram limpos, contendo basicamente: inícios e fins de processamento de cada arquivo, um contador a cada X linhas (colocamos um log info a cada 1000 mil operações para saber que estava avançando) e quanto de memória o processo esta consumindo. Todo o resto – detalhes de falhas – ficou confinado ao arquivo de erro específico.

Monitorando progresso e métricas

Em migrações longas, é bom ter noção do andamento. Implementamos, como dito, um log de progresso a cada bloco de linhas processadas (ex: "10k linhas processadas...", "20k..."). Isso ajuda a estimar tempo restante e verificar que não está travado. Também é possível coletar métricas, como tempo total, contagem de sucessos vs erros, etc., e expor no final. Essas informações podem ir para um relatório final ou mesmo enviados a um sistema de monitoramento.

Considerações para múltiplos serviços

No nosso cenário hipotético, precisamos mudar dados em vários bancos de diferentes serviços simultaneamente. Como coordenar isso? Existem algumas abordagens possíveis:

• Job distribuído: Colocar um job de migração em cada serviço (por exemplo, uma migration no estilo ActiveRecord/Ecto própria de cada microserviço) e acioná-los todos juntos (manualmente ou via orquestração). Cada um cuidaria do seu banco local. A coordenação aqui é mais humana: garantir que todos rodem na mesma janela de manutenção, talvez colocando os serviços em modo de espera enquanto roda, se necessário.

• Orquestrador central: Criar um serviço ou script mestre que conecta em todos os bancos necessários (ou chama APIs dos serviços) para aplicar as mudanças. Isso centraliza em um ponto, mas pode ser mais complexo de implementar (precisa de credenciais de todos, etc) e um ponto de falha único.

• Feature Flags e migração gradual: Dependendo da mudança, às vezes podemos introduzir o novo formato de dado em paralelo ao antigo, fazer os sistemas aceitarem ambos temporariamente (feature flag), migrar dados gradualmente, e depois desligar o formato antigo. Essa é a estratégia blue-green ou expand-contract. Contudo, em exigências regulatórias rígidas de “trocar tudo até data X”, talvez não seja possível conviver com dois formatos; de qualquer forma, é sempre bom avaliar se uma migração big bang é realmente necessária ou se dá pra fazer incrementalmente.

No caso de precisar do “ao mesmo tempo”, uma prática é agendar uma janela de manutenção onde todos os serviços ficam indisponíveis ou em modo read-only enquanto a migração roda. Assim, não há problema se um terminar antes do outro, pois nada está processando dados inconsistentes nesse meio tempo. Claro, downtime coordenado é algo a se evitar em geral, mas às vezes é a opção mais simples para eliminar riscos.

Se optar por downtime zero, aí a coisa fica mais complexa: tem que garantir que cada passo seja compatível com o sistema rodando. Por exemplo, você poderia:

1. Adicionar a nova coluna/identificador em todas bases (passo preparatório).
2. Popular a nova coluna com base nos dados antigos (talvez com scripts de migração como descrito).
3. Alterar os serviços para usar a nova coluna a partir de um timestamp.
4. Remover a coluna antiga depois de verificar que tudo está usando a nova.

Essa estratégia de expandir e contrair o esquema minimiza indisponibilidade, mas requer planejar a aplicação também.

No nosso cenário, não entraremos nos detalhes de negócio, mas vale destacar que utilizamos um orquestrador para conduzir o processamento e aplicar os scripts de migração em paralelo nos múltiplos bancos — tudo isso sem qualquer downtime. Foi como trocar a turbina com o avião em pleno voo, em uma operação extremamente coordenada e controlada.

Considerações Finais e Dicas

Reunindo os pontos principais que aprendemos com essa migração de dados em larga escala, aqui vai um pequeno guia de melhores práticas:

• Planejamento é tudo: Entenda exatamente quais sistemas e dados serão afetados. Desenhe o plano completo, incluindo ordem das operações, dependências, plano de rollback (o que fazer se algo der muito errado), e teste em ambiente de staging com uma cópia dos dados se possível.

• Use streams e processamento incremental: Evite ler ou carregar grande volumes de uma vez. Em Elixir, aproveite do Stream para ler arquivos ou consultas aos poucos, em Python use geradores, em Java use iteradores ou streams, etc. Isso torna seu processo mais leve e previsível.

• Conheça suas ferramentas: No nosso caso, saber das limitações do Ecto/Postgrex (um statement por vez) e funcionalidades úteis como Repo.insert_all, log: false e controle de timeouts fez a diferença. Seja qual for sua linguagem, pesquise se há suporte a bulk operations, qual o comportamento de transações grandes, etc.

• Não abuse das transações: A menos que estritamente necessário, não coloque terabytes de operações numa única transação. Prefira checkpointar (se é que esse termo existe) progresso (commits parciais) para não perder tudo em falhas pontuais.

• Logs e comunicação: Uma migração desse porte deve ter logs claros. Consolidar erros em um arquivo separado provou ser uma boa decisão no nosso caso – facilitou compartilhar os problemas específicos com o time para corrigir aqueles registros manualmente depois, por exemplo. Também mantenha stakeholders informados do progresso (um simples “já foram processados X% dos registros” a cada tanto tempo acalma os ânimos durante uma manutenção prolongada).

• Teste de performance e tuning: Se possível, rode um subset da migração (ex: 10 mil operações) para medir o tempo e extrapolar, identificando gargalos. Ajuste parâmetros como chunk_size do download, tamanho de pool de conexão, paralelismo, etc., conforme os recursos disponíveis. No Kubernetes, assegure-se de solicitar CPU/memória adequadas para o job de migração – ele provavelmente vai usar bem mais recursos do que sua aplicação no dia-a-dia.

• Ferramentas específicas de migração: Às vezes, vale avaliar ferramentas desenhadas para ETL ou migração, como Apache Beam, Spark, etc., especialmente se for transformar dados. No nosso caso, como era basicamente SQL direto no banco, o Elixir deu conta de orquestrar bem. Mas se fosse algo como migrar dados entre bancos diferentes, talvez um pipeline de dados dedicado fosse mais adequado.

• Elixir como aliado: De modo geral, Elixir se mostrou uma ótima escolha para essa tarefa pela sua facilidade em lidar com IO e streams, pelas primitivas de concorrência (que nos deram a opção de paralelizar caso quiséssemos, sem complicação), e pela robustez do ecosistema (ExAws para S3, Ecto para DB, etc.). Outros ecossistemas também conseguem, mas apreciar essas vantagens ajuda a valorizar a ferramenta certa para o job certo.

Conclusão

Grandes migrações de dados não precisam ser um bicho de sete cabeças. Com as técnicas certas, é possível processar milhões de registros de forma eficiente, segura e sem estresse desnecessário. Recapitulando o que vimos: primeiro, opte por leitura e processamento streaming para manter os recursos sob controle (memória constante, sem sobrecarga); segundo, garanta execução ordenada e atomicidade na medida certa – nem de menos (não perder consistência), nem demais (não travar tudo numa mega transação); terceiro, aplique otimizações de desempenho viáveis, agrupando operações quando possível e evitando repetir trabalho desnecessário; quarto, trate erros como cidadãos de primeira classe – não deixe uma exceção interromper sua migração no meio, capture, logue e siga adiante sempre que fizer sentido; e por último, planeje a coordenação entre sistemas se for um cenário distribuído, seja através de sincronização manual ou automações de orquestração.

Ao seguir essas práticas, no caso real obtivemos sucesso na migração: todos os sistemas passaram a usar a nova referência de titular sem downtime e sem incidentes graves. Os poucos ajustes manuais necessários (devido a erros registrados) puderam ser feitos rapidamente depois, graças ao nosso log consolidado. E, talvez mais importante, ganhamos confiança para enfrentar a próxima migração que surgir – porque em um ambiente de rápido crescimento e evolução, sempre haverá a próxima!

📚 Referências

📦 Elixir e Ecossistema

1. ExAws.S3 – Documentação Oficial (Hexdocs)
   
   Uso de download_file/4, stream!/2 e estratégias de leitura em chunks.

2. Stream.chunk_while/4 – Elixir (Hexdocs)
   
   Para dividir binários de chunks em linhas.

3. File.stream!/3 – Leitura eficiente de arquivos locais

4. Ecto.Repo.query/3 – Execução de SQL cru
   
   Com uso de log: false para suprimir logs excessivos.

5. Ecto.Adapters.SQL.insert_all/3 – Inserções em lote

6. DBConnection.run/3 – Controle manual de conexões

7. Postgrex – Discussão sobre múltiplos statements
   
   Limitações do driver ao lidar com múltiplos comandos SQL em uma query.

---

🌐 Comunidade e Casos de Uso

1. ElixirForum – insert_all vs múltiplos inserts
   
   Comparação prática entre performance de batches e inserts unitários.

2. Poeticoding – Streaming de arquivos grandes do S3 ou HTTP
   
   Aborda leitura eficiente via S3, streaming HTTP e processamento incremental.

---

🗄️ Banco de Dados – PostgreSQL

1. PostgreSQL SQLSTATE Error Codes Lista oficial de códigos de erro (ex: deadlocks, timeout, violação de chave única).

---

🧰 Infraestrutura e Operações

1. Kubernetes Jobs – Documentação oficial
   
   Execução de jobs batch e configuração de memória/timeout.

2. Logger – Documentação do Elixir Logger
   
   Controle de níveis de log durante execuções longas.

3. StreamGzip – Descompactação em stream no Elixir
   
   Permite leitura de arquivos .gz diretamente em fluxo.