Escalonar clusters automaticamente

Objetivo :entender como configurar e usar o escalonamento automático do Serviço Gerenciado para Apache Spark para escalonar de forma automática e dinâmica o número de VMs de worker em clusters do Serviço Gerenciado para Apache Spark para atender às demandas de carga de trabalho.

O que é escalonamento automático

É difícil estimar o número "certo" de workers de cluster (nós) para uma carga de trabalho, e um único tamanho de cluster para um pipeline inteiro geralmente não é o ideal. O escalonamento de cluster iniciado pelo usuário aborda parcialmente esse desafio, mas requer o monitoramento da utilização de cluster e da intervenção manual.

A API ScalingPolicies do Dataproc fornece um mecanismo para automatização do gerenciamento de recursos do cluster. Além disso, permite o escalonamento automático da VM de worker do cluster. Um Autoscaling Policy é uma configuração reutilizável, que descreve como os workers do cluster que usam a política de escalonamento automático precisam ser escalonados. Ela define os limites, a frequência e a intensidade de escalonamento para fornecer controle detalhado dos recursos do cluster durante todo o ciclo de vida dele.

• O escalonamento automático do Serviço Gerenciado para Apache Spark dimensiona os nós de trabalho do cluster, não os mestres do cluster.
• O escalonamento automático do Serviço Gerenciado para Apache Spark é compatível com o escalonamento horizontal (escalonamento do número de nós), não com o escalonamento vertical (escalonamento de tipos de máquina).

Quando usar o escalonamento automático

Use o escalonamento automático:

em clusters que armazenam dados em serviços externos, como o Cloud Storage ou o BigQuery

em clusters que processam muitos jobs

para escalonar verticalmente clusters de job único

com Modo de flexibilidade aprimorado para jobs em lote do Spark

O escalonamento automático não é recomendado com/para:

• HDFS: o escalonamento automático não se destina ao escalonamento de HDFS no cluster porque:

  1. A utilização de HDFS não é um sinal para o escalonamento automático.
  2. Os dados do HDFS são hospedados apenas em workers primários. O número de workers primários precisa ser suficiente para hospedar todos os dados do HDFS.
  3. A desativação dos DataNodes do HDFS pode atrasar a remoção de workers. Os nós de dados copiam blocos do HDFS para outros nós de dados antes da remoção de um worker. Dependendo do tamanho dos dados e do fator de replicação, esse processo pode levar horas.

• Rótulos de nós do YARN:o escalonamento automático não é compatível com rótulos de nós do YARN nem com a propriedade dataproc:am.primary_only devido ao YARN-9088. O YARN relata incorretamente as métricas do cluster quando os rótulos dos nós são usados.

• Spark Structured Streaming:o escalonamento automático não oferece suporte ao Spark Structured Streaming (consulte Escalonamento automático e Spark Structured Streaming).

• Clusters inativos: o escalonamento automático não é recomendado para escalonar um cluster até seu tamanho mínimo, quando ele estiver ocioso. O tempo necessário para a criação ou o redimensionamento de um cluster é o mesmo, por isso, pense em excluir clusters inativos e criar novos. As seguintes ferramentas são compatíveis com esse modelo efêmero:

  Use os fluxos de trabalho do Serviço Gerenciado para Apache Spark para programar um conjunto de jobs em um cluster dedicado e, em seguida, exclua o cluster quando os jobs forem concluídos. Para uma orquestração mais avançada, use o Cloud Composer, baseado no Apache Airflow.

  Para clusters que processam consultas ad hoc ou cargas de trabalho agendadas externamente, use a Exclusão agendada de cluster para excluir o cluster após um determinado período de ociosidade ou em um horário específico.

• Cargas de trabalho de tamanhos diferentes:quando jobs pequenos e grandes são executados em um cluster, a redução de escala vertical de desativação otimizada aguarda a conclusão dos jobs grandes. Como resultado, um job de longa duração vai atrasar o escalonamento automático de recursos para jobs menores em execução no cluster até que o job de longa duração seja concluído. Para evitar esse resultado, agrupe jobs menores de tamanho semelhante em um cluster e isole cada job de longa duração em um cluster separado.

Ativar o escalonamento automático

Para ativar o escalonamento automático em um cluster:

1. Crie uma política de escalonamento automático.

2. Siga uma destas instruções:

   1. Crie um cluster de escalonamento automático ou
   2. ative o escalonamento automático em um cluster atual.

Criar uma política de escalonamento automático

Console doGoogle Cloud

Para criar uma política de escalonamento automático, selecione CRIAR POLÍTICA na página Políticas de escalonamento automático do Managed Service for Apache Spark usando o console do Google Cloud . Na página Criar política, é possível selecionar um painel de recomendação de política para preencher os campos de política de escalonamento automático para um tipo de serviço ou objetivo de escalonamento específico.

CLI da gcloud

Você pode usar o comando gcloud dataproc autoscaling-policies import para criar uma política de escalonamento automático. Ele lê um arquivo YAML local que define uma política de escalonamento automático. O formato e o conteúdo do arquivo precisam corresponder aos objetos de config e aos campos definidos pela API REST autoscalingPolicies.

