Como fazer o escalonamento automático de clusters

Objetivo: entender como configurar e usar o escalonamento automático do Dataproc para redimensionar de forma automática e dinâmica os clusters, 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 AutoscalingPolicies do Dataproc fornece um mecanismo para automatizar o gerenciamento de recursos de cluster e permite o escalonamento automático de clusters. Um Autoscaling Policy é uma configuração reutilizável que descreve como os clusters que usam a política de escalonamento automático precisam ser escalonados. Ele define limites de escala, frequência e agressividade para fornecer controle refinado sobre os recursos do cluster durante a vida útil do cluster.

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 clusters de job único

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

• Clusters de alta disponibilidade: em vez disso, use clusters padrão, que são mais estáveis após operações de redimensionamento sucessivas.

• HDFS: o escalonamento automático não se destina ao escalonamento de HDFS no cluster. Se você usar o escalonamento automático com o HDFS, verifique se o número mínimo de workers principais é suficiente para processar todos os dados do HDFS. Observe também que, desativar os Datanodes HDFS pode atrasar o processo de remoção de workers.

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

• Structured Streaming do Spark: o escalonamento automático não é compatível com o Structured Streaming do Spark. Veja Escalonamento automático e Structured Streaming do Spark.

• 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 Dataproc para programar um conjunto de jobs em um cluster dedicado e 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.

Como 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

Comando 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 YAML a seguir define uma política que especifica todos os campos obrigatórios. Ele também fornece valores maxInstances para workers primários e secundários (preemptivos) e também especifica um cooldownPeriod de quatro minutos (o padrão é dois minutos).

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

Veja outro exemplo de YAML que especifica todos os campos de política de escalonamento automático opcionais e obrigatórios.

workerConfig:
  minInstances: 2
  maxInstances: 100
  weight: 1
secondaryWorkerConfig:
  minInstances: 0
  maxInstances: 100
  weight: 1
basicAlgorithm:
  cooldownPeriod: 4m
  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 do arquivo 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.

Console

No momento, não é possível criar uma política de escalonamento automático no Console do Google Cloud.

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.

Comando 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 conseguido 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

Console

É possível selecionar uma política de escalonamento automático atual para aplicar a um novo cluster na seção Política de escalonamento automático na página Criar um cluster do Dataproc do Console do Cloud.

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.

Comando 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 conseguido 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

Console

No momento, não é possível ativar uma política de escalonamento automático em um cluster atual no Console do Google Cloud.

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. Em cada avaliação, o escalonamento automático examina a média de memória do cluster pendente e disponível no último cooldown_period para determinar a alteração exata necessária no número de workers:

   exact Δworkers = avg(pending memory - available memory) / memory per node manager

   • pending memory é um sinal de que há tarefas em fila no cluster, ainda não executadas, e que talvez seja necessário escaloná-lo para que haja melhor tratamento da respectiva carga de trabalho.
   • available memory é um sinal de que o cluster tem largura de banda extra em nós íntegros e pode precisar ser reduzido para economizar recursos.
   • Para mais informações sobre essas métricas do YARN do Apache Hadoop, veja Escalonamento automático com o Hadoop e o Spark.

2. Dada a alteração exata necessária para o número de workers, o escalonamento automático usa um scaleUpFactor ou scaleDownFactor para calcular a alteração real no número de workers:

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

   Um scaleUpFactor ou scaleDownFactor de 1.0 significa que o escalonamento automático será dimensionado para que a memória pendente/disponível seja zero (utilização perfeita).

3. Quando a alteração real no número de workers for calculada, scaleUpMinWorkerFraction e scaleDownMinWorkerFraction age como um limite para determinar se o cluster será escalonado pelo escalonamento automático. Uma pequena fração significa que o escalonamento automático precisa ser escalonado mesmo que o Δworkers seja pequeno. Uma fração maior significa que o escalonamento só pode ocorrer quando o Δworkers é grande.

   if (Δworkers >  scaleUpMinWorkerFraction* cluster size) then scale up

   ou

   if (abs(Δworkers) >  scaleDownMinWorkerFraction* cluster size) 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.

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 máximo da duração que o YARN aguardará pela execução dos aplicativos (o aplicativo que estava em execução durante a desativação) antes como remover nós.

Observação: o escalonamento automático não aumentará durante a desativação otimizada. Ele aguarda a conclusão de uma operação de atualização, incluindo a desativação otimizada, continua até um período de resfriamento (por exemplo, 2 minutos) e emite outra operação de atualização.

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 esteja usando a política, desative o escalonamento automático no cluster antes de atualizá-la.

Comando gcloud

Execute o comando gcloud a seguir 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.

Console

Atualmente, não é possível desativar o escalonamento automático em um cluster no Console do Google Cloud.

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

Recomendações para o escalonamento automático

