Usar GPUs com o Google Cloud Serverless para Apache Spark

É possível anexar aceleradores de GPU às cargas de trabalho em lote do Google Cloud Serverless para Apache Spark para alcançar os seguintes resultados:

  • Acelere o processamento de cargas de trabalho de análise de dados em grande escala.

  • Acelere treinamento de modelo em grandes conjuntos de dados usando bibliotecas de machine learning de GPU.

  • Realizar análises de dados avançadas, como processamento de vídeo ou linguagem natural.

Todos os runtimes do Spark sem servidor para Apache Spark compatíveis adicionam a biblioteca RAPIDS do Spark a cada nó de carga de trabalho. A versão 1.1 do ambiente de execução do Spark sem servidor para Apache Spark também adiciona a biblioteca XGBoost aos nós de carga de trabalho. Essas bibliotecas oferecem ferramentas eficientes de transformação de dados e machine learning que podem ser usadas em cargas de trabalho aceleradas por GPU.

Benefícios da GPU

Confira alguns dos benefícios de usar GPUs com suas cargas de trabalho do Serverless para Apache Spark:

  • Melhoria de performance:a aceleração de GPU pode aumentar significativamente a performance das cargas de trabalho do Spark, principalmente para tarefas que exigem computação intensiva, como machine learning e aprendizado profundo, processamento de gráficos e análises complexas.

  • Treinamento de treinamento de modelo mais rápido:para tarefas de machine learning, anexar GPUs pode reduzir drasticamente o tempo necessário para treinar modelos, permitindo que cientistas e engenheiros de dados iterem e façam experimentos rapidamente.

  • Escalonabilidade:os clientes podem adicionar mais nós de GPU ou GPUs mais potentes aos nós para lidar com necessidades de processamento cada vez mais complexas.

  • Eficiência de custo:embora as GPUs exijam um investimento inicial, é possível economizar custos ao longo do tempo devido à redução dos tempos de processamento e ao uso mais eficiente dos recursos.

  • Análise de dados avançada:a aceleração de GPU permite realizar análises avançadas, como análise de imagens e vídeos e processamento de linguagem natural, em grandes conjuntos de dados.

  • Produtos aprimorados:o processamento mais rápido permite tomar decisões mais rapidamente e ter aplicativos mais responsivos.

Limitações e considerações

Preços

Consulte Preços do Serverless para Apache Spark para informações sobre preços de aceleradores.

Antes de começar

Antes de criar uma carga de trabalho em lote sem servidor com aceleradores de GPU anexados, faça o seguinte:

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

  5. Install the Google Cloud CLI.

  6. Se você estiver usando um provedor de identidade externo (IdP), primeiro faça login na CLI gcloud com sua identidade federada.

  7. Para inicializar a CLI gcloud, execute o seguinte comando: 

    gcloud init

  8. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  9. Verify that billing is enabled for your Google Cloud project.

  10. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

  11. Install the Google Cloud CLI.

  12. Se você estiver usando um provedor de identidade externo (IdP), primeiro faça login na CLI gcloud com sua identidade federada.

  13. Para inicializar a CLI gcloud, execute o seguinte comando: 

    gcloud init
  14. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets

  15. Click Create.
  16. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
    1. In the Get started section, do the following:
      • Enter a globally unique name that meets the bucket naming requirements.
      • To add a bucket label, expand the Labels section (), click Add label, and specify a key and a value for your label.
    2. In the Choose where to store your data section, do the following:
      1. Select a Location type.
      2. Choose a location where your bucket's data is permanently stored from the Location type drop-down menu.
      3. To set up cross-bucket replication, select Add cross-bucket replication via Storage Transfer Service and follow these steps:

        Set up cross-bucket replication

        1. In the Bucket menu, select a bucket.

        2. In the Replication settings section, click Configure to configure settings for the replication job.

          The Configure cross-bucket replication pane appears.

          • To filter objects to replicate by object name prefix, enter a prefix that you want to include or exclude objects from, then click Add a prefix.
          • To set a storage class for the replicated objects, select a storage class from the Storage class menu. If you skip this step, the replicated objects will use the destination bucket's storage class by default.
          • Click Done.
    3. In the Choose how to store your data section, do the following:
      1. Select a default storage class for the bucket or Autoclass for automatic storage class management of your bucket's data.
      2. To enable hierarchical namespace, in the Optimize storage for data-intensive workloads section, select Enable hierarchical namespace on this bucket.
    4. In the Choose how to control access to objects section, select whether or not your bucket enforces public access prevention, and select an access control method for your bucket's objects.
    5. In the Choose how to protect object data section, do the following:
      • Select any of the options under Data protection that you want to set for your bucket.
        • To enable soft delete, click the Soft delete policy (For data recovery) checkbox, and specify the number of days you want to retain objects after deletion.
        • To set Object Versioning, click the Object versioning (For version control) checkbox, and specify the maximum number of versions per object and the number of days after which the noncurrent versions expire.
        • To enable the retention policy on objects and buckets, click the Retention (For compliance) checkbox, and then do the following:
          • To enable Object Retention Lock, click the Enable object retention checkbox.
          • To enable Bucket Lock, click the Set bucket retention policy checkbox, and choose a unit of time and a length of time for your retention period.
      • To choose how your object data will be encrypted, expand the Data encryption section (), and select a Data encryption method.
  17. Click Create.

    18. Criar uma carga de trabalho em lote sem servidor com aceleradores de GPU

    Envie uma carga de trabalho em lote do Serverless para Apache Spark que use GPUs NVIDIA L4 para executar uma tarefa paralela do PySpark. Siga estas etapas usando a CLI gcloud:

    1. Clique em Expandir e crie e salve o código PySpark listado em um arquivo test-py-spark-gpu.py na máquina local usando um editor de texto ou de código.

      #!/usr/bin/env python