O exemplo de YAML a seguir define uma política para clusters padrão do Serviço Gerenciado para Apache Spark com todos os campos obrigatórios. Ele também fornece os valores minInstances e maxInstances para os workers primários, o valor maxInstances para os workers secundários (preemptivos) e especifica um cooldownPeriod de 4 minutos (o padrão é 2 minutos). O workerConfig configura os workers primários. Neste exemplo, minInstances e maxInstances são definidos com o mesmo valor para evitar o escalonamento dos workers primários.

workerConfig:
  minInstances: 10
  maxInstances: 10
secondaryWorkerConfig:
  maxInstances: 50
basicAlgorithm:
  cooldownPeriod: 4m
  yarnConfig:
    scaleUpFactor: 0.05
    scaleDownFactor: 1.0
    gracefulDecommissionTimeout: 1h

O exemplo de YAML a seguir define uma política para clusters padrão do Serviço Gerenciado para Apache Spark, com todos os campos obrigatórios e opcionais da política de escalonamento automático.

clusterType: STANDARD
workerConfig:
  minInstances: 10
  maxInstances: 10
  weight: 1
secondaryWorkerConfig:
  minInstances: 0
  maxInstances: 100
  weight: 1
basicAlgorithm:
  cooldownPeriod: 2m
  yarnConfig:
    scaleUpFactor: 0.05
    scaleDownFactor: 1.0
    scaleUpMinWorkerFraction: 0.0
    scaleDownMinWorkerFraction: 0.0
    gracefulDecommissionTimeout: 1h

O exemplo de YAML a seguir define uma política para clusters de escalonamento zero.

Para clusters de escala zero, não inclua workerConfig.

clusterType: ZERO_SCALE
secondaryWorkerConfig:
  minInstances: 0
  maxInstances: 100
  weight: 1
basicAlgorithm:
  cooldownPeriod: 2m
  yarnConfig:
    scaleUpFactor: 0.05
    scaleDownFactor: 1.0
    scaleUpMinWorkerFraction: 0.0
    scaleDownMinWorkerFraction: 0.0
    gracefulDecommissionTimeout: 1h

Execute o comando gcloud a seguir em um terminal local ou no Cloud Shell para criar a política de escalonamento automático. Forneça um nome para a política. Esse nome se tornará a política id, que poderá ser usada em comandos gcloud posteriores para fazer referência à política. Use a sinalização --source para especificar o caminho local e o nome do arquivo YAML da política de escalonamento automático a ser importado.

gcloud dataproc autoscaling-policies import policy-name \
    --source=filepath/filename.yaml \
    --region=region

API REST

Crie uma política de escalonamento automático definindo uma AutoscalingPolicy como parte de uma solicitação autoscalingPolicies.create.

Criar um cluster de escalonamento automático

Depois de criar uma política de escalonamento automático, crie um cluster que use a política de escalonamento automático. O cluster precisa estar na mesma região da política de escalonamento automático.

Console doGoogle Cloud

Para aplicar uma política de escalonamento automático a um novo cluster:

1. Abra a página Criar cluster do Serviço Gerenciado para Apache Spark.
2. Na seção Definir o cluster, em Escolher o modo de escalonamento de workers, selecione Escalonamento automático e, em seguida, selecione sua política no campo Política.

CLI da gcloud

Execute o seguinte comando gcloud em um terminal local ou no Cloud Shell para criar um cluster de escalonamento automático. Forneça um nome para o cluster e use a sinalização --autoscaling-policy para especificar o policy ID (o nome da política especificada quando você criou a política) ou a política resource URI (resource name) (consulte os campos AutoscalingPolicy id e name).

gcloud dataproc clusters create cluster-name \
    --autoscaling-policy=policy id or resource URI \
    --region=region

O nome completo do recurso (ou URI) de uma política pode ser obtido executando o seguinte comando gcloud :

gcloud dataproc autoscaling-policies \
    describe policy-id \
    --region=region

API REST

Crie um cluster de escalonamento automático incluindo um AutoscalingConfig como parte de uma solicitação clusters.create.

Para visualizar o AutoscalingConfig.policyUri (nome do recurso da política), execute o seguinte comando gcloud :

gcloud dataproc autoscaling-policies \
    describe policy-name \
    --region=region

Ativar o escalonamento automático em um cluster atual

Depois de criar uma política de escalonamento automático, é possível ativar a política em um cluster atual na mesma região.

Console doGoogle Cloud

Não é possível ativar uma política de escalonamento automático em um cluster atual no console do Google Cloud .

CLI da gcloud

Execute o seguinte comando gcloud em um terminal local ou no Cloud Shell para ativar uma política de escalonamento automático em um cluster atual. Forneça o nome do cluster e use a sinalização --autoscaling-policy para especificar o policy ID (o nome da política especificada ao criar a política). ou a política resource URI (resource name) (consulte os campos AutoscalingPolicy id e name).

gcloud dataproc clusters update cluster-name \
    --autoscaling-policy=policy id or resource URI \
    --region=region

O nome completo do recurso (ou URI) de uma política pode ser obtido executando o seguinte comando gcloud :

gcloud dataproc autoscaling-policies \
    describe policy-id \
    --region=region

API REST

Para ativar uma política de escalonamento automático em um cluster atual, defina o AutoscalingConfig.policyUri da política no updateMask de uma solicitação clusters.patch.

