הרצת Spark באשכול Ray ב-Gemini Enterprise Agent Platform

ספריית Python‏ RayDP מאפשרת להריץ Spark באשכול Ray. במאמר הזה מוסבר איך להתקין, להגדיר ולהפעיל את RayDP ב-Ray ב-Agent Platform (אשכול Ray ב-Gemini Enterprise Agent Platform).

התקנה

‫Ray on Agent Platform מאפשר למשתמשים להריץ את האפליקציות שלהם באמצעות מסגרת Ray בקוד פתוח. ‫RayDP מספק ממשקי API להפעלת Spark ב-Ray. קובצי האימג' בקונטיינר שזמינים ליצירת אשכול Ray ב-Gemini Enterprise Agent Platform לא כוללים את RayDP. כלומר, אתם צריכים ליצור תמונה של אשכול Ray מותאם אישית ב-Gemini Enterprise Agent Platform כדי שאשכול Ray ב-Gemini Enterprise Agent Platform יוכל להריץ אפליקציות RayDP באשכול Ray ב-Gemini Enterprise Agent Platform. בקטע הבא מוסבר איך ליצור תמונה מותאמת אישית של RayDP.

יצירת קובץ אימג' של קונטיינר בהתאמה אישית של Ray ב-Gemini Enterprise Agent Platform

אפשר להשתמש בקובץ Docker הזה כדי ליצור קובץ אימג' של קונטיינר בהתאמה אישית ל-Ray ב-Gemini Enterprise Agent Platform, שמותקן בו 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

אתם יכולים להשתמש באשכול Ray העדכני ביותר בתמונה מוכנה מראש של Gemini Enterprise Agent Platform כדי ליצור את התמונה המותאמת אישית של RayDP. אפשר גם להתקין חבילות Python אחרות שאתם צופים שתשתמשו בהן באפליקציות שלכם. הבעיה pyarrow==14.0 נובעת מאילוץ תלות של Ray 2.9.3.

יצירה (build) ושליחה (push) של קובץ אימג' של הקונטיינר בהתאמה אישית

לפני שיוצרים קובץ אימג' בהתאמה אישית, צריך ליצור מאגר Docker ב-Artifact Registry (במאמר עבודה עם קובצי אימג' של קונטיינרים מוסבר איך ליצור ולהגדיר מאגר Docker). אחרי שיוצרים את מאגר Docker, יוצרים את קובץ אימג' של קונטיינר המותאם אישית ומעבירים אותו בדחיפה באמצעות קובץ Docker.

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]

כאשר:

  • ‫LOCATION: המיקום ב-Cloud Storage (לדוגמה, us-central1) שיצרתם ב-Artifact Registry.
  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
  • ‫DOCKER_REPOSITORY: השם של מאגר Docker שיצרתם.
  • ‫IMAGE_NAME: השם של קובץ האימג' בקונטיינר בהתאמה אישית.

יצירת אשכול Ray ב-Gemini Enterprise Agent Platform

משתמשים בקובץ האימג' של הקונטיינר בהתאמה אישית שנוצר בשלב הקודם כדי ליצור אשכול Ray ב-Gemini Enterprise Agent Platform. אתם יכולים להשתמש ב-Agent Platform SDK for Python כדי ליצור אשכול Ray ב-Gemini Enterprise Agent Platform.

אם עדיין לא עשיתם זאת, תצטרכו להתקין את ספריות Python הנדרשות.

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

מגדירים צמתי Head ו-Worker ויוצרים את האשכול באמצעות Agent Platform SDK ל-Python. לדוגמה:

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

כאשר:

  • ‫CUSTOM_CONTAINER_IMAGE_URI: ה-URI של קובץ אימג' של קונטיינר מותאם אישית שהועלה ל-Artifact Registry.
  • ‫CLUSTER_NAME: השם של אשכול Ray ב-Gemini Enterprise Agent Platform.

אשכול Spark ב-Ray ב-Gemini Enterprise Agent Platform

לפני שמריצים את אפליקציית Spark, צריך ליצור סשן Spark באמצעות RayDP API. אפשר להשתמש בלקוח Ray כדי לעשות זאת באופן אינטראקטיבי, או להשתמש ב-Ray job API. מומלץ להשתמש ב-Ray job API, במיוחד באפליקציות שפועלות בסביבת הייצור ובאפליקציות שפועלות לאורך זמן. ב-RayDP API יש פרמטרים להגדרת סשן Spark, והוא תומך גם בהגדרת Spark. מידע נוסף על RayDP API ליצירת Spark Session זמין במאמר בנושא Spark master actors node affinity.

‫RayDP עם לקוח Ray

