E-MapReduce：向Ray叢集提交任務

方式一：互動式開發

適合快速入門和調試。在EMR Serverless Spark控制台的Notebook環境中直接連接Ray叢集（無需額外配置網路）。如從本地串連，需確保本地機器與Ray叢集之間網路互連，且環境為Python 3.12。

1. 安裝Ray用戶端：

   pip install ray[client]==2.47.1

2. 在叢集的調用資訊頁面，擷取gRPC 地址和Token。

3. 使用以下範例程式碼串連Ray叢集，進行互動式開發：

   import ray
   import os
   
   def get_metadata():
       headers = {"ray-token": "<yourToken>"}
       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("平方結果:", results)

   

方式二：SDK提交（批任務）

適合將Ray作業嵌入工程代碼中提交。本地環境要求：Python 3.12。

1. 安裝Ray Job Submission用戶端庫：

   pip install "ray[default]==2.47.1"

2. 在叢集的調用資訊頁面，擷取公網調用地址和Token。

3. 使用以下範例程式碼串連Ray叢集並提交作業：

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<yourToken>"
   }
   
   client = JobSubmissionClient("<公網調用地址>", headers=custom_headers)
   
   # 或者用內網，需要保證提交端在同一region vpc下，建議內網提交，更加穩定
   # 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. （可選）如需在作業中訪問掛載目錄中的檔案，可在建立Ray叢集時掛載納管檔案目錄，然後在提交作業時通過掛載路徑引用檔案。以下樣本假設已將OSS目錄掛載到/mnt/myoss路徑下：

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<yourToken>"
   }
   
   client = JobSubmissionClient("<公網調用地址>", headers=custom_headers)
   
   # 通過掛載路徑直接引用OSS中的指令檔作為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)

   說明

   在Ray任務代碼中，也可以直接通過掛載路徑讀寫檔案。例如，讀取/mnt/myoss/test.txt或寫入檔案到/mnt/myoss/目錄下，寫入的檔案會同步到對應的OSS路徑中。

   以下樣本展示如何在Ray任務中讀寫掛載目錄中的檔案：

   # 下面是提交到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 '寫入成功'
   
   print("--- 開始運行遠程任務 ---")
   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)

方式三：命令列提交（批任務）

適合與外部系統整合。本地環境要求：Python 3.12。

1. 安裝Ray Job Submission用戶端庫：

   pip install "ray[default]==2.47.1"

2. 在叢集的調用資訊頁面，擷取公網調用地址和Token。

3. 執行以下命令提交Ray作業：

   # ray job submit前必須設定ray address和headers
   export RAY_ADDRESS='<公網調用地址>'               
   export RAY_JOB_HEADERS='{"ray-token": "<yourToken>"}'
   
   ## 提交任務
   ### working dir 支援本地 & Ray core distributed sort
   ray job submit --working-dir "." -- python  test-ray-core-sort.py
   
   ### 支援OSS讀寫
   ray job submit --working-dir "." -- python test-saving-data-test.py
   
   ### 支援OSS-HDFS讀寫
   ray job submit --working-dir "." -- python test-saving-data-oss-hdfs-test.py
   
   ### 代碼支援Mount讀寫
   ray job submit --working-dir "." -- python  test-saving-data-test-mount.py

更多命令列參數，請參見Ray Job Submission CLI官方文檔。