Para visualizar o AutoscalingConfig.policyUri (nome do recurso da política), execute o seguinte comando gcloud :

gcloud dataproc gcloud dataproc autoscaling-policies \
    describe policy-name \
    --region=region

Uso da política de vários clusters

• Uma política de escalonamento automático define o comportamento de escalonamento que pode ser aplicado a vários clusters. Uma política de escalonamento automático é melhor aplicada em vários clusters quando eles compartilham cargas de trabalho semelhantes ou executam jobs com padrões de uso de recursos semelhantes.

• É possível atualizar uma política que está sendo usada por vários clusters. As atualizações afetam imediatamente o comportamento de escalonamento automático de todos os clusters que usam a política. Consulte autoscalingPolicies.update. Se você não quiser que uma atualização de política seja aplicada a um cluster que está usando a política, desative o escalonamento automático no cluster antes de atualizar a política.

Console doGoogle Cloud

Não é possível desativar o escalonamento automático em um cluster no console do Google Cloud .

CLI da gcloud

Execute o comando gcloud em um terminal local ou no Cloud Shell para desativar o escalonamento automático em um cluster.

gcloud dataproc clusters update cluster-name --disable-autoscaling \
    --region=region

API REST

Para desativar o escalonamento automático em um cluster, defina AutoscalingConfig.policyUri para a string vazia e defina update_mask=config.autoscaling_config.policy_uri em uma solicitação clusters.patch.

• Uma política em uso por um ou mais clusters não pode ser excluída. Consulte autoscalingPolicies.delete.

Como o escalonamento automático funciona

As métricas da YARN do Hadoop do cluster são verificadas pelo escalonamento automático, no decorrer de cada período de "resfriamento", para definir se o cluster precisa ser escalonado e qual a magnitude da atualização.

1. O valor da métrica de recurso pendente do YARN (memória pendente ou núcleos pendentes) determina se é necessário escalonar verticalmente ou reduzir a escala. Um valor maior que 0 indica que os jobs do YARN estão aguardando recursos e que pode ser necessário escalonar verticalmente. Um valor 0 indica que o YARN tem recursos suficientes para que uma redução de escala vertical ou outras mudanças não sejam necessárias.

   Se o recurso pendente for > 0:

   $estimated\_worker\_count =$

   \[ \Biggl \lceil AVERAGE\ during\ cooldown\ period\Big(\frac{Pending + Available + Allocated + Reserved}{Resource\ per\ worker}\Big)\Biggr \rceil \]

   Se o recurso pendente for 0:

   $estimated\_worker\_count =$

   \[ \Biggl \lceil AVERAGE\ during\ cooldown\ period\Big(\frac{Allocated + Reserved}{Resource\ per\ worker}\Big)\Biggr \rceil \]

   Por padrão, a partir da imagem 2.2 do Serviço Gerenciado para Apache Spark, o escalonador automático monitora a memória e os núcleos do YARN para queestimated_worker_count seja avaliado separadamente para memória e núcleos, e o número maior de workers resultante seja selecionado. Para versões de imagem anteriores, o escalonador automático monitora apenas a memória YARN, a menos que você ative o escalonamento automático baseado em núcleos para monitorar a memória YARN e os núcleos.

   $estimated\_worker\_count =$

   \[ max(estimated\_worker\_count\_by\_memory,\ estimated\_worker\_count\_by\_cores) \]

   \[ estimated\ \Delta worker = estimated\_worker\_count - current\_worker\_count \]

   Observação :os workers secundários, incluindo aqueles programados para criação, estão incluídos na contagem atual de workers.

2. Considerando a mudança estimada necessária no número de workers, o escalonamento automático usa um scaleUpFactor ou scaleDownFactor para calcular a mudança real no número de workers:

   if estimated Δworkers > 0:
     actual Δworkers = ROUND_UP(estimated Δworkers * scaleUpFactor)
     # examples:
     # ROUND_UP(estimated Δworkers=5 * scaleUpFactor=0.5) = 3
     # ROUND_UP(estimated Δworkers=0.8 * scaleUpFactor=0.5) = 1
   else:
     actual Δworkers = ROUND_DOWN(estimated Δworkers * scaleDownFactor)
     # examples:
     # ROUND_DOWN(estimated Δworkers=-5 * scaleDownFactor=0.5) = -2
     # ROUND_DOWN(estimated Δworkers=-0.8 * scaleDownFactor=0.5) = 0
     # ROUND_DOWN(estimated Δworkers=-1.5 * scaleDownFactor=0.5) = 0
   

   Um scaleUpFactor ou scaleDownFactor de 1.0 significa que o escalonamento automático será feito de modo que o recurso pendente ou disponível seja 0 (utilização perfeita).

3. Depois que a mudança no número de workers é calculada, o scaleUpMinWorkerFraction e o scaleDownMinWorkerFraction atuam como um limite para determinar se o escalonamento automático vai escalonar o cluster. Uma fração pequena significa que o escalonamento automático deve ser feito mesmo que o Δworkers seja pequeno. Uma fração maior significa que o escalonamento só pode ocorrer quando o Δworkers é grande.

   IF (Δworkers >  scaleUpMinWorkerFraction * current_worker_count) then scale up
   

   OU

   IF (abs(Δworkers) >  scaleDownMinWorkerFraction * current_worker_count),
   THEN scale down.
   

