E-MapReduce:Kirim pekerjaan ke kluster Ray

Metode 1: Pengembangan interaktif

Metode ini ideal untuk memulai dengan cepat dan debugging. Anda dapat terhubung langsung ke kluster Ray dari lingkungan Notebook di Konsol EMR Serverless Spark tanpa konfigurasi jaringan tambahan. Jika Anda terhubung dari mesin lokal, pastikan mesin lokal Anda memiliki konektivitas jaringan ke kluster Ray dan menggunakan lingkungan Python 3.12.

1. Instal klien Ray:

   pip install ray[client]==2.47.1

2. Pada halaman invocation information kluster, dapatkan gRPC address dan token.

3. Gunakan kode contoh berikut untuk terhubung ke kluster Ray dalam pengembangan interaktif:

   import ray
   import os
   
   def get_metadata():
       headers = {"ray-token": "<your_token>"}
       return [(key.lower(), value) for key, value in headers.items()]
   
   ray.init(address="<your_gRPC_address>", _metadata=get_metadata())
   
   import time
   @ray.remote
   def square(x):
       time.sleep(0.1)
       return x * x
   
   futures = [square.remote(i) for i in range(10)]
   
   results = ray.get(futures)
   print("Square results:", results)

   

Metode 2: Pengiriman melalui SDK

Metode ini cocok untuk menyematkan pengiriman pekerjaan Ray dalam kode aplikasi Anda. Lingkungan lokal Anda harus memiliki Python 3.12 yang terinstal.

1. Instal library klien pengiriman pekerjaan Ray:

   pip install "ray[default]==2.47.1"

2. Pada halaman invocation information kluster, dapatkan public endpoint dan token.

3. Gunakan kode contoh berikut untuk terhubung ke kluster Ray dan mengirim pekerjaan:

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # Alternatifnya, Anda dapat menggunakan jaringan internal. Ini memerlukan klien berada di wilayah dan VPC yang sama dengan kluster. 
   # Mengirim melalui jaringan internal direkomendasikan untuk stabilitas yang lebih baik.
   # client = JobSubmissionClient("http://emr-spark-ray-gateway-cn-beijing-internal.spark.emr.aliyuncs.com", headers=custom_headers)
   
   job_id = client.submit_job(
       entrypoint="python -c 'print(\"Hello from Ray Client!\")'"
   )
   
   print(f"Submitted job with ID: {job_id}")
   
   while True:
       status = client.get_job_status(job_id)
       print(f"Job status: {status}")
       if status.is_terminal():
           break
       time.sleep(1)
   
   logs = client.get_job_logs(job_id)
   print("Job logs:")
   print(logs) 

   

4. (Opsional) Untuk mengakses file dalam direktori yang dimount dari pekerjaan Anda, mount direktori file terkelola saat membuat kluster Ray. Lalu, rujuk file tersebut menggunakan path mount-nya saat mengirim pekerjaan. Contoh berikut mengasumsikan Anda telah memount direktori OSS ke path /mnt/myoss:

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # Gunakan path mount untuk langsung merujuk file skrip di OSS sebagai entrypoint.
   job_id = client.submit_job(
       entrypoint="python /mnt/myoss/main.py"
   )
   
   print(f"Submitted job with ID: {job_id}")
   
   while True:
       status = client.get_job_status(job_id)
       print(f"Job status: {status}")
       if status.is_terminal():
           break
       time.sleep(1)
   
   logs = client.get_job_logs(job_id)
   print("Job logs:")
   print(logs)

   Catatan

   Dalam kode pekerjaan Ray Anda, Anda juga dapat membaca dan menulis file secara langsung menggunakan path mount-nya. Misalnya, Anda dapat membaca dari /mnt/myoss/test.txt atau menulis file ke direktori /mnt/myoss/. File yang ditulis ke direktori ini akan disinkronkan ke path OSS yang sesuai.

   Contoh berikut menunjukkan cara membaca dan menulis file dalam direktori yang dimount dalam pekerjaan Ray:

   # Kode berikut dikirim ke kluster Ray.
   
   import ray
   import time
   
   ray.init()
   
   @ray.remote
   def read_from_mount_path(i):
       with open('/mnt/myoss/test.txt', 'r', encoding='utf-8') as f:
           content = f.read()
       return content
   
   @ray.remote
   def write_to_mount_path(i):
       content = "Hello world " + str(i)
       with open('/mnt/myoss/output' + str(i) + '.txt', 'w', encoding='utf-8') as f:
           f.write(content)
       return 'Write succeeded'
   
   print("--- Starting remote tasks ---")
   start_time = time.time()
   obj_refs = [read_from_mount_path.remote(i) for i in range(4)]
   obj_refs2 = [write_to_mount_path.remote(i) for i in range(4)]
   results = ray.get(obj_refs)
   results2 = ray.get(obj_refs2)

Metode 3: Pengiriman melalui command line

Metode ini ideal untuk integrasi dengan sistem eksternal. Lingkungan lokal Anda harus memiliki Python 3.12 yang terinstal.

1. Instal library klien pengiriman pekerjaan Ray:

   pip install "ray[default]==2.47.1"

2. Pada halaman invocation information kluster, dapatkan public endpoint dan token.

3. Jalankan perintah berikut untuk mengirim pekerjaan Ray:

   # Anda harus mengatur variabel lingkungan RAY_ADDRESS dan RAY_JOB_HEADERS sebelum menjalankan `ray job submit`.
   export RAY_ADDRESS='<your_public_endpoint>'               
   export RAY_JOB_HEADERS='{"ray-token": "<your_token>"}'
   
   ## Kirim pekerjaan
   ### Contoh: Kirim pekerjaan yang melakukan pengurutan terdistribusi.
   ray job submit --working-dir "." -- python  test-ray-core-sort.py
   
   ### Mendukung pembacaan dan penulisan ke OSS.
   ray job submit --working-dir "." -- python test-saving-data-test.py
   
   ### Mendukung pembacaan dan penulisan ke OSS-HDFS.
   ray job submit --working-dir "." -- python test-saving-data-oss-hdfs-test.py
   
   ### Kode mendukung pembacaan dan penulisan ke direktori yang dimount.
   ray job submit --working-dir "." -- python  test-saving-data-test-mount.py

Untuk parameter command line lainnya, lihat dokumentasi resmi CLI pengiriman pekerjaan Ray.