Resolver problemas de jobs de streaming lentos ou travados

Nesta página, explicamos como resolver problemas comuns de jobs de streaming lentos ou travados do Dataflow.

Se você notar os seguintes sintomas, talvez o job de streaming do Dataflow esteja sendo executado lentamente ou esteja travado:

Use as informações das seções a seguir para identificar e diagnosticar o problema.

Identificar a causa raiz

  1. Verifique as métricas de atualização de dados e bytes de backlog.

    • Se as duas métricas estiverem aumentando monotonicamente, isso significa que o pipeline está travado e não está progredindo.
    • Se a atualização de dados estiver aumentando, mas os bytes de backlog permanecerem normais, isso significa que um ou mais itens de trabalho estão travados no pipeline.

    Procure os estágios em que essas métricas estão aumentando para identificar qualquer estágio com problemas e as operações realizadas nele.

  2. Verifique o gráfico de processamento paralelo para conferir se algum estágio está travado devido ao paralelismo excessivo ou insuficiente. Consulte Resolver problemas de paralelismo.

  3. Verifique os registros do job em busca de problemas como limites de cota, problemas de falta de estoque ou esgotamento de endereços IP.

  4. Verifique os registros do worker em busca de avisos e erros.

    • Se os registros do worker contiverem erros, consulte o stack trace. Investigue se o erro é causado por um bug no seu código.
    • Procure erros do Dataflow. Consulte Resolver erros do Dataflow.
    • Procure erros que mostrem que o job excedeu um limite, como o tamanho máximo da mensagem do Pub/Sub.
    • Procure erros de falta de memória, que podem causar um pipeline travado. Se você encontrar erros de falta de memória, siga as etapas em Resolver erros de memória insuficiente do Dataflow.
    • Para identificar uma etapa lenta ou travada, verifique se há mensagens Operation ongoing nos registros do worker. Consulte o stack trace para conferir onde a etapa está gastando tempo. Para mais informações, consulte Processamento travado ou operação em andamento.

  5. Se um item de trabalho estiver travado em um worker específico, reinicie a VM desse worker.

  6. Se você não estiver usando o Streaming Engine, verifique os registros do shuffler em busca de avisos e erros. Se você encontrar um erro de tempo limite de RPC na porta 12345 ou 12346, talvez o job esteja sem uma regra de firewall. Consulte Regras de firewall para o Dataflow.

  7. Verifique se há chaves de uso intenso.

  8. Se o Portable Runner estiver ativado, verifique se há erros nos registros do harness. Para mais informações, consulte Resolver problemas do Portable Runner.

Investigar falhas repetidas

Em um job de streaming, algumas falhas são repetidas indefinidamente. Essas novas tentativas impedem o progresso do pipeline. Para identificar falhas repetidas, verifique se há exceções nos registros do worker.

  • Se a exceção for o código do usuário, depure e corrija o problema do código ou dos dados.
  • Para evitar que falhas inesperadas paralisem o pipeline, implemente uma fila de mensagens inativas. Para conferir um exemplo de implementação, consulte Padrões do BigQuery na documentação do Apache Beam.
  • Se a exceção for um erro de memória insuficiente (OOM, na sigla em inglês), consulte Como resolver erros de memória insuficiente do Dataflow.
  • Para outras exceções, consulte Resolver erros do Dataflow.

Identificar workers não íntegros

Se os workers que processam o job de streaming não estiverem íntegros, o job pode ficar lento ou parecer travado. Para identificar workers não íntegros:

Identificar stragglers

Um straggler é um item de trabalho lento em relação a outros itens de trabalho da fase. Para informações sobre como identificar e corrigir stragglers, consulte Resolver problemas de stragglers em jobs de streaming.

Resolver problemas de paralelismo

Para escalonabilidade e eficiência, o Dataflow executa as fases do pipeline em paralelo em vários workers. A menor unidade de processamento paralelo do Dataflow é uma chave. As mensagens recebidas para cada fase combinada são associadas a uma chave. A chave é definida de uma das seguintes maneiras:

  • A chave é definida implicitamente pelas propriedades da origem, como partições do Kafka.
  • A chave é definida explicitamente pela lógica de agregação no pipeline, como GroupByKey.

No Dataflow, as linhas de execução do worker são responsáveis por processar pacotes de trabalho (mensagens) para uma chave. O número de linhas de execução disponíveis para processar as chaves do job é igual a num_of_workers * threads_per_worker. A contagem de linhas de execução por worker é determinada com base no SDK (Java, Python ou Go) e no tipo de serviço (lote ou streaming).