4. Se o número de workers a escalonar for grande o suficiente para acionar o escalonamento, o escalonamento automático usará os limites minInstances maxInstances de workerConfig e secondaryWorkerConfig e weight (proporção de workers primários e secundários) para determinar como dividir o número de workers entre os grupos de instâncias de worker primário e secundário. O resultado desses cálculos é a alteração final de escalonamento automático no cluster para o período de escalonamento.

5. As solicitações de redução de escalonamento automático serão canceladas em clusters criados com versões de imagem 2.0.57+, 2.1.5+ e versões de imagem posteriores se:

   1. um redução de escala está em andamento com um valor de tempo limite de desativação otimizada diferente de zero, e

   2. o número de workers ATIVOS do YARN ("workers ativos") mais a alteração no número total de workers recomendados pelo escalonador automático (Δworkers) é igual ou maior que DECOMMISSIONING workers do YARN ("workers em desativação"), conforme mostrado na seguinte fórmula:

      IF (active workers + Δworkers ≥ active workers + decommissioning workers)
      THEN cancel the scaledown operation
      

      A fórmula de cancelamento da operação de redução de escala vertical considera o número de workers ativos, não o current_worker_count (usado nas fórmulas de escalonamento automático anteriores).

   Para ver um exemplo de cancelamento de redução, consulte Quando o escalonamento automático cancela uma operação de redução?.

Recomendações de configuração de escalonamento automático

Nesta seção, você encontra recomendações para configurar o escalonamento automático.

Evitar o escalonamento de workers primários

Os workers primários executam Datanodes do HDFS, enquanto os secundários são apenas de computação. O uso de workers secundários permite escalonar recursos de computação de maneira eficiente sem a necessidade de provisionar armazenamento, resultando em recursos de escalonamento mais rápidos. Os Namenodes do HDFS podem ter várias disputas que corrompem o HDFS, fazendo com que a desativação fique travada indefinidamente. Para evitar esse problema, não escalone os workers primários. Por exemplo: none workerConfig: minInstances: 10 maxInstances: 10 secondaryWorkerConfig: minInstances: 0 maxInstances: 100

Algumas modificações precisam ser feitas no comando de criação do cluster:

1. Defina --num-workers=10 para corresponder ao tamanho do grupo de workers principais da política de escalonamento automático.
2. Defina --secondary-worker-type=non-preemptible para configurar workers secundários como não preemptivos. (a menos que as VMs preemptivas sejam necessárias).
3. Copie a configuração de hardware dos workers primários para os secundários. Por exemplo, defina --secondary-worker-boot-disk-size=1000GB para corresponder a --worker-boot-disk-size=1000GB.

Use o modo de flexibilidade aprimorado para jobs em lote do Spark

Use o Modo de flexibilidade aprimorado (EFM, na sigla em inglês) com o escalonamento automático para:

permitir o redução de escala mais rápida do cluster enquanto os jobs estão em execução

evitam interrupções nos jobs em execução devido à redução de escala vertical do cluster

minimizar a interrupção de jobs em execução devido à preempção de workers secundários preemptivos

Com o EFM ativado, o tempo limite de desativação otimizada de uma política de escalonamento automático precisa ser definido como 0s. A política de escalonamento automático só precisa escalonar os workers secundários automaticamente.

Escolher um tempo limite de desativação otimizada

O escalonamento automático é compatível com a desativação otimizada do YARN ao remover nós de um cluster. A desativação otimizada permite que os aplicativos concluam a reprodução aleatória de dados entre cenários para evitar a definição do andamento do job. O ⁠tempo limite de desativação otimizada fornecido em uma política de escalonamento automático é o limite superior da duração que o YARN aguardará para executar aplicativos (aplicativos que estavam em execução quando a desativação começou) antes de remover nós.

Observação :o escalonamento automático não vai escalonar verticalmente durante a desativação otimizada. Ele aguarda a conclusão de uma operação de atualização, incluindo a desativação otimizada, continua aguardando por um período de espera (por exemplo, 2m) e, em seguida, emite outra operação de atualização.

Quando um processo não é concluído dentro do período de tempo limite de desativação otimizada especificado, o nó de trabalho é desligado à força, o que pode causar perda de dados ou interrupção do serviço. Para evitar essa possibilidade, defina o tempo limite de desativação otimizada como um valor maior do que o job mais longo que o cluster processará. Por exemplo, se você espera que o job mais longo seja executado por uma hora, defina o tempo limite como pelo menos uma hora (1h).

Pense na migração de jobs que levam mais de uma hora para os próprios clusters efêmeros para evitar o bloqueio da desativação otimizada.

Como definir scaleUpFactor

scaleUpFactor controla a intensidade com que o escalonador automático aumenta um cluster. Especifique um número entre 0.0 e 1.0 para definir o valor fracionário do recurso pendente do YARN que causa a adição de nós.

Por exemplo, se houver 100 contêineres pendentes solicitando 512 MB cada, haverá 50 GB de memória pendente do YARN. Se scaleUpFactor for 0.5, o escalonador automático adicionará nós suficientes para adicionar 25 GB de memória do YARN. Da mesma forma, se for 0.1, o escalonador automático adicionará nós suficientes para 5 GB. Observe que esses valores correspondem à memória do YARN, não à memória total disponível fisicamente em uma VM.

