Esegui Spark su un cluster Ray su Vertex AI

La libreria Python RayDP consente di eseguire Spark su un cluster Ray. Questo documento illustra l'installazione, la configurazione e l'esecuzione di RayDP su Ray on Vertex AI (cluster Ray su Vertex AI).

Installazione

Ray su Vertex AI consente agli utenti di eseguire le proprie applicazioni utilizzando il framework Ray open source. RayDP fornisce API per l'esecuzione di Spark su Ray. Le immagini container predefinite disponibili per creare un cluster Ray su Vertex AI non vengono fornite con RayDP preinstallato, il che significa che devi creare un'immagine personalizzata del cluster Ray su Vertex AI affinché il cluster Ray su Vertex AI esegua le applicazioni RayDP sul cluster Ray su Vertex AI. La sezione seguente spiega come creare un'immagine personalizzata RayDP.

Crea un'immagine container personalizzata di Ray su Vertex AI

Utilizza questo Dockerfile per creare un'immagine container personalizzata per Ray su Vertex AI con RayDP installato.

FROM us-docker.pkg.dev/vertex-ai/training/ray-cpu.2-9.py310:latest

RUN apt-get update -y \
    && pip install --no-cache-dir raydp pyarrow==14.0

Puoi utilizzare l'ultima immagine predefinita del cluster Ray su Vertex AI per creare l'immagine personalizzata RayDP. Puoi anche installare altri pacchetti Python che prevedi di utilizzare nelle tue applicazioni. pyarrow==14.0 è dovuto a un vincolo di dipendenza di Ray 2.42.0.

Crea ed esegui il push dell'immagine container personalizzata

Devi creare un repository Docker in Artifact Registry prima di poter creare la tua immagine personalizzata (consulta la sezione Utilizzare le immagini container per scoprire come creare e configurare il repository Docker). Una volta creato il repository Docker, crea ed esegui il push dell'immagine container personalizzata utilizzando il Dockerfile.

docker build . -t [LOCATION]-docker.pkg.dev/[PROJECT_ID]/[DOCKER_REPOSITORY]/[IMAGE_NAME]
docker push [LOCATION]-docker.pkg.dev/[PROJECT_ID]/[DOCKER_REPOSITORY]/[IMAGE_NAME]

Dove:

  • LOCATION: la posizione Cloud Storage (ad esempio us-central1) che hai creato in Artifact Registry.
  • PROJECT_ID: il tuo ID progetto Google Cloud .
  • DOCKER_REPOSITORY: il nome del repository Docker che hai creato.
  • IMAGE_NAME: Il nome delle tue immagini container personalizzate.

Crea un cluster Ray su Vertex AI

Utilizza l'immagine container personalizzata creata nel passaggio precedente per creare un cluster Ray su Vertex AI. Puoi utilizzare l'SDK Vertex AI per Python per creare un cluster Ray su Vertex AI.

Se non l'hai ancora fatto,installa le librerie Python richieste.

pip install --quiet google-cloud-aiplatform \
             ray[all]==2.9.3 \
             google-cloud-aiplatform[ray]

Configura i nodi head e worker e crea il cluster utilizzando l'SDK Vertex AI per Python. Ad esempio:

import logging
import ray
from google.cloud import aiplatform
from google.cloud.aiplatform import vertex_ray
from vertex_ray import Resources

head_node_type = Resources(
    machine_type="n1-standard-16",
    node_count=1,
    custom_image=[CUSTOM_CONTAINER_IMAGE_URI],
)

worker_node_types = [Resources(
    machine_type="n1-standard-8",
    node_count=2,
    custom_image=[CUSTOM_CONTAINER_IMAGE_URI],
)]

ray_cluster_resource_name = vertex_ray.create_ray_cluster(
    head_node_type=head_node_type,
    worker_node_types=worker_node_types,
    cluster_name=[CLUSTER_NAME],
)

Dove:

  • CUSTOM_CONTAINER_IMAGE_URI: l'URI dell'immagine container personalizzata di cui è stato eseguito il push in Artifact Registry.
  • CLUSTER_NAME: il nome del tuo cluster Ray su Vertex AI.

Cluster Spark su Ray su Vertex AI

