raydp-spark-remote.py

raydp-spark-remote.py

import os
import ray
import raydp


HEAD_SERVICE_IP_ENV = "EXAMPLE_CLUSTER_RAY_HEAD_SERVICE_HOST"
HEAD_SERVICE_CLIENT_PORT_ENV = "EXAMPLE_CLUSTER_RAY_HEAD_SERVICE_PORT_CLIENT"

head_service_ip = os.environ[HEAD_SERVICE_IP_ENV]
client_port = os.environ[HEAD_SERVICE_CLIENT_PORT_ENV]
ray.util.connect(f"{head_service_ip}:{client_port}")


@ray.remote
class SparkCluster:
    def __init__(self):
        self.spark = None
    def start(self):
        self.spark = raydp.init_spark('word_count',
                                       num_executors=2,
                                       executor_cores=1,
                                       executor_memory='1G')

    def run(self):
        df = self.spark.createDataFrame([('look',), ('spark',), ('tutorial',), ('spark',), ('look', ), ('python', )], ['word'])
        df.show()
        word_count = df.groupBy('word').count()
        word_count.show()

    def stop(self):
        raydp.stop_spark()
    
 
spark = SparkCluster.remote()
ray.get(spark.start.remote())
ray.get(spark.run.remote())
ray.get(spark.stop.remote())

In conclusion, I should ray.util.connect(f"{head_service_ip}:{client_port}") with @ray.remote and use ray.init(address="auto") with original codes.

Does XGBoost side need any changes?