Um bom ponto de partida é 0.05 para jobs do MapReduce e jobs do Spark com a alocação dinâmica ativada. Para os jobs do Spark com uma contagem de executores fixos e jobs do Tez, use 1.0. Um scaleUpFactor de 1.0 significa que o escalonamento automático será feito para que o recurso pendente ou disponível seja 0 (utilização perfeita).

Como definir scaleDownFactor

scaleDownFactor controla o nível de agressividade com que o escalonador automático reduz a escala vertical de um cluster. Especifique um número entre 0.0 e 1.0 para definir o valor fracionário do recurso disponível do YARN que causa a remoção do nó.

Deixe esse valor como 1.0 para a maioria dos clusters com vários jobs que precisam ser escalonar verticalmente e para baixo com frequência. Como resultado da desativação otimizada, as operações de redução de escala são significativamente mais lentas do que as de escalonamento vertical. A definição de scaleDownFactor=1.0 define uma taxa de redução agressiva, que minimiza o número de operações de redução necessárias para atingir o tamanho adequado do cluster.

Para clusters que precisam de mais estabilidade, defina um scaleDownFactor menor para uma taxa de redução de escala vertical mais lenta.

Defina esse valor como 0.0 para evitar a redução da escala vertical do cluster, por exemplo, ao usar clusters temporários ou de job único.

Observação :se você definir scaleUpFactor e scaleDownFactor como 0, o cluster não será escalonado.

Definindo scaleUpMinWorkerFraction e scaleDownMinWorkerFraction

scaleUpMinWorkerFraction e scaleDownMinWorkerFraction são usados com scaleUpFactor ou scaleDownFactor e têm valores padrão de 0.0. Eles representam os limites em que o escalonador automático vai escalonar verticalmente ou reduzir escala vertical o cluster: o aumento ou a diminuição fracionária mínima no tamanho do cluster necessária para emitir solicitações de escalonamento vertical ou redução escala vertical.

Exemplos: o escalonador automático não emitirá uma solicitação de atualização para adicionar cinco workers a um cluster de 100 nós, a menos que scaleUpMinWorkerFraction seja menor ou igual a 0.05 (5%). Se definido como 0.1, o escalonador automático não vai emitir a solicitação para escalonar verticalmente o cluster. Da mesma forma, se scaleDownMinWorkerFraction for 0.05, o escalonador automático não removerá pelo menos cinco nós.

O valor padrão de 0.0 significa que não há limite.

Definir um valor de scaleDownMinWorkerFractionthresholds mais alto em clusters grandes (> 100 nós) para evitar operações de escalonamento pequenas e desnecessárias é altamente recomendado.

Escolher um período de espera

O cooldownPeriod define um período em que o escalonador automático não emite solicitações para alterar o tamanho do cluster. É possível usá-lo para limitar a frequência de mudanças do escalonador automático no tamanho do cluster.

O cooldownPeriod mínimo e padrão é de dois minutos. Se um cooldownPeriod mais curto for definido em uma política, as mudanças na carga de trabalho vão afetar mais rapidamente o tamanho do cluster, mas os clusters poderão ser escalonar verticalmente e horizontalmente de maneira desnecessária. A prática recomendada é definir scaleUpMinWorkerFraction e scaleDownMinWorkerFraction da política como um valor diferente de zero ao usar um cooldownPeriod mais curto. Isso garante que o cluster só seja escalonado vertical ou horizontalmente quando a mudança na utilização de recursos for suficiente para justificar uma atualização do cluster.

Se a carga de trabalho for sensível a mudanças no tamanho do cluster, aumente o período de espera. Por exemplo, se você estiver executando um job de processamento em lote, defina o período de espera como 10 minutos ou mais. Teste diferentes períodos de espera para encontrar o valor que funciona melhor para sua carga de trabalho.

Limites de contagem de workers e pesos de grupo

Cada grupo de workers tem minInstances e maxInstances que configuram um limite absoluto no tamanho de cada grupo.

Cada grupo também tem um parâmetro chamado weight que configura o equilíbrio de destino entre os dois grupos. Observe que esse parâmetro é apenas uma dica e, se um grupo atingir o tamanho mínimo ou máximo, os nós só serão adicionados ou removidos do outro grupo. Assim, weight quase sempre pode ser deixado no 1 padrão.

Observação: se minInstances ou maxInstances forem atualizados em uma política de escalonamento automático para que o tamanho de um cluster esteja fora dos limites, o escalonador automático não vai escalonar verticalmente ou diminuir imediatamente para os novos limites. Em vez disso, o escalonador automático apenas evitará o escalonamento do cluster para além dos limites. Consulte https://issuetracker.google.com/160640545.

Atenção: Se você definir minInstances = maxInstances, o cluster não fará o escalonamento.

Usar o escalonamento automático com base em núcleos

Para aplicativos que usam muita CPU, uma prática recomendada é usar a Calculadora de recursos dominantes para alocação de recursos. Essa é a configuração padrão do YARN a partir da versão 2.2 da imagem do Serviço Gerenciado para Apache Spark. Com versões de imagem anteriores, o Serviço Gerenciado para Apache Spark configura o YARN para usar métricas de memória para alocação de recursos, a menos que você defina a seguinte propriedade ao criar um cluster para configurar o YARN para usar a Calculadora de Recursos Dominantes:

