Dicas de ajuste de jobs do Spark

As seções a seguir dão dicas para ajudar você a ajustar seus aplicativos do Spark no Serviço Gerenciado para Apache Spark.

Usar clusters temporários

Ao usar o modelo de cluster "temporário" do Serviço Gerenciado para Apache Spark, você cria um cluster dedicado para cada job e, quando o job é concluído, você exclui o cluster. Com o modelo temporário, é possível tratar o armazenamento e a computação separadamente, salvando os dados de entrada e saída do job no Cloud Storage ou no BigQuery e usando o cluster apenas para computação e armazenamento temporário de dados.

Dificuldades com clusters permanentes

O uso de clusters temporários de um job evita as seguintes armadilhas e possíveis problemas associados ao uso de clusters "persistentes" compartilhados e de longa duração:

  • Pontos únicos de falha: um estado de erro de cluster compartilhado pode fazer com que todos os jobs falhem, bloqueando um pipeline de dados inteiro. Investigar e se recuperar de um erro pode levar horas. Como os clusters temporários mantêm apenas estados temporários no cluster, quando ocorre um erro, eles podem ser excluídos e recriados rapidamente.
  • Dificuldade para manter e migrar estados de cluster no HDFS, MySQL ou sistemas de arquivos locais
  • Conflitos de recursos entre jobs que afetam negativamente os SLOs
  • Daemons de serviço que não respondem causados por pressão de memória
  • Acúmulo de registros e arquivos temporários que podem exceder a capacidade do disco
  • Falha no escalonamento vertical devido à falta de estoque na zona do cluster
  • Falta de suporte para versões desatualizadas de imagens de cluster.

Benefícios do cluster temporário

Por outro lado, os clusters temporários permitem que você faça o seguinte:

  • Configure diferentes permissões do IAM para diferentes jobs com diferentes contas de serviço de VM do Serviço Gerenciado para Apache Spark.
  • Otimizar as configurações de hardware e software de um cluster para cada job, alterando as configurações do cluster conforme necessário.
  • Faça upgrade das versões de imagem em novos clusters para receber os patches de segurança, as correções de bugs e as otimizações mais recentes.
  • Solucione problemas mais rapidamente em um cluster isolado de job único.
  • Economize pagando apenas pelo tempo de execução do cluster temporário, não pelo tempo de inatividade entre jobs em um cluster compartilhado.

Usar o Spark SQL

A API DataFrame do Spaark SQL é uma otimização significativa da API RDD. Se você interagir com o código que usa RDDs, considere ler dados como um DataFrame antes de transmitir um RDD no código. Em código Java ou Scala, use a API Dataset Spark SQL como um superconjunto de RDDs e DataFrames.

Usar o Apache Spark 3

O Serviço Gerenciado para Apache Spark 2.0 instala o Spark 3, que inclui os seguintes recursos e melhorias de desempenho:

  • Suporte a GPUs
  • Capacidade de ler arquivos binários
  • Melhorias de desempenho
  • Remoção dinâmica de partições
  • Execução de consulta adaptável, que otimiza jobs do Spark em tempo real

Usar alocação dinâmica

O Apache Spark inclui um recurso de alocação dinâmica que escalona o número de executores do Spark nos workers de um cluster. Esse recurso permite que um job use todo o cluster do Serviço Gerenciado para Apache Spark, mesmo quando o cluster é escalonado verticalmente. Esse recurso é ativado por padrão no Serviço Gerenciado para Apache Spark (spark.dynamicAllocation.enabled é definido como true). Consulte Alocação dinâmica do Spark para mais informações.

Usar o escalonamento automático do Serviço Gerenciado para Apache Spark

O escalonamento automático do Serviço Gerenciado para Apache Spark adiciona e remove dinamicamente workers do Serviço Gerenciado para Apache Spark de um cluster para garantir que os jobs do Spark tenham os recursos necessários para serem concluídos rapidamente.

É uma prática recomendada configurar a política de escalonamento automático para escalonar apenas workers secundários.

Usar o modo de flexibilidade avançada do Serviço Gerenciado para Apache Spark

Os clusters com VMs preemptivas ou uma política de escalonamento automático podem receber exceções FetchFailed quando os workers forem interrompidos ou removidos antes de terminarem de exibir dados de embaralhamento para redutores. Essa exceção pode causar novas tentativas de tarefas e tempos de conclusão de jobs mais longos.

Recomendação: use o Modo de Flexibilidade Avançada do Serviço Gerenciado para Apache Spark, que não armazena dados de embaralhamento intermediários em workers secundários, para que eles possam ser preemptivos ou reduzidos com segurança.

Configurar o particionamento e o embaralhamento

O Spark armazena dados em partições temporárias no cluster. Se o aplicativo agrupar ou mesclar DataFrames, os dados serão embaralhados em novas partições de acordo com o agrupamento e a configuração de baixo nível.