Prima di poter eseguire l'applicazione Spark, devi creare una sessione Spark utilizzando l'API RayDP. Puoi utilizzare il client Ray per eseguire questa operazione in modo interattivo o utilizzare l'API Ray Job. L'API Ray Job è consigliata, soprattutto per le applicazioni di produzione e a lunga esecuzione. L'API RayDP fornisce parametri per configurare la sessione Spark e supporta la configurazione Spark. Scopri di più sull'API RayDP per la creazione di sessioni Spark in Affinità dei nodi degli attori master Spark.

RayDP con il client Ray

Puoi utilizzare Task o Actor di Ray per creare un cluster e una sessione Spark sul cluster Ray su Vertex AI. Ray Task o Actor è necessario per utilizzare un client Ray per creare una sessione Spark sul cluster Ray su Vertex AI. Il seguente codice mostra come utilizzare un attore Ray per creare una sessione Spark, eseguire un'applicazione Spark e arrestare un cluster Spark su un cluster Ray su Vertex AI utilizzando RayDP.

Per informazioni su come connetterti in modo interattivo al cluster Ray su Vertex AI, consulta Connettersi a un cluster Ray tramite Ray Client.

@ray.remote
class SparkExecutor:
  import pyspark

  spark: pyspark.sql.SparkSession = None

  def __init__(self):

    import ray
    import raydp

    self.spark = raydp.init_spark(
      app_name="RAYDP ACTOR EXAMPLE",
      num_executors=1,
      executor_cores=1,
      executor_memory="500M",
    )

  def get_data(self):
    df = self.spark.createDataFrame(
        [
            ("sue", 32),
            ("li", 3),
            ("bob", 75),
            ("heo", 13),
        ],
        ["first_name", "age"],
    )
    return df.toJSON().collect()

  def stop_spark(self):
    import raydp
    raydp.stop_spark()

s = SparkExecutor.remote()
data = ray.get(s.get_data.remote())
print(data)
ray.get(s.stop_spark.remote())

RayDP con l'API Ray Job

Il client Ray è utile per piccoli esperimenti che richiedono una connessione interattiva con il cluster Ray. L'API Ray Job è il modo consigliato per eseguire job di produzione e di lunga durata su un cluster Ray. Questo vale anche per l'esecuzione di applicazioni Spark sul cluster Ray su Vertex AI.

Crea uno script Python che contenga il codice dell'applicazione Spark. Ad esempio:

import pyspark
import raydp

def get_data(spark: pyspark.sql.SparkSession):
    df = spark.createDataFrame(
        [
            ("sue", 32),
            ("li", 3),
            ("bob", 75),
            ("heo", 13),
        ],
        ["first_name", "age"],
    )
    return df.toJSON().collect()

def stop_spark():
    raydp.stop_spark()

if __name__ == '__main__':
    spark = raydp.init_spark(
      app_name="RAYDP JOB EXAMPLE",
        num_executors=1,
        executor_cores=1,
        executor_memory="500M",
    )
    print(get_data(spark))
    stop_spark()

Invia il job per eseguire lo script Python utilizzando l'API Ray Job. Ad esempio:

from ray.job_submission import JobSubmissionClient

client = JobSubmissionClient(RAY_ADDRESS)

job_id = client.submit_job(
  # Entrypoint shell command to execute
  entrypoint="python [SCRIPT_NAME].py",
  # Path to the local directory that contains the python script file.
  runtime_env={
    "working_dir": ".",
  }
)

Dove:

  • SCRIPT_NAME: il nome file dello script che hai creato.

Lettura dei file Cloud Storage dall'applicazione Spark

È prassi comune archiviare i file di dati in un bucket Google Cloud Storage. Esistono diversi modi per leggere questi file da un'applicazione Spark in esecuzione sul cluster Ray su Vertex AI. Questa sezione illustra due tecniche per leggere i file Cloud Storage dalle applicazioni Spark in esecuzione su Ray Cluster su Vertex AI.

Utilizzare il connettore Google Cloud Storage

Puoi utilizzare il Google Cloud Connector per Hadoop per leggere i file da un bucket Cloud Storage dalla tua applicazione Spark. Questa operazione viene eseguita utilizzando alcuni parametri di configurazione quando viene creata una sessione Spark utilizzando RayDP. Il seguente codice mostra come un file CSV archiviato in un bucket Cloud Storage può essere letto da un'applicazione Spark sul cluster Ray su Vertex AI.

import raydp