capacity-scheduler:yarn.scheduler.capacity.resource-calculator=org.apache.hadoop.yarn.util.resource.DominantResourceCalculator

Métricas e registros de escalonamento automático

Os recursos e ferramentas a seguir podem ajudá-lo a monitorar operações de escalonamento automático e respectivos efeitos no seu cluster e nos seus jobs.

Cloud Monitoring

Use o Cloud Monitoring para:

• ver as métricas usadas pelo escalonamento automático;
• visualizar o número de gerenciadores de nós no cluster
• entender porque o escalonamento automático fez ou não o escalonamento do cluster.

Cloud Logging

Use o Cloud Logging para conferir os registros do escalonador automático do Serviço Gerenciado para Apache Spark.

1) Encontre registros para o cluster.

2) Selecione dataproc.googleapis.com/autoscaler.

3) Expanda as mensagens de registro para visualizar o campo status. Os registros estão em JSON, um formato legível por máquina.

4) Expanda a mensagem de registro para ver recomendações de escalonamento, métricas usadas para decisões de escalonamento, o tamanho do cluster original e o novo tamanho do cluster de destino.

Segundo plano: como fazer o escalonamento automático com o Apache Hadoop e o Apache Spark

Nas seções a seguir, abordamos a maneira de interoperação do escalonamento automático com o YARN do Hadoop e o Hadoop Mapreduce e com o Apache Spark, o Spark Streaming e o Spark Structured Streaming.

Métricas do YARN do Hadoop

É possível configurar o escalonamento automático para qualquer mecanismo de execução baseado em YARN.

Observação: o "recurso" do YARN aqui pode ser a memória ou o núcleo do YARN.

O escalonamento automático é centralizado nas seguintes métricas do YARN do Hadoop:

1. Allocated resource refere-se ao total de recursos do YARN ocupados pela execução de contêineres em todo o cluster. Se houver seis contêineres em execução que podem usar até uma unidade de recurso, haverá seis recursos alocados.

2. Available resource é o recurso do YARN no cluster que não é usado por contêineres alocados. Se houver 10 unidades de recursos em todos os gerenciadores de nós e 6 delas estiverem alocadas, haverá 4 recursos disponíveis. Se houver recursos disponíveis (não utilizados) no cluster, o escalonamento automático poderá remover workers dele.

3. Pending resource é a soma das solicitações de recursos do YARN para contêineres pendentes. Os contêineres pendentes aguardam espaço para serem executados no YARN. O recurso pendente é diferente de zero apenas se o recurso disponível for zero ou muito pequeno para ser alocado ao próximo contêiner. Se houver contêineres pendentes, com o escalonamento automático será possível adicionar workers ao cluster.

É possível visualizar essas métricas no Cloud Monitoring. Por padrão, a memória YARN será de 0,8 * de memória total no cluster, com memória restante reservada para outros daemons e uso do sistema operacional, como o cache da página. É possível substituir o valor padrão pela configuração de configuração "yarn.nodemanager.resource.memory-mb" do YARN. Consulte Apache Hadoop YARN, HDFS, Spark e propriedades relacionadas.

Escalonamento automático e o Hadoop MapReduce

Com o MapReduce é executado cada mapa e reduzida a tarefa como um contêiner YARN separado. Quando um job começa, por meio do MapReduce são enviadas solicitações de contêiner para cada tarefa do mapa, resultando em grande aumento na memória pendente do YARN. À medida que as tarefas do mapa terminam, a memória pendente diminui.

Quando mapreduce.job.reduce.slowstart.completedmaps são concluídos (95% por padrão no Serviço Gerenciado para Apache Spark), o MapReduce enfileira as solicitações de contêiner para todos os redutores, resultando em outro pico de memória pendente.

A menos que as tarefas de mapa e redução demorem vários minutos ou mais, não defina um valor alto para o escalonamento automático scaleUpFactor. Adicionar workers ao cluster demora pelo menos 1,5 minuto, portanto, verifique se há trabalho pendente suficiente para utilizar o novo worker por vários minutos. Um bom ponto de partida é definir scaleUpFactor como 0,05 (5%) ou 0,1 (10%) da memória pendente.

Escalonamento automático e o Spark

Por meio do Spark, uma camada extra de agendamento é adicionada no topo do YARN. Especificamente, a alocação dinâmica do Spark Core faz solicitações ao YARN de contêineres para executar executores do Spark e, em seguida, programa tarefas do Spark em linhas de execução nesses executores. Os clusters do Serviço Gerenciado para Apache Spark ativam a alocação dinâmica por padrão, de modo que os executores são adicionados e removidos conforme necessário.

No Spark, sempre é feita a solicitação de contêineres para o YARN, mas sem a alocação dinâmica, a solicitação só é feita no início do job. Com a alocação dinâmica, será feita a solicitação para a remoção de contêineres ou criação de novos, conforme a necessidade.