• Período de resfriamento: o mínimo e o padrão cooldownPeriod são dois minutos. Se um cooldownPeriod mais curto for definido em uma política, as alterações de carga de trabalho afetarão mais rapidamente o tamanho do cluster, mas os clusters podem aumentar e diminuir desnecessariamente. A prática recomendada é definir scale_up, scale_down e min_worker_fractions da política como um valor diferente de zero ao usar um cooldownPeriod mais curto. Isso garante que o cluster só aumente ou diminua quando a alteração na utilização da memória for suficiente para garantir uma atualização do cluster.

• Escalonamento vertical: o MapReduce e o Spark gravam dados intermediários embaralhados no disco local. Remover workers com dados embaralhados atrasará o progresso do job, já que as tarefas de mapa precisarão ser executadas novamente para reproduzir os dados. O Spark executa novamente todos os estágios se detectar que os arquivos aleatórios estão ausentes.

  • Para reduzir apenas quando um cluster estiver inativo, defina scale_down factor e scale_down min_worker_fraction como 1.0.

  • Para clusters com carga contínua, configure o escalonamento horizontal entre jobs definindo o fator scale_down como 1.0 e definindo um graceful_decommission_timeout diferente de zero para remover os nós do cluster quando eles não estiverem executando contêineres YARN (consulte Desativação otimizada). Defina graceful_decommission_timeout para ser mais longo que o job de cluster de longa duração, para garantir que os nós não sejam desativados antes de um job ser concluído.

  • Use o Modo de flexibilidade aprimorado para evitar ou reduzir a chance de perder o progresso do job ao remover nós.

• Jobs do Spark com dados em cache: defina spark.dynamicAllocation.cachedExecutorIdleTimeout ou retire os conjuntos de dados de cache quando não forem mais necessários. Por padrão, o Spark não remove executores que tenham dados armazenados em cache.

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.

A configuração do YARN do Hadoop para agendar jobs com base nas solicitações de memória do YARN é feita no escalonamento automático e não nas solicitações principais do YARN.

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

1. Allocated memory refere-se ao total de memória YARN ocupada pela execução de contêineres em todo o cluster. Se houver 6 contêineres em execução que podem usar até 1 GB, haverá 6 GB de memória alocada.

2. Available memory é a memória YARN no cluster que não é usada pelos contêineres alocados. Se houver 10 GB de memória em todos os gerenciadores de nós e 6 GB de memória alocada, há 4 GB de memória disponível. Se houver memória disponível (não utilizada) no cluster, com o escalonamento automático é possível remover workers do cluster.

3. Pending memory é a soma das solicitações de memória do YARN para contêineres pendentes. Os contêineres pendentes aguardam espaço para serem executados no YARN. A memória pendente só será diferente de zero se a memória disponível for zero ou pequena demais para ser alocada no 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 é concluído (95% por padrão no Dataproc), o MapReduce enfileira solicitações de contêiner para todos os redutores, resultando em outro pico na 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. Com os clusters do Dataproc é feita a alocação dinâmica por padrão, portanto, 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 a alocação dinâmica do Spark 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.

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 sinais específicos de fluxo 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 este último não oferece suporte à alocação dinâmica. Veja SPARK-24815: o Structured Streaming é compatível com a alocação dinâmica.

Como controlar o escalonamento automático com 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:

• Mesmo que o Dataproc 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.

Configurações de propriedades padrão de escalonamento automático do Spark/Hadoop

Os clusters de escalonamento automático têm valores de propriedade de cluster padrão que ajudam a evitar a falha nos jobs, quando os workers principais são removidos ou quando os workers secundários são preteridos. É 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

Padrões para fazer com que o mecanismo de alocação dinâmica do Spark de inicialização lenta comece com um tamanho pequeno:

spark:spark.executor.instances=2

Observação: mesmo que você encontre falhas de busca aleatória quando os workers principais forem removidos ou os secundários forem preemptivos, não tente aumentar as tentativas de busca aleatória acima dos valores padrão: spark.shuffle.io.maxRetries=3; spark.shuffle.io.retryWaitMs=5000 (5 segundos) para um valor total de duração de repetição máxima de 10 segundos. Isso porque, uma vez que os arquivos do embaralhados são perdidos, novas tentativas apenas consomem mais tempo.

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;
• ver o número de administradores de nodes no cluster;
• entender porque o escalonamento automático fez ou não o escalonamento do cluster.

Cloud Logging

Use o Cloud Logging para ver os registros do Cloud Dataproc Autoscaler.

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.

Perguntas frequentes

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

O escalonamento automático pode ser ativado em clusters de alta disponibilidade, mas não em clusters de nó único, porque estes não são compatíveis com 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 atualmente usam a política (consulte Uso da política de vários clusters).

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

Como receber suporte do Dataproc.

Quais versões de imagens são compatíveis com o escalonamento automático? Quais são as versões da API?

O escalonamento automático é compatível com as API v1 em versões de imagem de cluster 1.0.99+, 1.1.90+, 1.2.22+, 1.3.0+ e 1.4.0+ (consulte o Lista de versões do Cloud Dataproc ) e comandos gcloud dataproc autoscaling-policies.

Qual é a diferença entre o Dataproc e o escalonamento automático do Dataflow?

Veja Comparação entre o escalonamento automático do Cloud Dataflow, do Spark e do Hadoop