spark = raydp.init_spark(
  app_name="RayDP Cloud Storage Example 1",
  configs={
      "spark.jars": "https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop3-2.2.22.jar",
      "spark.hadoop.fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS",
      "spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
  },
  num_executors=2,
  executor_cores=4,
  executor_memory="500M",
)

spark_df = spark.read.csv([GCS_FILE_URI], header = True, inferSchema = True)

Dove:

  • GCS_FILE_URI: l'URI di un file archiviato in un bucket Cloud Storage. Ad esempio: gs://my-bucket/my-file.csv.

Utilizzare i dati Ray

Il Google Cloud connettore fornisce un modo per leggere i file da un bucket Google Cloud e potrebbe essere sufficiente per la maggior parte dei casi d'uso. Potresti voler utilizzare Ray Data per leggere i file dal bucket quando devi utilizzare l'elaborazione distribuita di Ray per leggere i dati o quando riscontri problemi di lettura del fileGoogle Cloud con il connettore Google Google Cloud , il che potrebbe verificarsi a causa di conflitti di dipendenze Java quando vengono aggiunte altre dipendenze dell'applicazione al classpath Java di Spark utilizzando spark.jars.packages o spark.jars. Google Cloud 

import raydp
import ray

spark = raydp.init_spark(
  app_name="RayDP Cloud Storage Example 2",
  configs={
      "spark.jars.packages": "com.microsoft.azure:synapseml_2.12:0.11.4-spark3.3",
      "spark.jars.repositories": "https://mmlspark.azureedge.net/maven",
      "spark.jars": "https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop3-2.2.22.jar",
      "spark.hadoop.fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS",
      "spark.hadoop.fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
  },
  num_executors=2,
  executor_cores=4,
  executor_memory="500M",
)

# This doesn't work even though the Cloud Storage connector Jar and other parameters have
been added to the Spark configuration.
#spark.read.csv([GCS_FILE_URI], header = True, inferSchema = True)

ray_dataset = ray.data.read_csv(GCS_FILE_URI)
spark_df = ray_dataset.to_spark(spark)

UDF Pandas Pyspark sul cluster Ray su Vertex AI

Le UDF Pandas di PySpark a volte potrebbero richiedere codice aggiuntivo quando le utilizzi nell'applicazione Spark in esecuzione su un cluster Ray su Vertex AI. Questo è in genere necessario quando la UDF Pandas utilizza una libreria Python non disponibile sul cluster Ray su Vertex AI. È possibile creare pacchetti delle dipendenze Python di un'applicazione utilizzando l'ambiente di runtime con l'API Ray Job e quando il job Ray viene inviato al cluster, Ray installa queste dipendenze nell'ambiente virtuale Python che crea per l'esecuzione del job. Le UDF Pandas, tuttavia, non utilizzano lo stesso ambiente virtuale. Utilizzano invece l'ambiente Python System predefinito. Se questa dipendenza non è disponibile nell'ambiente di sistema, potrebbe essere necessario installarla all'interno della funzione definita dall'utente Pandas. Nell'esempio seguente, la libreria statsmodels deve essere installata all'interno della UDF.

import pandas as pd
import pyspark
import raydp
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType

def test_udf(spark: pyspark.sql.SparkSession):
    import pandas as pd
    
    df = spark.createDataFrame(pd.read_csv("https://www.datavis.ca/gallery/guerry/guerry.csv"))
    return df.select(func('Lottery','Literacy', 'Pop1831')).collect()

@pandas_udf(StringType())
def func(s1: pd.Series, s2: pd.Series, s3: pd.Series) -> str:
    import numpy as np
    import subprocess
    import sys
    subprocess.check_call([sys.executable, "-m", "pip", "install", "statsmodels"])
    import statsmodels.api as sm
    import statsmodels.formula.api as smf
    
    d = {'Lottery': s1, 
         'Literacy': s2,
         'Pop1831': s3}
    data = pd.DataFrame(d)

    # Fit regression model (using the natural log of one of the regressors)
    results = smf.ols('Lottery ~ Literacy + np.log(Pop1831)', data=data).fit()
    return results.summary().as_csv()

if __name__ == '__main__':
    
    spark = raydp.init_spark(
      app_name="RayDP UDF Example",
      num_executors=2,
      executor_cores=4,
      executor_memory="1500M",
    )
    
    print(test_udf(spark))
    
    raydp.stop_spark()