O Spark é iniciado com um número reduzido de executores - 2, em clusters de escalonamento automático - e o número de executores é duplicado enquanto houver tarefas acumuladas. Isso suaviza a memória pendente (menor número de picos de memória pendentes). É recomendado definir scaleUpFactor de escalonamento automático como um número grande, como 1.0 (100%), para jobs do Spark.

Como desativar a alocação dinâmica do Spark

Se você estiver executando jobs separados do Spark que não se beneficiam da alocação dinâmica do Spark, desative essa opção definindo spark.dynamicAllocation.enabled=false e spark.executor.instances. É possível ainda usar o escalonamento automático para escalonar os clusters para mais ou para menos, enquanto os jobs separados do Spark são executados.

Jobs do Spark com dados armazenados em cache

Defina spark.dynamicAllocation.cachedExecutorIdleTimeout ou remova os conjuntos de dados em cache quando eles não forem mais necessários. Por padrão, o Spark não remove os executores que têm dados armazenados em cache, o que evitaria a redução de escalonamento do cluster.

Escalonamento automático e o Spark Streaming

1. Como o Spark Streaming tem a própria versão de alocação dinâmica, que usa indicadores específicos de streaming para adicionar e remover executores, defina spark.streaming.dynamicAllocation.enabled=true e desative a alocação dinâmica do Spark Core definindo spark.dynamicAllocation.enabled=false.

2. Não use Desativação otimizada (escalonamento automático gracefulDecommissionTimeout) com jobs do Spark Streaming. Em vez disso, para remover com segurança os workers com escalonamento automático, configure o checkpoint para tolerância a falhas.

Como alternativa, para usar o Spark Streaming sem o escalonamento automático:

1. Desative a alocação dinâmica do Spark Core (spark.dynamicAllocation.enabled=false) e
2. defina o número de executores (spark.executor.instances) para o job. Veja Propriedades do cluster.

Escalonamento automático e o Spark Structured Streaming

O escalonamento automático não é compatível com o Spark Structured Streaming, porque ele não oferece suporte à alocação dinâmica (consulte SPARK-24815: o Structured Streaming precisa oferecer suporte à alocação dinâmica).

Controlar o escalonamento automático por particionamento e paralelismo

Geralmente, o paralelismo é definido ou determinado pelos recursos do cluster. Por exemplo, o número de blocos do HDFS bloqueia o número de tarefas, com o escalonamento automático, a conversão se aplica: o escalonamento automático define o número de workers de acordo com o paralelismo do job. A seguir, as diretrizes para definir o paralelismo de jobs:

• Embora o Serviço Gerenciado para Apache Spark defina o número padrão de tarefas de redução do MapReduce com base no tamanho inicial do cluster, é possível definir mapreduce.job.reduces para aumentar o paralelismo da fase de redução.
• O paralelismo do Spark SQL e do DataFrame é determinado por spark.sql.shuffle.partitions, que tem como padrão 200.
• As funções RDD do Spark são padronizadas para spark.default.parallelism, que é definido como o número de núcleos nos nós de trabalho quando o job é iniciado. No entanto, todas as funções RDD que criam embaralhamentos usam um parâmetro para o número de partições, que substitui spark.default.parallelism.

É preciso garantir que os dados sejam particionados de maneira uniforme. Se houver uma distorção significativa de chave, uma ou mais tarefas podem demorar bem mais do que outras, resultando em baixa utilização.

Escalonamento automático das configurações de propriedade padrão do Spark e do Hadoop

Os clusters de escalonamento automático têm valores de propriedade padrão que ajudam a evitar falhas de jobs quando os workers primários são removidos ou os secundários são preemptivos. É possível substituir esses valores padrão ao criar um cluster com escalonamento automático (consulte Propriedades do cluster).

Para ajudar a aumentar a estabilidade do job, crie um cluster com workers secundários não preemptivos. Para ver um exemplo, consulte Como usar preemptivos em um cluster.

Padrões para aumentar o número máximo de novas tentativas para tarefas, mestres de aplicativos e cenários:

yarn:yarn.resourcemanager.am.max-attempts=10
mapred:mapreduce.map.maxattempts=10
mapred:mapreduce.reduce.maxattempts=10
spark:spark.task.maxFailures=10
spark:spark.stage.maxConsecutiveAttempts=10

Padrões para redefinir os contadores de novas tentativas (útil para jobs de execução lenta do Spark Streaming):

spark:spark.yarn.am.attemptFailuresValidityInterval=1h
spark:spark.yarn.executor.failuresValidityInterval=1h

Defina o mecanismo de alocação dinâmica de início lento do Spark para começar com um tamanho pequeno:

spark:spark.executor.instances=2

Observação: mesmo que você veja falhas de busca de redistribuição quando os workers primários são removidos ou os workers secundários são forçados a interromper, não aumente as novas tentativas de busca de redistribuição para mais do que os valores padrão (spark.shuffle.io.maxRetries=3; spark.shuffle.io.retryWaitMs=5000 (5 segundos) para um valor total máximo de duração de novas tentativas de 10 segundos). O motivo: os arquivos de embaralhamento são perdidos, então novas tentativas só consomem mais tempo.

Perguntas frequentes

Esta seção contém perguntas e respostas comuns sobre o escalonamento automático.

O escalonamento automático pode ser ativado em clusters de alta disponibilidade e de nó único?

