Menjalankan Spark di cluster Ray di Gemini Enterprise Agent Platform

Library Python RayDP memungkinkan menjalankan Spark di cluster Ray. Dokumen ini membahas penginstalan, konfigurasi, dan menjalankan RayDP di Ray di Agent Platform (cluster Ray di Gemini Enterprise Agent Platform).

Penginstalan

Ray di Agent Platform memungkinkan pengguna menjalankan aplikasi mereka menggunakan open source framework Ray. RayDP menyediakan API untuk menjalankan Spark di Ray. Image container bawaan yang tersedia untuk membuat cluster Ray di Gemini Enterprise Agent Platform tidak dilengkapi dengan RayDP yang telah diinstal sebelumnya. Artinya, Anda harus membuat image cluster Ray kustom di Gemini Enterprise Agent Platform agar cluster Ray di Gemini Enterprise Agent Platform dapat menjalankan aplikasi RayDP di cluster Ray di Gemini Enterprise Agent Platform. Bagian berikut menjelaskan cara membuat image kustom RayDP.

Membangun image container kustom Ray di Gemini Enterprise Agent Platform

Gunakan dockerfile ini untuk membuat image container kustom untuk Ray di Gemini Enterprise Agent Platform yang telah menginstal RayDP.

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

Anda dapat menggunakan image bawaan cluster Ray terbaru di Gemini Enterprise Agent Platform untuk membuat image kustom RayDP. Anda juga dapat menginstal paket Python lainnya yang diperkirakan akan digunakan dalam aplikasi. pyarrow==14.0 disebabkan oleh batasan dependensi Ray 2.9.3.

Membangun dan mengirim image container kustom

Buat repositori Docker di Artifact Registry sebelum Anda membuat image kustom (lihat Menggunakan image container untuk mengetahui cara membuat dan mengonfigurasi repositori Docker). Setelah membuat repositori Docker, bangun dan kirim image container kustom menggunakan 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]

Dengan:

  • LOCATION: Lokasi Cloud Storage (misalnya, us-central1) yang Anda buat di Artifact Registry.
  • PROJECT_ID: Project ID Anda Google Cloud .
  • DOCKER_REPOSITORY: Nama repositori Docker yang Anda buat.
  • IMAGE_NAME: Nama image container kustom Anda.

Membuat cluster Ray di Gemini Enterprise Agent Platform

Gunakan image container kustom yang dibuat pada langkah sebelumnya untuk membuat cluster Ray di Gemini Enterprise Agent Platform. Anda dapat menggunakan Agent Platform SDK untuk Python guna membuat cluster Ray di Gemini Enterprise Agent Platform.

Jika belum melakukannya, instal library Python yang diperlukan.

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

Konfigurasi node Head dan Worker, lalu buat cluster menggunakan Agent Platform SDK untuk Python. Contoh:

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],
)

Dengan:

  • CUSTOM_CONTAINER_IMAGE_URI: URI image container kustom yang dikirim ke Artifact Registry.
  • CLUSTER_NAME: Nama cluster Ray di Gemini Enterprise Agent Platform.

Cluster Spark di Ray di Gemini Enterprise Agent Platform

Sebelum menjalankan aplikasi Spark, buat sesi Spark menggunakan RayDP API. Anda dapat menggunakan klien Ray untuk melakukannya secara interaktif atau menggunakan Ray Job API. Ray Job API direkomendasikan, terutama untuk aplikasi produksi dan aplikasi yang berjalan lama. RayDP API menyediakan parameter untuk mengonfigurasi sesi Spark, serta mendukung Konfigurasi Spark. Pelajari lebih lanjut RayDP API untuk membuat Sesi Spark, lihat Afinitas node aktor master Spark.

RayDP dengan klien Ray