אתם יכולים להשתמש ב-Ray Task או ב-Actor כדי ליצור אשכול Spark וסשן באשכול Ray ב-Gemini Enterprise Agent Platform. כדי להשתמש ב-Ray Client כדי ליצור סשן Spark באשכול Ray ב-Gemini Enterprise Agent Platform, צריך Ray Task או Actor. בדוגמה הבאה אפשר לראות איך Ray Actor יכול ליצור Spark Session, להריץ אפליקציית Spark ולהפסיק Spark cluster ב-Ray cluster ב-Gemini Enterprise Agent Platform באמצעות RayDP.

כדי להתחבר לאשכול Ray ב-Gemini Enterprise Agent Platform באופן אינטראקטיבי, אפשר לעיין במאמר בנושא התחברות לאשכול Ray באמצעות 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 עם Ray Job API

לקוח Ray שימושי לניסויים קטנים שדורשים חיבור אינטראקטיבי לאשכול Ray. Ray Job API הוא הדרך המומלצת להפעלת משימות ארוכות טווח ומשימות בסביבת ייצור באשכול Ray. הדבר נכון גם לגבי הפעלת אפליקציות Spark באשכול Ray ב-Gemini Enterprise Agent Platform.

יוצרים סקריפט Python שמכיל את קוד האפליקציה של Spark. לדוגמה:

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

שולחים את העבודה להרצת סקריפט Python באמצעות Ray Job API. לדוגמה:

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

כאשר:

  • ‫SCRIPT_NAME: שם הקובץ של הסקריפט שיצרתם.

קריאת קבצים ב-Cloud Storage מאפליקציית Spark

נהוג לאחסן קובצי נתונים בקטגוריה של Google Cloud Storage. אפשר לקרוא את הקבצים האלה בכמה דרכים מאפליקציית Spark שפועלת באשכול Ray ב-Gemini Enterprise Agent Platform. בקטע הזה מוסברות שתי טכניקות לקריאת קבצים מ-Cloud Storage מאפליקציות Spark שפועלות ב-Ray Cluster ב-Gemini Enterprise Agent Platform.

שימוש במחבר Google Cloud Storage

אפשר להשתמש ב- Google Cloud Connector for Hadoop כדי לקרוא קבצים מקטגוריה של Cloud Storage מאפליקציית Spark. אחרי שיוצרים סשן Spark באמצעות RayDP, אפשר לקרוא קבצים באמצעות כמה פרמטרים של הגדרה. בדוגמת הקוד הבאה אפשר לראות איך קוראים קובץ CSV שמאוחסן בקטגוריה של Cloud Storage מאפליקציית Spark באשכול Ray ב-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)

כאשר:

  • ‫GCS_FILE_URI: ה-URI של קובץ שמאוחסן בקטגוריה של Cloud Storage. לדוגמה: gs://my-bucket/my-file.csv.

שימוש בנתוני Ray

מחבר Google Cloud מאפשר לקרוא קבצים ממאגר Google Cloud ועשוי להספיק לרוב תרחישי השימוש. אולי תרצו להשתמש ב-Ray Data כדי לקרוא קבצים מ- Google Cloud bucket כשאתם צריכים להשתמש בעיבוד המבוזר של Ray כדי לקרוא נתונים, או כשאתם נתקלים בבעיות בקריאת קובץGoogle Cloud באמצעות המחבר Google Cloud . יכול להיות שהבעיה נובעת מסתירות בתלות ב-Java, כשמוסיפים תלויות של אפליקציה אחרת לנתיב המחלקה של Spark Java באמצעות spark.jars.packages או 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)

פונקציית Pyspark Pandas UDF באשכול Ray ב-Gemini Enterprise Agent Platform

פונקציות מוגדרות על ידי המשתמש (UDF) של Pyspark Pandas לפעמים דורשות קוד נוסף כשמשתמשים בהן באפליקציית Spark שפועלת באשכול Ray בפלטפורמת הסוכנים של Gemini Enterprise. בדרך כלל נדרש להשתמש באפשרות הזו כשפונקציית Pandas UDF משתמשת בספריית Python שלא זמינה באשכול Ray בפלטפורמת Gemini Enterprise Agent. אתם יכולים לארוז את יחסי התלות של Python של אפליקציה באמצעות סביבת זמן הריצה עם Ray Job API. אחרי ששולחים את משימת Ray לאשכול, Ray מתקין את התלות האלה בסביבה הווירטואלית של Python שהוא יוצר להרצת המשימה. אבל פונקציות Pandas UDF לא משתמשות באותה סביבה וירטואלית. במקום זאת, הם משתמשים בסביבת ברירת המחדל של מערכת Python. אם התלות הזו לא זמינה בסביבת המערכת, יכול להיות שתצטרכו להתקין אותה ב-UDF של Pandas. בדוגמה הבאה, ספריית statsmodels מותקנת בתוך ה-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()