O escalonamento automático pode ser ativado em clusters de alta disponibilidade, mas não em clusters de nó único (eles não aceitam redimensionamento).

É possível redimensionar manualmente um cluster de escalonamento automático?

Sim É possível redimensionar manualmente um cluster para interromper intervalos, ao ajustar uma política de escalonamento automático. No entanto, essas alterações terão apenas um efeito temporário, e o escalonamento automático acabará redimensionando novamente o cluster.

Em vez de redimensionar manualmente um cluster de escalonamento automático, considere:

Atualização da política de escalonamento automático. Todas as alterações feitas na política de escalonamento automático afetarão todos os clusters que estão usando a política no momento (consulte Uso de políticas de vários clusters).

Desanexar a política e escalonar manualmente o cluster para o tamanho preferencial.

Como receber suporte do Serviço Gerenciado para Apache Spark.

Qual é a diferença entre o Serviço Gerenciado para Apache Spark e o escalonamento automático do Dataflow?

Consulte Escalonamento automático horizontal do Dataflow e Escalonamento automático vertical do Dataflow Prime.

A equipe de desenvolvimento do Serviço Gerenciado para Apache Spark pode redefinir o status do cluster de ERROR para RUNNING?

Em geral, não. Isso exige um esforço manual para verificar se é seguro redefinir o estado do cluster. Muitas vezes, um cluster não pode ser redefinido sem outras etapas manuais, como reiniciar o NameNode do HDFS.

O Serviço Gerenciado para Apache Spark define o status de um cluster como ERROR quando não consegue determinar o status de um cluster após uma operação com falha. Os clusters em ERROR não são escalonados automaticamente. Estas são as causas mais comuns:

• Erros retornados da API Compute Engine, geralmente durante interrupções do Compute Engine.

• O HDFS entra em um estado corrompido devido a bugs na desativação do HDFS.

• Erros da API Dataproc Control, como "Task lease expired".

Exclua e recrie clusters com o status ERROR.

Quando o escalonamento automático cancela uma operação de redução?

O gráfico a seguir é uma ilustração que demonstra quando o escalonamento automático cancela uma operação de redução de escala (consulte também Como funciona o escalonamento automático).

Observações:

• O cluster tem o escalonamento automático ativado com base apenas nas métricas de memória do YARN (padrão).
• T1-T9 representam intervalos de espera em que o escalonador automático avalia o número de workers (o tempo do evento foi simplificado).
• As barras empilhadas representam as contagens de workers do YARN do cluster ativos, em desativação e desativados.
• O número recomendado de workers do escalonador automático (linha preta) se baseia nas métricas de memória do YARN, na contagem de workers ativos do YARN e nas configurações da política de escalonamento automático. Consulte Como funciona o escalonamento automático.
• A área de fundo vermelha indica o período em que a operação de redução está em execução.
• A área de fundo amarela indica o período em que a operação de redução está sendo cancelada.
• A área de fundo verde indica o período da operação de expansão.

As operações a seguir ocorrem nos seguintes horários:

• T1: o escalonador automático inicia uma operação de redução de escala vertical de desativação otimizada para reduzir escala vertical verticalmente cerca de metade dos workers do cluster atual.

• T2: o escalonador automático continua monitorando as métricas do cluster. Ela não altera a recomendação de redução, e a operação de redução continua. Alguns workers foram desativados, e outros estão em processo de desativação (o Serviço Gerenciado para Apache Spark vai excluir os workers desativados).

• o escalonador automático calcula que o número de workers pode reduzir escala vertical ainda mais, talvez devido à disponibilidade de mais memória do YARN. No entanto, como o número de workers ativos mais a mudança recomendada no número de workers não é igual ou maior que o número de workers ativos mais os desativados, os critérios para cancelamento do redução de escala não são atendidos, e o escalonador automático não cancela a operação.

• T4: o YARN informa um aumento na memória pendente. No entanto, o escalonador automático não muda a recomendação de contagem de workers. Assim como no T3, os critérios de cancelamento da redução de escala não são atendidos, e o escalonador automático não cancela a operação.

• T5: a memória pendente do YARN aumenta, e a alteração no número de workers recomendada pelo escalonador automático aumenta. No entanto, como o número de workers ativos mais a mudança recomendada no número de workers é menor do que o número de workers ativos mais os desativados, os critérios de cancelamento não são atendidos, e a operação de redução não é cancelada.

• T6: a memória pendente do YARN aumenta ainda mais. O número de workers ativos mais a mudança no número de workers recomendada pelo escalonador automático agora é maior que o número de workers ativos mais os em desativação. Os critérios de cancelamento são atendidos, e o escalonador automático cancela a operação de redução de escala vertical.

• T7: O escalonador automático está aguardando a conclusão do cancelamento da operação de redução. O escalonador automático não avalia nem recomenda uma mudança no número de workers durante esse intervalo.

• T8: o cancelamento da operação de redução horizontal é concluído. Os workers desativados são adicionados ao cluster e ficam ativos. O escalonador automático detecta a conclusão do cancelamento da operação de redução de escala vertical e aguarda o próximo período de avaliação (T9) para calcular o número recomendado de workers.

• T9: não há operações ativas no horário T9. Com base na política de escalonamento automático e nas métricas do YARN, o escalonador automático recomenda uma operação de escalonamento vertical.