Esegui Spark sul cluster Ray su Gemini Enterprise Agent Platform

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 Agent Platform (cluster Ray su Gemini Enterprise Agent Platform).

Installazione

Ray on Agent Platform consente agli utenti di eseguire le proprie applicazioni utilizzando il framework open source Ray. RayDP fornisce API per l'esecuzione di Spark su Ray. Le immagini container predefinite disponibili per creare un cluster Ray su Gemini Enterprise Agent Platform non includono RayDP preinstallato. Ciò significa che devi creare un'immagine cluster Ray personalizzata su Gemini Enterprise Agent Platform per il tuo cluster Ray su Gemini Enterprise Agent Platform per eseguire applicazioni RayDP sul cluster Ray su Gemini Enterprise Agent Platform. La sezione seguente spiega come creare un'immagine personalizzata RayDP.

Crea un'immagine container personalizzata di Ray su Gemini Enterprise Agent Platform

Utilizza questo Dockerfile per creare un'immagine container personalizzata per Ray su Gemini Enterprise Agent Platform 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 Gemini Enterprise Agent Platform per creare l'immagine personalizzata RayDP. Puoi anche installare altri pacchetti Python che prevedi di utilizzare nelle tue applicazioni. La versione pyarrow==14.0 è dovuta a un vincolo di dipendenza di Ray 2.9.3.

Crea ed esegui il push dell'immagine container personalizzata

Crea un repository Docker in Artifact Registry prima di creare l'immagine personalizzata (consulta Utilizza le immagini container per scoprire come creare e configurare il repository Docker). Dopo aver 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 località di Cloud Storage (ad esempio us-central1) che hai creato in Artifact Registry.
  • PROJECT_ID: l'ID progetto Google Cloud .
  • DOCKER_REPOSITORY: il nome del repository Docker che hai creato.
  • IMAGE_NAME: il nome delle immagini container personalizzate.

Crea un cluster Ray su Gemini Enterprise Agent Platform

Utilizza l'immagine container personalizzata creata nel passaggio precedente per creare un cluster Ray su Gemini Enterprise Agent Platform. Puoi utilizzare l'SDK Agent Platform per Python per creare un cluster Ray su Gemini Enterprise Agent Platform.

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 Agent Platform 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 ad Artifact Registry.
  • CLUSTER_NAME: il nome del cluster Ray su Gemini Enterprise Agent Platform.

Spark su cluster Ray su Gemini Enterprise Agent Platform

Prima di eseguire l'applicazione Spark, crea una sessione Spark utilizzando l'API RayDP. Per farlo in modo interattivo, puoi utilizzare il client Ray o 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, oltre a supportare la configurazione di Spark. Scopri di più sull'API RayDP per la creazione di sessioni Spark, consulta Affinità dei nodi degli attori master Spark.

RayDP con client Ray

Puoi utilizzare Task o Actor di Ray per creare un cluster e una sessione Spark sul cluster Ray su Gemini Enterprise Agent Platform. Per utilizzare un client Ray per creare una sessione Spark sul cluster Ray su Gemini Enterprise Agent Platform, è necessario un attore o un'attività Ray. Il seguente codice mostra come un attore Ray può creare una sessione Spark, eseguire un'applicazione Spark e arrestare un cluster Spark su un cluster Ray su Gemini Enterprise Agent Platform utilizzando RayDP.

Per connetterti in modo interattivo al cluster Ray su Gemini Enterprise Agent Platform, consulta Connettiti a un cluster Ray tramite il client Ray

@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 Gemini Enterprise Agent Platform.

Crea uno script Python contenente 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 di Cloud Storage dall'applicazione Spark

È prassi comune archiviare i file di dati in un bucket Cloud Storage. Puoi leggere questi file in diversi modi da un'applicazione Spark in esecuzione sul cluster Ray su Gemini Enterprise Agent Platform. Questa sezione spiega due tecniche per leggere i file di Cloud Storage dalle applicazioni Spark in esecuzione sul cluster Ray su Gemini Enterprise Agent Platform.

Utilizza il connettore Cloud Storage

Puoi utilizzare il connettore Google Cloud per Hadoop per leggere i file da un bucket Cloud Storage dalla tua applicazione Spark. Dopo aver creato una sessione Spark utilizzando RayDP, puoi leggere i file utilizzando alcuni parametri di configurazione. Il seguente codice mostra come leggere un file CSV archiviato in un bucket Cloud Storage da un'applicazione Spark sul cluster Ray su Gemini Enterprise Agent Platform.

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.

Utilizza i dati Ray

Il connettore Google Cloud fornisce un modo per leggere i file da un bucket Google Cloude potrebbe essere sufficiente per la maggior parte dei casi d'uso. Potresti voler utilizzare Ray Data per leggere i file dal bucket Google Cloud 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 Cloud . Questo potrebbe accadere a causa di conflitti di dipendenza Java quando alcune altre dipendenze dell'applicazione vengono aggiunte al classpath Java di Spark utilizzando spark.jars.packages o spark.jars.

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 PySpark Pandas su cluster Ray su Gemini Enterprise Agent Platform

Le funzioni definite dall'utente di PySpark Pandas a volte richiedono codice aggiuntivo quando le utilizzi nella tua applicazione Spark in esecuzione su un cluster Ray su Gemini Enterprise Agent Platform. In genere, questo è necessario quando la funzione definita dall'utente Pandas utilizza una libreria Python non disponibile nel cluster Ray su Gemini Enterprise Agent Platform. Puoi pacchettizzare le dipendenze Python di un'applicazione utilizzando l'ambiente di runtime con l'API Ray Job. Dopo aver inviato il job Ray al cluster, Ray installa queste dipendenze nell'ambiente virtuale Python che crea per l'esecuzione del job. Le UDF Pandas, invece, non utilizzano lo stesso ambiente virtuale. Utilizzano invece l'ambiente di sistema Python predefinito. Se questa dipendenza non è disponibile nell'ambiente di sistema, potresti doverla installare all'interno della funzione definita dall'utente di Pandas. Nell'esempio seguente, installa la libreria statsmodels all'interno della funzione definita dall'utente.

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()