Anda dapat menggunakan Ray Tugas atau Aktor untuk membuat cluster dan sesi Spark di cluster Ray di Gemini Enterprise Agent Platform. Tugas Ray, atau Aktor, diperlukan untuk menggunakan Klien Ray guna membuat sesi Spark di cluster Ray di Gemini Enterprise Agent Platform. Kode berikut menunjukkan cara Aktor Ray dapat membuat Sesi Spark, menjalankan aplikasi Spark, dan menghentikan cluster Spark di cluster Ray di Gemini Enterprise Agent Platform menggunakan RayDP.

Untuk terhubung secara interaktif ke cluster Ray di Gemini Enterprise Agent Platform, lihat Menghubungkan ke cluster Ray melalui Klien 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 dengan Ray Job API

Klien Ray berguna untuk eksperimen kecil yang memerlukan koneksi interaktif dengan cluster Ray. Ray Job API adalah cara yang direkomendasikan untuk menjalankan tugas produksi dan tugas yang berjalan lama di cluster Ray. Hal ini juga berlaku untuk menjalankan aplikasi Spark di cluster Ray di Gemini Enterprise Agent Platform.

Buat skrip Python yang berisi kode aplikasi Spark. Contoh:

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

Kirim tugas untuk menjalankan skrip Python menggunakan Ray Job API. Contoh:

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": ".",
  }
)

Dengan:

  • SCRIPT_NAME: Nama file skrip yang Anda buat.

Membaca file Cloud Storage dari aplikasi Spark

Menyimpan file data di bucket Google Cloud Storage adalah praktik yang umum dilakukan. Anda dapat membaca file ini dengan berbagai cara dari aplikasi Spark yang berjalan di cluster Ray di Gemini Enterprise Agent Platform. Bagian ini menjelaskan dua teknik untuk membaca file Cloud Storage dari aplikasi Spark yang berjalan di Cluster Ray di Gemini Enterprise Agent Platform.

Menggunakan Konektor Google Cloud Storage

Anda dapat menggunakan Google Cloud Konektor untuk Hadoop guna membaca file dari bucket Cloud Storage dari aplikasi Spark. Setelah membuat sesi Spark menggunakan RayDP, Anda dapat membaca file menggunakan beberapa parameter konfigurasi. Kode berikut menunjukkan cara membaca file CSV yang disimpan di bucket Cloud Storage dari aplikasi Spark di cluster Ray di 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)

Dengan:

  • GCS_FILE_URI: URI file yang disimpan di bucket Cloud Storage. Contoh: gs://my-bucket/my-file.csv.

Menggunakan data Ray

Konektor menyediakan cara untuk membaca file dari Google Cloud bucket dan mungkin cukup untuk sebagian besar kasus penggunaan. Google Cloud Anda mungkin ingin menggunakan Data Ray untuk membaca file dari Google Cloud bucket saat perlu menggunakan pemrosesan terdistribusi Ray untuk membaca data, atau saat Anda mengalami masalah saat membaca Google Cloud file dengan Google Cloud konektor. Hal ini mungkin terjadi karena konflik dependensi Java saat beberapa dependensi aplikasi lainnya ditambahkan ke classpath Java Spark menggunakan spark.jars.packages atau 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 di cluster Ray di Gemini Enterprise Agent Platform

UDF Pyspark Pandas terkadang memerlukan kode tambahan saat Anda menggunakannya di aplikasi Spark yang berjalan di cluster Ray di Gemini Enterprise Agent Platform. Hal ini biasanya diperlukan saat UDF Pandas menggunakan library Python yang tidak tersedia di cluster Ray di Gemini Enterprise Agent Platform. Anda dapat mengemas the dependensi Python aplikasi menggunakan lingkungan runtime dengan Ray Job API. Setelah Anda mengirimkan tugas Ray ke cluster, Ray akan menginstal dependensi tersebut di lingkungan virtual Python yang dibuat untuk menjalankan tugas. Namun, UDF Pandas tidak menggunakan lingkungan virtual yang sama. Sebagai gantinya, UDF Pandas menggunakan lingkungan Sistem Python default. Jika dependensi tersebut tidak tersedia di lingkungan Sistem, Anda mungkin perlu menginstalnya dalam UDF Pandas. Dalam contoh berikut, instal library statsmodels dalam 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()