> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Orchestrate Ray jobs on Anyscale with Apache Airflow®
This page hasn't yet been updated for Airflow 3. The concepts shown are relevant, but some code may need to be updated. If you run any examples, take care to update import statements and watch for any other breaking changes.
[Anyscale](https://www.anyscale.com/) is a compute platform for AI/ML workloads built on the open-source [Ray](https://www.ray.io/) framework, providing the layer for parallel processing and distributed computing. The [Anyscale provider package](https://github.com/astronomer/astro-provider-anyscale) for [Apache Airflow®](https://airflow.apache.org/) allows you to interact with Anyscale from your Airflow DAGs. This tutorial shows a simple example of how to use the Anyscale provider package to orchestrate Ray jobs on Anyscale with Airflow. For more in-depth information, see the [Anyscale provider documentation](https://astronomer.github.io/astro-provider-anyscale/).
For instructions on how to run open-source Ray jobs with Airflow, see the [Orchestrate Ray jobs with Apache Airflow®](/docs/learn/airflow-ray) tutorial.
This tutorial shows a simple implementation of the Anyscale provider package. For a more complex example, see the [Processing User Feedback: an LLM-fine-tuning reference architecture with Ray on Anyscale](/docs/learn/reference-architecture-fine-tuning-anyscale) reference architecture.
## Time to complete
This tutorial takes approximately 30 minutes to complete.
## Assumed knowledge
To get the most out of this tutorial, make sure you have an understanding of:
* Ray basics. See the [Getting Started section of the Ray documentation](https://docs.ray.io/en/latest/ray-overview/getting-started.html).
* Anyscale basics. See [Get started section of the Anyscale documentation](https://docs.anyscale.com/get-started).
* Airflow operators. See [Airflow operators](/docs/learn/what-is-an-operator).
## Prerequisites
* The [Astro CLI](/docs/cli/v1.43/get-started-cli).
* An [Anyscale](https://www.anyscale.com/) account with AI platform features enabled. You also need to have at least one [suitable image](https://docs.anyscale.com/reference/anyscale-base-images/) and [compute config](https://docs.anyscale.com/configuration/compute-configuration/#create-a-compute-config) available in your Anyscale account.
## Step 1: Configure your Astro project
Use the Astro CLI to create and run an Airflow project on your local machine.
1. Create a new Astro project:
```sh wrap theme={null}
$ mkdir astro-anyscale-tutorial && cd astro-anyscale-tutorial
$ astro dev init
```
2. In the `requirements.txt` file, add the [Anyscale provider](https://github.com/astronomer/astro-provider-anyscale).
```text wrap theme={null}
astro-provider-anyscale==1.0.1
```
3. Run the following command to start your Airflow project:
```sh wrap theme={null}
astro dev start
```
## Step 2: Configure a Ray connection
For Astro customers, Astronomer recommends taking advantage of the [Astro Environment Manager](/docs/astro/manage-connections-variables#astro-environment-manager) to store connections in an Astro-managed secrets backend. These connections can be shared across multiple deployed and local Airflow environments. See [Manage Astro connections in branch-based deploy workflows](/docs/astro/best-practices/connections-branch-deploys).
1. In the Airflow UI, go to **Admin** -> **Connections** and click **+**.
2. Create a new connection and choose the **Anyscale** connection type. Enter the following information:
* **Connection ID**: `anyscale_conn`
* **API Key**: Your [Anyscale API key](https://docs.anyscale.com/endpoints/text-generation/authenticate)
3. Click **Save**.
## Step 3: Write a DAG to orchestrate Anyscale jobs
1. Create a new file in your `dags` directory called `anyscale_script.py` and add the following code:
```python wrap theme={null}
# anyscale_script.py
import numpy as np
import ray
import argparse
@ray.remote
def square(x):
return x**2
def main(data):
ray.init()
data = np.array(data)
futures = [square.remote(x) for x in data]
results = ray.get(futures)
mean = np.mean(results)
print(f"Mean squared value: {mean}")
return mean
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Process some integers.")
parser.add_argument(
"data", nargs="+", type=float, help="List of numbers to process"
)
args = parser.parse_args()
data = args.data
main(data)
```
2. Create a new file in your `dags` directory called `anyscale_tutorial.py`.
3. Copy and paste the code below into the file:
```python expandable wrap theme={null}
"""
## Anyscale Tutorial
This tutorial demonstrates how to use the Anyscale provider in Airflow to
parallelize a task using Ray on Anyscale.
"""
from airflow.decorators import dag
from airflow.operators.python import PythonOperator
from anyscale_provider.operators.anyscale import SubmitAnyscaleJob
from airflow.models.baseoperator import chain
from pathlib import Path
CONN_ID = "anyscale_conn"
FOLDER_PATH = Path(__file__).parent
def _generate_data() -> list:
"""
Generate sample data
Returns:
list: List of integers
"""
import random
return [random.randint(1, 100) for _ in range(10)]
@dag(
start_date=None,
schedule=None,
catchup=False,
tags=["ray", "example"],
doc_md=__doc__,
)
def anyscale_tutorial():
data = PythonOperator(
task_id="generate_data",
python_callable=_generate_data,
)
get_mean_squared_value = SubmitAnyscaleJob(
task_id="SubmitRayJob",
conn_id=CONN_ID,
name="AstroJob",
image_uri="< your image uri >", # e.g. "anyscale/ray:2.35.0-slim-py312-cpu"
compute_config="< your compute config >", # e.g. airflow-integration-testing:1
entrypoint="python anyscale_script.py {{ ti.xcom_pull(task_ids='generate_data') | join(' ') }}",
working_dir=str(FOLDER_PATH), # the folder containing the script
requirements=["requests", "pandas", "numpy", "torch"],
max_retries=1,
job_timeout_seconds=3000,
poll_interval=30,
)
chain(data, get_mean_squared_value)
anyscale_tutorial()
```
* The `generate_data` task randomly generates a list of 10 integers.
* The `get_mean_squared_value` task submits a Ray job on Anyscale to calculate the mean squared value of the list of integers.
## Step 4: Run the DAG
1. In the Airflow UI, click the play button to manually run your DAG.
2. After the DAG runs successfully, check your Anyscale account to see the job submitted by Airflow.
## Conclusion
Congratulations! You've run a Ray job on Anyscale using Apache Airflow. You can now use the Anyscale provider package to orchestrate more complex jobs, see [Processing User Feedback: an LLM-fine-tuning reference architecture with Ray on Anyscale](/docs/learn/reference-architecture-fine-tuning-anyscale) for an example.