Trigger Pipelines with Assets in Apache Airflow 3

Table of Contents

1. Challenge

   Verify the lab environment

   Introduction

   Apache Airflow 3 replaces the Datasets API with Assets a first-class primitive for data-aware scheduling. Instead of triggering a downstream DAG on a clock interval, you declare that a DAG cares about a named data resource (an Asset). When an upstream task marks that Asset as updated, Airflow's scheduler automatically queues every consumer DAG that depends on it.

   This lab walks you through the complete producer-consumer pattern using Assets. You write a producer DAG that emits an Asset event after each run, then write a consumer DAG that replaces a cron schedule with a dependency on that same Asset. By the end, you have a working event-driven pipeline you can observe end-to-end in the Airflow UI.

   Learning objectives

   After completing this lab you will be able to:

   • Define an Asset and write a producer DAG whose task updates it.
   • Schedule a consumer DAG on the Asset instead of a time interval.
   • Run the producer with the scheduler active and verify the consumer triggers automatically.

   Lab environment

   The lab environment includes:

   • Python 3.11
   • Apache Airflow 3.3.1 installed and pre-configured
   • VS Code editor available in the browser
   • A ~/workspace/ directory containing three starter files: step2_asset.py, step3_producer.py, and step4_consumer.py
   • A ~/workspace/solution/ directory containing reference files for each step
   • A ~/airflow/dags/ directory where completed files are copied for Airflow to discover
   • A ~/workspace/data/ directory containing a starter daily_summary.csv file

   Note: The Airflow web UI runs on port 8080. Credentials are generated when you start the standalone process in Step 1.

   ---

2. Challenge

   Define the Asset

   In Airflow 3, an Asset is a URI-addressed data resource. Defining it once in a shared module lets both the producer and consumer DAGs import the same object, ensuring Airflow treats them as the same dependency.

   The Asset class lives in airflow.sdk. Its constructor accepts a URI string — a logical identifier for the data resource. Rather than hardcoding a full absolute path, you will use os.path.expanduser to resolve the ~ shorthand at import time, making the URI portable across different user accounts.

   Note: Keep the URI consistent between files. A one-character difference produces two separate Assets and breaks the producer-consumer link. Never pass ~ directly inside a URI string — it will not be expanded and the producer and consumer will not match.

3. Challenge

   Create the producer DAG

   The producer DAG runs on a time-based schedule and contains a task decorated with outlets. The outlets parameter tells Airflow which Assets the task updates when it completes successfully. After each successful run, Airflow records an Asset event for every URI listed in outlets.

4. Challenge

   Create the consumer DAG

   The consumer DAG replaces a schedule string with a list containing the Asset object. When Airflow detects a new Asset event for every URI in the schedule list, it queues a new DAG run automatically — no clock trigger required.

5. Challenge

   Run the producer and verify automatic triggering

   With both DAGs active, you can now observe the end-to-end event-driven behavior. Triggering the producer manually causes it to emit an Asset event, which the scheduler detects and uses to queue the consumer automatically.

   Summary

   In this lab you built a complete event-driven pipeline using Apache Airflow 3 Assets:

   • You defined an Asset with a URI in step2_asset.py and called it at module level so both DAGs could import the same object.
   • You created a producer DAG in step3_producer.py that runs on a daily schedule and marks the Asset as updated by adding outlets=[daily_sales_asset] to the task decorator.
   • You created a consumer DAG in step4_consumer.py that uses schedule=[daily_sales_asset] in place of a cron expression, making Airflow responsible for queuing it whenever the Asset is updated.
   • You triggered the producer manually and confirmed that Airflow queued and ran the consumer automatically, demonstrating data-aware scheduling without any polling or manual coordination.

   ---

   Solution directory

   The ~/workspace/solution/ directory contains reference files for each step in this lab.

   | File | Associated step | |---|---| | step2_asset.py | Step 2 — Shared asset definition | | step3_producer.py | Step 3 — Producer DAG | | step4_consumer.py | Step 4 — Consumer DAG |