Overview

I created a custom Trigger for Airflow in Composer.

It was quite challenging, but using it helps reduce resource consumption and cut costs, so I highly recommend it.

Why I Used a Custom Trigger

I wanted to periodically check the update status of a BigQuery table via API and, once it was updated, trigger an update for another table that depends on it.

If I ran that process on an Airflow Worker, it would occupy CPU and memory until the table was updated, which gets expensive.

There are a large number of tables to monitor for updates, and the wait time can be on the order of several hours.

The Trigger process was introduced for exactly these kinds of use cases.

When you call the defer method in an Airflow Operator, processing is handed off to the Triggerer process, which handles multiple tasks asynchronously in a single process.

When a Trigger's processing completes, control is returned to the calling Operator and the task finishes.

While processing is handed off to the Triggerer process, the task enters a deferred status and the resources used by the Worker are released — so even tasks that need to wait for a table update can run with minimal resource consumption.

How to Run a Custom Trigger

This time I ran a custom Trigger in Airflow on GCP Composer.

Unlike DAGs, simply placing a Python file in the Composer bucket won't work.

The Triggerer process is loaded and started when Airflow starts up.

Therefore, you need to install it in Composer as a PyPI package.

There are various ways to install a custom PyPI package in Composer, but this time I uploaded a Python package I generated to a standard repository in Artifact Registry and installed it in Composer from there.

I'll write a separate post about that.

Code

You should be able to implement it without issues by following the reference below.

www.astronomer.io

If you have any questions or unresolved issues, feel free to message me.