"""S8s Accelerators Example."""

import subprocess
from typing import Any
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import IntegerType
from pyspark.sql.types import StructField
from pyspark.sql.types import StructType

spark = SparkSession.builder.appName("joindemo").getOrCreate()


def get_num_gpus(_: Any) -> int:
  """Returns the number of GPUs."""
  p_nvidia_smi = subprocess.Popen(
      ["nvidia-smi", "-L"], stdin=None, stdout=subprocess.PIPE
  )
  p_wc = subprocess.Popen(
      ["wc", "-l"],
      stdin=p_nvidia_smi.stdout,
      stdout=subprocess.PIPE,
      stderr=subprocess.PIPE,
      universal_newlines=True,
  )
  [out, _] = p_wc.communicate()
  return int(out)


num_workers = 5
result = (
    spark.sparkContext.range(0, num_workers, 1, num_workers)
    .map(get_num_gpus)
    .collect()
)
num_gpus = sum(result)
print(f"Total accelerators: {num_gpus}")

# Run the join example
schema = StructType([StructField("value", IntegerType(), True)])
df = (
    spark.sparkContext.parallelize(range(1, 10000001), 6)
    .map(lambda x: (x,))
    .toDF(schema)
)
df2 = (
    spark.sparkContext.parallelize(range(1, 10000001), 6)
    .map(lambda x: (x,))
    .toDF(schema)
)
joined_df = (
    df.select(col("value").alias("a"))
    .join(df2.select(col("value").alias("b")), col("a") == col("b"))
    .explain()
)

    2. Use a CLI gcloud na sua máquina local para enviar o job em lote sem servidor do Serverless para Apache Spark com cinco workers, cada um acelerado com GPUs L4:

      gcloud dataproc batches submit pyspark test-py-spark-gpu.py \
    --project=PROJECT_ID \
    --region=REGION \
    --deps-bucket=BUCKET_NAME \
    --version=1.1 \
    --properties=spark.dataproc.executor.compute.tier=premium,spark.dataproc.executor.disk.tier=premium,spark.dataproc.executor.resource.accelerator.type=l4,spark.executor.instances=5,spark.dataproc.driverEnv.LANG=C.UTF-8,spark.executorEnv.LANG=C.UTF-8,spark.shuffle.manager=com.nvidia.spark.rapids.RapidsShuffleManager

    Observações:

    • PROJECT_ID: o ID do projeto do Google Cloud .
    • REGION: uma região do Compute Engine disponível para executar a carga de trabalho.
    • BUCKET_NAME: o nome do bucket do Cloud Storage. O Spark faz upload das dependências da carga de trabalho para uma pasta /dependencies nesse bucket antes de executar a carga de trabalho em lote.
    • --version::todos os runtimes do Serverless para Apache Spark compatíveis Google Cloud adicionam a biblioteca RAPIDS a cada nó de uma carga de trabalho acelerada por GPU. Somente a versão 1.1 do ambiente de execução adiciona a biblioteca XGBoost a cada nó de uma carga de trabalho acelerada por GPU.

    • --properties (consulte Propriedades de alocação de recursos do Spark) :

      • spark.dataproc.driverEnv.LANG=C.UTF-8 e spark.executorEnv.LANG=C.UTF-8 (obrigatório com versões de tempo de execução anteriores a 2.2): essas propriedades definem o conjunto de caracteres padrão como C.UTF-8.

      • spark.dataproc.executor.compute.tier=premium (obrigatório): as cargas de trabalho aceleradas por GPU são faturadas usando unidades de computação de dados (DCUs) premium. Consulte os preços dos aceleradores do Serverless para Apache Spark.

      • spark.dataproc.executor.disk.tier=premium (obrigatório): nós com aceleradores A100-40, A100-80 ou L4 precisam usar o nível de disco premium.

      • spark.dataproc.executor.resource.accelerator.type=l4 (obrigatório): apenas um tipo de GPU precisa ser especificado. O job de exemplo seleciona a GPU L4. Os seguintes tipos de acelerador podem ser especificados com os seguintes nomes de argumentos:

        Tipo de GPU Nome do argumento
        A100 40 GB a100-40
        A100 80 GB a100-80

      • spark.executor.instances=5 (obrigatório): é necessário indicar, pelo menos, dois. Defina como cinco para este exemplo.

      • spark.executor.cores (opcional): defina essa propriedade para especificar o número de vCPUs principais. Os valores válidos para GPUs L4 são 4 (padrão) ou 8, 12, 16, 24, 48 ou 96. O único valor válido e padrão para GPUs A100 é 12. As configurações com GPUs L4 e núcleos 24, 48 ou 96 têm GPUs 2, 4 ou 8 anexadas a cada executor. Todas as outras configurações têm uma GPU 1 anexada.

      • spark.dataproc.executor.disk.size (obrigatório): as GPUs L4 têm um tamanho de disco fixo de 375 GB, exceto para configurações com núcleos 24, 48 ou 96, que têm 750, 1,500 ou 3,000 GB, respectivamente. Se você definir essa propriedade com um valor diferente ao enviar uma carga de trabalho acelerada por L4, um erro vai ocorrer. Se você selecionar uma GPU A100 40 ou A100 80, os tamanhos válidos serão 375g, 750g, 1500g, 3000g, 6000g e 9000g.

      • spark.executor.memory (opcional) e spark.executor.memoryOverhead (opcional): você pode definir uma dessas propriedades, mas não ambas. A quantidade de memória disponível não consumida pela propriedade definida é aplicada à propriedade não definida. Por padrão, spark.executor.memoryOverhead é definido como 40% da memória disponível para cargas de trabalho em lote do PySpark e 10% para outras cargas de trabalho. Consulte Propriedades de alocação de recursos do Spark.

        A tabela a seguir mostra a quantidade máxima de memória que pode ser definida para diferentes configurações de GPU A100 e L4. O valor mínimo para qualquer uma das propriedades é 1024 MB.

        A100 (40 GB) A100 (80 GB) L4 (4 núcleos) L4 (8 núcleos) L4 (12 núcleos) L4 (16 núcleos) L4 (24 núcleos) L4 (48 núcleos) L4 (96 núcleos)
        Memória total máxima (MB) 78040 165080 13384 26768 40152 53536 113072 160608 321216

      • Propriedades do Spark RAPIDS (opcional): por padrão, o Serverless para Apache Spark define os seguintes valores de propriedade do Spark RAPIDS:

        • spark.plugins=com.nvidia.spark.SQLPlugin
        • spark.executor.resource.gpu.amount=1
        • spark.task.resource.gpu.amount=1/$spark_executor_cores
        • spark.shuffle.manager=''. Por padrão, essa propriedade não é definida. A NVIDIA recomenda ativar o gerenciador de embaralhamento do RAPIDS ao usar GPUs para melhorar a performance. Para fazer isso, defina spark.shuffle.manager=com.nvidia.spark.rapids.RapidsShuffleManager ao enviar uma carga de trabalho.
        • spark.rapids.sql.concurrentGpuTasks= mínimo de (gpuMemoryinMB / 8, 4)
        • spark.rapids.shuffle.multiThreaded.writer.threads= mínimo de (núcleos de CPU na VM / contagem de GPU por VM, 32)
        • spark.rapids.shuffle.multiThreaded.reader.threads= mínimo de (núcleos de CPU na VM / contagem de GPU por VM, 32)

        Consulte Configuração do acelerador RAPIDS para Apache Spark para definir propriedades do Spark RAPIDS e Configuração avançada do acelerador RAPIDS para Apache Spark para definir propriedades avançadas do Spark.