O particionamento de dados afeta significativamente o desempenho do aplicativo: um número muito pequeno de partições limita o paralelismo de jobs e a utilização de recursos do cluster. Muitas partições reduzem o job devido ao processamento e particionamento adicionais de partição.

Como configurar partições

As seguintes propriedades regem o número e o tamanho das suas partições:

  • spark.sql.files.maxPartitionBytes: o tamanho máximo das partições referentes à leitura de dados do Cloud Storage. O padrão é 128 MB, o que é suficientemente grande para a maioria dos aplicativos que processam menos de 100 TB.

  • spark.sql.shuffle.partitions: o número de partições após a execução de uma ordem aleatória. O padrão é 1000 para clusters de versão de imagem 2.2 e posteriores. Recomendação: defina como 3 vezes o número de vCPUs no cluster.

  • spark.default.parallelism: o número de partições retornadas após a execução de transformações RDD que exigem embaralhamentos, como join, reduceByKey e parallelize. O padrão é o número total de vCPUs no cluster. Ao usar RDDs em jobs do Spark, é possível definir esse número como três vezes suas vCPUs

Limitar o número de arquivos

Há uma perda de desempenho quando o Spark lê muitos arquivos pequenos. Armazene dados em arquivos maiores, por exemplo, com tamanhos entre 256 MB e 512 MB. Da mesma forma, limite o número de arquivos de saída (para forçar uma ordem aleatória, consulte Evitar ordens aleatórias desnecessárias).

Configurar execução de consulta adaptável (Spark 3)

A execução de consultas adaptáveis (ativada por padrão na versão 2.0 da imagem do Serviço Gerenciado para Apache Spark) oferece melhorias no desempenho de jobs do Spark, incluindo:

Embora as configurações padrão sejam corretas para a maioria dos casos de uso, definir spark.sql.adaptive.advisoryPartitionSizeInBytes como spark.sql.files.maxPartitionBytes (padrão de 128 MB) pode ser útil.

Evite embaralhamentos desnecessários

O Spark permite que os usuários acionem manualmente uma ordem aleatória para reequilibrar os dados com a função repartition. Como os embaralhamentos são caros, a reprodução aleatória de dados precisa ser usada com cuidado. Definir as configurações de partição de maneira apropriada deve ser suficiente para permitir que o Spark particionar automaticamente os dados.

Exceção: ao gravar dados particionados em colunas no Cloud Storage, o reparticionamento em uma coluna específica evita a gravação de muitos arquivos pequenos para conseguir tempos de gravação mais rápidos. 

df.repartition("col_name").write().partitionBy("col_name").save("gs://...")

Armazenar dados em Parquet ou Avro

Por padrão, o Spark SQL lê e grava dados em arquivos Parquet compactados Snappy. O Parquet é um formato de arquivo em colunas eficiente que permite ao Spark ler apenas os dados necessários para executar um aplicativo. Essa é uma vantagem importante ao trabalhar com grandes conjuntos de dados. Outros formatos de colunas, como o Apache ORC, também têm bom desempenho.

Para dados não colunares, o Apache Avro fornece um formato de arquivo de linha binária eficiente. Embora geralmente seja mais lento que o Parquet, o desempenho do Avro é melhor do que o de formatos baseados em texto, como CSV ou JSON.

Otimizar o tamanho do disco

A capacidade dos discos permanentes é escalonada de acordo com o tamanho do disco, o que pode afetar o desempenho do job do Spark. Isso acontece porque os jobs gravam metadados e embaralham os dados no disco. Ao usar discos permanentes padrão, o tamanho do disco precisa ser de pelo menos 1 terabytes por worker. Consulte Desempenho por tamanho do disco permanente.

Para monitorar a capacidade de disco do worker no console doCloud de Confiance :

  1. Clique no nome do cluster na página Clusters.
  2. Clique na guia INSTÂNCIAS de VM.
  3. Clique no nome de qualquer worker.
  4. Clique na guia MONITORAMENTO e role até Capacidade de disco para ver a capacidade do worker.

Considerações sobre disco

Os clusters temporários do Serviço Gerenciado para Apache Spark, que não se beneficiam do armazenamento permanente, podem usar SSDs locais. Os SSDs locais estão fisicamente conectados ao cluster e fornecem maior capacidade do que os discos permanentes. Consulte a Tabela de desempenho. Os SSDs locais estão disponíveis em um tamanho fixo de 375 gigabytes, mas é possível adicionar vários SSDs para melhorar o desempenho.

Os SSDs locais não mantêm os dados depois que um cluster é encerrado. Se você precisar de armazenamento permanente, use discos permanentes SSD, que oferecem uma capacidade de processamento maior para o tamanho deles do que os discos permanentes padrão. Os discos permanentes SSD também são uma boa opção se o tamanho da partição for menor que 8 KB (no entanto, evite partições pequenas).