Se o pipeline não tiver chaves suficientes para uma determinada fase, ele vai limitar o processamento paralelo. Essa fase pode se tornar um gargalo.

Se o pipeline usar um número muito grande de chaves para um determinado estágio, ele poderá limitar a capacidade de processamento do estágio e acumular backlog nos estágios upstream, porque há alguma sobrecarga por chave. A sobrecarga pode incluir o back-end se comunicando com workers, RPCs externos para um coletor, como o BigQuery, e outros processamentos. Por exemplo, se o processamento de uma chave com uma mensagem levar 100 ms, também poderá levar cerca de 100 ms para processar 1.000 mensagens nesse pacote de chaves.

Identificar fases com baixo paralelismo

Para identificar se a lentidão do pipeline é causada por baixo paralelismo, consulte as métricas de utilização da CPU. Se a CPU estiver baixa, mas distribuída de maneira uniforme entre os workers, o job pode ter paralelismo insuficiente. Se o job estiver usando o Streaming Engine, consulte as métricas de paralelismo na guia Métricas do job para verificar se uma fase tem baixo paralelismo. Para minimizar esse problema:

Identificar fases com alto paralelismo

Uma combinação de baixa latência do sistema, atualização crescente de dados e aumento do backlog e CPUs de worker subutilizadas sugere que o pipeline está sendo limitado devido a um grande número de chaves. Verifique o gráfico de processamento paralelo para identificar fases com um grande número de chaves.

Transformações como Reshuffle podem gerar milhões de chaves se você não especificar explicitamente withNumBuckets. Grandes números de chaves podem levar à criação de vários pacotes de trabalho menores, cada um dos quais exige uma linha de execução de worker dedicada para processamento. Como as linhas de execução de worker disponíveis são limitadas, isso pode levar a um backlog significativo de chaves de processamento, causando atrasos à medida que elas aguardam recursos. Como resultado, as linhas de execução do worker não são usadas com eficiência.

Recomendamos limitar o número de chaves definindo a opção withNumBuckets na transformação Reshuffle. O valor não pode exceder o número total de linhas de execução em todos os workers. A segmentação de chaves (threads_per_worker * max_workers) no pipeline pode não ser ideal. Às vezes, menos chaves e pacotes maiores são possíveis e são processados com mais eficiência pelo Dataflow devido ao uso de menos workers. Um número menor de chaves cria pacotes de trabalho maiores, que usam com eficiência as linhas de execução do worker e aumentam a capacidade de processamento da fase.

Se houver várias etapas Reshuffle no pipeline, divida o número total de linhas de execução pela contagem de etapas Reshuffle para calcular withNumBuckets.

Verificar se há chaves de uso intenso

Se as tarefas forem distribuídas de maneira desigual entre os workers e a utilização deles for muito desigual, talvez o pipeline tenha uma chave de uso intenso. Uma chave de uso intenso é uma chave que está super-representada nos dados.

Verifique se o pipeline tem chaves de uso intenso observando o status do gargalo na etapa lenta. Você pode ter chaves de uso intenso se o motivo do gargalo for "Chaves de uso intenso ou paralelismo de chaves insuficiente". Para informações gerais sobre gargalos, consulte Resolver problemas de gargalos.

Por padrão, o Dataflow registra apenas a presença de uma chave de uso intenso, não o valor da chave específica. Para imprimir a chave legível para os registros quando uma chave de uso intenso é detectada, defina a seguinte opção de pipeline de depuração ao executar o pipeline:

Java

--hotKeyLoggingEnabled=true

Python

--dataflow_service_options=enable_hot_key_logging

Se você estiver executando um job de modelo flexível, consulte Especificar opções de pipeline para mais informações sobre como usar --additional-pipeline-options para transmitir a opção.

Em seguida, verifique se há chaves de uso intenso usando o seguinte filtro de registro:

  resource.type="dataflow_step"
  resource.labels.job_id=JOB_ID
  jsonPayload.logger="org.apache.beam.runners.dataflow.worker.HotKeyLogger"

Substitua JOB_ID pelo ID do job.

Também é possível configurar alertas baseados em registros para notificar você quando chaves de uso intenso forem detectadas no pipeline. Use o filtro de amostra como ponto de partida.

