How to Batch AI Jobs with Apache Airflow100

Apache Airflow is a powerful workflow management platform that can be used to automate complex data-driven processes. One of the key features of Airflow is its ability to batch jobs, which can significantly improve the efficiency of your workflows.

In this tutorial, we will show you how to batch AI jobs with Apache Airflow. We will use the Google Cloud Platform (GCP) as our example, but the concepts should be applicable to any cloud platform or on-premises environment.

Prerequisites

Before you begin, you will need the following:
An Apache Airflow environment
A Google Cloud Platform (GCP) project
A Google Cloud Storage (GCS) bucket
A Google Cloud AI Platform (AIPlatform) endpoint

Create a Batch AI Job

To create a batch AI job, you will need to create a Python script that defines the job. The following script creates a batch AI job that uses the AIPlatform Batch Prediction Job Operator to predict the probability that a given image contains a cat:```python
import datetime
from import aiplatform_v1
from airflow import models
from import bash_operator, python_operator

def create_batch_prediction_job():
"""Creates a batch prediction job."""
aiplatform_client = ()
# The AI Platform services require regional API endpoints.
client_options = {'api_endpoint': ''}
# Initialize client that will be used to create and send requests.
# This client only needs to be created once, and can be reused for multiple requests.
client = (client_options=client_options)
gcs_source_uri = 'gs://YOUR_GCS_BUCKET/path/to/your/source/'
gcs_destination_output_uri_prefix = 'gs://YOUR_GCS_BUCKET/path/to/save/results/'
model_name = 'projects/YOUR_PROJECT/locations/YOUR_LOCATION/models/YOUR_MODEL'
batch_prediction_job = {
'display_name': 'YOUR_BATCH_PREDICTION_DISPLAY_NAME',
# Format: 'gs:///'
'input_config': {
'instances_format': 'jsonl',
# Format: 'gs:///'
'gcs_source': {'uris': [gcs_source_uri]}
},
# Format: 'gs:///'
'output_config': {
'predictions_format': 'jsonl',
'gcs_destination': {'output_uri_prefix': gcs_destination_output_uri_prefix}
},
'model': model_name,
'model_parameters': {}
}
parent = f'projects/{project_id}/locations/{location}'
response = client.create_batch_prediction_job(parent=parent, batch_prediction_job=batch_prediction_job)
print("response:", response)

# Define the default arguments for the DAG
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email': ['airflow@'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': (minutes=5),
'start_date': (2023, 1, 1),
}
# Create the DAG
with (
'composer_batch_prediction_tutorial',
default_args=default_args,
description='Batch Prediction Tutorial',
schedule_interval=(days=1),
) as dag:
# Define the BashOperator
create_batch_prediction_job_operator = (
task_id='create_batch_prediction_job',
python_callable=create_batch_prediction_job
)
```

You can then create a batch AI job by running the following command:```
airflow dags backfill composer_batch_prediction_tutorial
```

Monitor the Batch AI Job

Once you have created a batch AI job, you can monitor its progress in the Airflow web interface. The job's status will be updated regularly, and you can view the job's logs to see more details.

Results

Once the batch AI job has completed, you can download the results from the GCS bucket that you specified in the job's output configuration.

Conclusion

Batching AI jobs can significantly improve the efficiency of your workflows by allowing you to process large amounts of data in a single batch. Apache Airflow provides a powerful way to batch AI jobs, and this tutorial has shown you how to do just that.

2025-02-12

Previous：Android Development Tutorial: Creating a Food Cookbook App

Next：Best iOS Development Tutorials: A Comprehensive Guide