Anexe GPUs ao cluster

O Spark 3 é compatível com GPUs. Use GPUs com a ação de inicialização do CUDA-X para acelerar os jobs do Spark usando a biblioteca NVIDIA cuDF para Apache Spark. A ação de inicialização do driver da GPU para configurar um cluster com GPUs.

Falhas e correções comuns de jobs

A seção a seguir aborda as causas e soluções comuns para falhas de jobs.

Memória insuficiente

Exemplos:

  • "Executor perdido"
  • "java.lang.OutOfMemoryError: limite de sobrecarga do GC excedido"
  • "Contêiner encerrado pelo YARN por exceder os limites de memória"

Correções possíveis:

Falhas na busca aleatória

Exemplos:

  • "FetchFailedException" (erro do Spark)
  • "Falha na conexão com..." (Erro do Spark)
  • "Falha ao buscar" (erro MapReduce)

Geralmente, é causado pela remoção prematura de workers que ainda têm dados embaralhados para serem exibidos.

Causas e correções possíveis:

  • VMs de workers preemptivos foram recuperadas, ou VMs de workers não preemptivos foram removidas pelo escalonador automático. Solução: use o Modo de flexibilidade aprimorado para tornar os workers secundários preemptivos ou escalonáveis com segurança.
  • O executor ou mapeador falhou devido a um erro de OutOfMemory. Solução: aumentar a memória do executor ou mapeador.
  • O serviço do Spark Shuffle pode estar sobrecarregado. Solução: diminuir o número de partições do job.

Os nós YARN estão NÃO ÍNTEGROS

Exemplos (de registros YARN): 

...reported UNHEALTHY with details: 1/1 local-dirs usable space is below
configured utilization percentage/no more usable space
[ /hadoop/yarn/nm-local-dir : used space above threshold of 90.0% ]

Muitas vezes, relacionados ao espaço em disco insuficiente para a reprodução aleatória de dados. Faça o diagnóstico visualizando os arquivos de registro:

  • Abra a página Clusters do projeto no console do Cloud de Confiance e clique no nome do cluster.
  • Clique em VER REGISTROS.
  • Filtre os registros por hadoop-yarn-nodemanager.
  • Pesquise "NÃO-ÍNTEGROS".

Possíveis correções:

  • O cache do usuário é armazenado no diretório especificado pela propriedade yarn.nodemanager.local-dirs no yarn-site.xml file. Esse arquivo está localizado em /etc/hadoop/conf/yarn-site.xml. Verifique o espaço livre no caminho /hadoop/yarn/nm-local-dir e libere espaço excluindo a pasta de cache do usuário /hadoop/yarn/nm-local-dir/usercache.
  • Se o registro informar o status "UNHEALTHY", recrie o cluster com mais espaço em disco, o que vai aumentar o limite de capacidade de processamento.

O cache do usuário está ocupando espaço em disco

No Serviço Gerenciado para Apache Spark, o cache do usuário é armazenado no diretório especificado pela propriedade yarn.nodemanager.local-dirs em yarn-site.xml (/etc/hadoop/conf/yarn-site.xml). Na maioria dos casos, o local é /hadoop/yarn/nm-local-dir.

Possíveis correções:

  • Limpeza manual:verifique o espaço livre no caminho /hadoop/yarn/nm-local-dir e, em seguida, libere espaço excluindo a pasta de cache do usuário /hadoop/yarn/nm-local-dir/usercache.

  • Resolução de longo prazo:gerencie o processo de limpeza definindo as seguintes propriedades do cluster:

    • yarn.nodemanager.localizer.cache.cleanup.interval-ms
    • yarn.nodemanager.localizer.cache.target-size-mb

    Se você não quiser recriar o cluster, atualize essas propriedades em yarn-site.xml em todos os nós e reinicie o NodeManager executando o seguinte comando nos nós de trabalho:

    sudo systemctl restart hadoop-yarn-nodemanager.service

  • Redimensionar o cluster:se o registro informar um status "UNHEALTHY", recrie o cluster com mais espaço em disco, o que vai aumentar o limite de taxa de transferência.

O job falha devido à memória insuficiente do driver

Ao executar jobs no modo de cluster, o job falhará se o tamanho da memória do nó de trabalho for excedido.

Exemplo de registros de driver: 

'Exception in thread "main" java.lang.IllegalArgumentException:
Required AM memory (32768+3276 MB) is above the max threshold (12288 MB) of this cluster!
Please check the values of 'yarn.scheduler.maximum -allocation-mb' and/or 'yarn.nodemanager.resource.memory-mb'.'

Possíveis correções:

  • Defina spark:spark.driver.memory como menor que yarn:yarn.scheduler.maximum-allocation-mb.
  • Use o mesmo tipo de máquina para nós mestre e de trabalho.

A seguir