Para resolver esse problema, siga uma ou mais das seguintes etapas:

  • Faça o rechaveamento dos dados. Para gerar novos pares de chave-valor, aplique uma transformação ParDo. Para mais informações, consulte a página de transformação ParDo do Java ou a página de transformação ParDo do Python na documentação do Apache Beam.
  • Use .withFanout nas transformações de combinação. Para mais informações, consulte a Combine.PerKey classe no SDK do Java ou a with_hot_key_fanout operação no SDK do Python.
  • Se você tiver um pipeline Java que processa PCollections ilimitados de alto volume, recomendamos fazer o seguinte:
    • Use Combine.Globally.withFanout, em vez de Combine.Globally.
    • Use Combine.PerKey.withHotKeyFanout, em vez de Count.PerKey.

Verificar se há cota insuficiente

Verifique se há cota suficiente para a origem e o coletor. Por exemplo, se o pipeline lê a entrada do Pub/Sub ou do BigQuery, é possível que seu Google Cloud projeto tenha uma cota insuficiente. Para mais informações sobre limites de cota para esses serviços, consulte Cota do Pub/Sub ou Cota do BigQuery.

Se o job estiver gerando um grande número de erros 429 (Rate Limit Exceeded), é possível que ele tenha uma cota insuficiente. Para verificar se há erros, tente as seguintes etapas:

  1. Acesse o Google Cloud console.
  2. No painel de navegação, clique em APIs e serviços.
  3. No menu, clique em Biblioteca.
  4. Use a caixa de pesquisa para procurar Pub/Sub.
  5. Clique em API Cloud Pub/Sub.
  6. Selecione Gerenciar.
  7. No gráfico Tráfego por código de resposta, procure códigos de erro de cliente (4xx).

Também é possível usar o Metrics Explorer para verificar o uso da cota. Se o pipeline usar uma origem ou um coletor do BigQuery, use as métricas da API BigQuery Storage para resolver problemas de cota. Por exemplo, para criar um gráfico que mostre a contagem de conexões simultâneas do BigQuery, siga estas etapas:

  1. No Google Cloud consoleno Google Cloud, selecione Monitoring:

    Acessar Monitoring

  2. No painel de navegação, selecione Metrics Explorer.

  3. No painel Selecionar uma métrica, em Métrica, filtre Projeto do BigQuery > Gravar > Contagem de conexões simultâneas.

Para instruções sobre como visualizar métricas do Pub/Sub, consulte Monitorar o uso da cota em "Monitorar o Pub/Sub no Cloud Monitoring". Para instruções sobre como conferir métricas do BigQuery, consulte Visualizar o uso e os limites de cota em "Criar painéis, gráficos e alertas".

Ferramentas de depuração

Quando você tem um pipeline lento ou travado, as ferramentas a seguir podem ajudar a diagnosticar o problema.

  • Para correlacionar incidentes e identificar gargalos, use o Cloud Monitoring para Dataflow.
  • Para monitorar o desempenho do pipeline, use Cloud Profiler.
  • Algumas transformações são mais adequadas para pipelines de alto volume do que outras. As mensagens de registro podem identificar uma transformação de usuário travada em pipelines em lote ou de streaming.
  • Para saber mais sobre um job travado, use as métricas do job do Dataflow. A lista a seguir tem métricas úteis:
    • A métrica Bytes de backlog (backlog_bytes) mede a quantidade de entrada não processada em bytes por fase. Use essa métrica para encontrar uma etapa combinada que não tenha capacidade de processamento. Da mesma forma, a métrica de elementos do backlog (backlog_elements) mede o número de elementos de entrada não processados de uma fase.
    • A métrica Chaves de paralelismo de processamento (processing_parallelism_keys) mede o número de chaves de processamento paralelo para uma fase específica do pipeline nos últimos cinco minutos. Use essa métrica para investigar das seguintes maneiras:
      • Restrinja o problema a fases específicos e confirme os avisos de teclas de atalho, como A hot key ... was detected.
      • Encontre gargalos de capacidade de processamento causados por paralelismo insuficiente. Esses gargalos podem resultar em pipelines lentos ou travados.
    • A métrica de atraso do sistema (system_lag) e a métrica de atraso do sistema por fase (per_stage_system_lag) medem o tempo máximo que um item de dados ficou em processamento ou aguardando processamento. Use essas métricas para identificar fases e gargalos ineficientes das fontes de dados.

Para conferir outras métricas não incluídas na interface da Web de monitoramento do Dataflow, consulte a lista completa de métricas do Dataflow em Google Cloud métricas.