E-MapReduce:Ray quick start

Pré-requisitos

Crie um cluster Ray e verifique se ele está em execução. Obtenha também o endpoint e o token na página Endpoint Information do cluster. Para mais informações, consulte Gerencie clusters Ray.

Escolha um método de envio

O cluster Ray oferece três métodos de envio. Escolha o mais adequado ao seu cenário.

Desenvolvimento interativo

Ideal para início rápido e depuração. Conecte-se diretamente a um cluster Ray pelo ambiente Notebook no console do EMR Serverless Spark, sem configurações adicionais de rede. Para usar sua máquina local, garanta conectividade de rede com o cluster Ray e um ambiente Python 3.12.

1. Instale o cliente Ray:

   pip install ray[client]==2.47.1

2. Na página invocation information do cluster, obtenha o gRPC address e o token.

3. Use o código de exemplo a seguir para conectar-se ao cluster Ray e desenvolver interativamente:

   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)

   Após conectar-se ao cluster Ray e executar a tarefa de computação, o terminal exibirá uma saída semelhante a esta:

   2025-12-22 17:32:04,514 INFO util.py:154 — Missing packages: ['ipywidgets']. Run `pip install -U ipywidgets`, then restart the notebook server for rich notebook output.
   2025-12-22 17:32:04,537 INFO client_builder.py:242 — Passing the following kwargs to ray.init() on the server: log_to_driver
   2025-12-22 17:32:05,842 WARNING utils.py:1565 — Python patch version mismatch: The cluster was started with:
       Ray: 2.47.1
       Python: 3.12.9
   This process on Ray Client was started with:
       Ray: 2.47.1
       Python: 3.12.12
   
   Square results: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Envio via SDK

Recomendado para incorporar envios de Ray jobs no código da aplicação. O ambiente local deve ter Python 3.12 instalado.

1. Instale a biblioteca cliente de envio de jobs do Ray:

   pip install "ray[default]==2.47.1"

2. Na página invocation information do cluster, obtenha o public endpoint e o token.

3. Use o código de exemplo a seguir para conectar-se ao cluster Ray e enviar um job:

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # Alternatively, you can use the internal network. This requires the client to be in the same region and VPC as the cluster. 
   # Submitting over the internal network is recommended for improved stability.
   # 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) 

   Após executar o script, o terminal exibirá uma saída semelhante a esta:

   (.venv) xxx % python '/Users/xxx/xxx.py'
   Submitted job with ID: raysubmit_9Q5JTEgG8tqLEJb4
   Job status: PENDING
   Job status: PENDING
   Job status: SUCCEEDED
   Job logs:
   2026-04-03 17:01:15,126 INFO job_manager.py:531 -- Runtime env is setting up.
   Hello from Ray Client!

4. (Opcional) Para acessar arquivos de um diretório montado no job, monte um diretório de arquivos gerenciados ao criar o cluster Ray. Depois, referencie os arquivos pelo caminho de montagem ao enviar o job. O exemplo a seguir considera que você montou um diretório OSS no caminho /mnt/myoss:

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # Use the mount path to directly reference a script file in OSS as the 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)

   Nota

   No código do Ray job, você também pode ler e gravar arquivos diretamente pelos caminhos de montagem. Por exemplo, leia o conteúdo de /mnt/myoss/test.txt ou grave um arquivo no diretório /mnt/myoss/. Os arquivos gravados nesse diretório são sincronizados automaticamente com o caminho correspondente no OSS.

   O exemplo a seguir mostra como ler e gravar arquivos em um diretório montado dentro de um Ray job:

   # The following code is submitted to the Ray cluster.
   
   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)

Envio via linha de comando

Ideal para integração com sistemas externos. O ambiente local deve ter Python 3.12 instalado.

1. Instale a biblioteca cliente de envio de jobs do Ray:

   pip install "ray[default]==2.47.1"

2. Na página invocation information do cluster, obtenha o public endpoint e o token.

3. Execute os comandos a seguir para enviar um Ray job:

   # You must set the RAY_ADDRESS and RAY_JOB_HEADERS environment variables before you run `ray job submit`.
   export RAY_ADDRESS='<your_public_endpoint>'               
   export RAY_JOB_HEADERS='{"ray-token": "<your_token>"}'
   
   ## Submit the job
   ### Example: working-dir supports local paths & Ray core distributed sort.
   ray job submit --working-dir "." -- python  test-ray-core-sort.py
   
   ### Supports reading from and writing to OSS.
   ray job submit --working-dir "." -- python test-saving-data-test.py
   
   ### Supports reading from and writing to OSS-HDFS.
   ray job submit --working-dir "." -- python test-saving-data-oss-hdfs-test.py
   
   ### The code supports reading from and writing to mounted directories.
   ray job submit --working-dir "." -- python  test-saving-data-test-mount.py

Para mais parâmetros de linha de comando, consulte a documentação oficial da CLI de envio de jobs do Ray.