Intro to Airflow tutorial: Get started and run your first pipeline

This tutorial will get you started as quickly as possible while explaining the core concepts of Apache Airflow. You will explore galaxies 🌌 while extending an existing workflow with modern Airflow features, setting you up for diving into the world of data orchestration with Apache Airflow.

Step 1: Set up your Astro trial and Astro IDE

1. The first step is to start a free Astro trial.

   All Astro accounts have access to the Astro IDE, which is the easiest way to develop Airflow Dags right in your browser. You can directly deploy your Dags from the Astro IDE to an Astro Deployment, an Airflow environment running in the cloud. After entering your email address, starting the trial includes 4 steps:

   1. Choose between professional and personal. The choice has no impact on this tutorial.
   2. Enter an organization and workspace name. Each customer has a dedicated organization on Astro. Each team or project has a workspace, which is a collection of deployments. A deployment is an Airflow environment hosted on Astro. For this tutorial, you can use any names.
   3. You can choose to upload Dags, use a template, or start with an empty workspace. For this tutorial, choose Start with a template.
   4. Choose the ETL template.

   

   Astro Concepts

   • Astro: Fully-managed platform that helps teams write and run data pipelines with Airflow at any scale.
   • Astro IDE: In-browser IDE with context-aware AI and zero local setup.
   • Organization: Each customer has a dedicated org on Astro.
   • Workspace: Each team or project has a dedicated workspace, containing a collection of deployments.
   • Deployment: Airflow environment hosted on Astro.
   • Summary: 1 Organization → n Workspace → n Deployment → 1 Airflow instance.

   After your environment is created, you’ll find yourself in the Astro IDE with your very first ETL Dag, ready to be deployed. The Python code is a programmatic representation of your workflow. Clicking Start Test Deployment in the top right starts a fully functional Airflow environment and deploys your code.

2. Click Start Test Deployment and wait for the deployment to finish.

   

3. Your first Airflow Dag is deployed and ready to be executed. Click the dropdown menu next to Sync to Test and select Open Airflow.

   

   The Airflow UI home dashboard of your Airflow instance will open in a new browser tab.

   

Step 2: Run your first Dag

1. Within the navbar on the left, click Dags.

   This view shows all your Dags defined in your Python code. The ETL template comes with one Dag named example_etl_galaxies.

   

   This ETL (Extract, Transform, Load) pipeline retrieves data about galaxies, filters them based on their distance from the Milky Way, and stores the results in a DuckDB database.

   

   The Dag has these tasks:

   • create_galaxy_table_in_duckdb: Creates a table in DuckDB with columns for galaxy name, distances, type, and characteristics.
   • extract_galaxy_data: Retrieves raw data about 20 galaxies and returns it as a pandas DataFrame.
   • transform_galaxy_data: Filters the galaxy data to keep only galaxies within a specified distance from the Milky Way (default: 500,000 light years).
   • load_galaxy_data: Inserts the filtered galaxy data into the DuckDB table and produces an Airflow Asset update.
   • print_loaded_galaxies: Queries and prints all stored galaxies from DuckDB, sorted by distance from the Milky Way.

   The tasks have these dependencies:

   • create_galaxy_table_in_duckdb → load_galaxy_data (table must exist before loading)
   • extract_galaxy_data → transform_galaxy_data (raw data is needed for filtering)
   • transform_galaxy_data → load_galaxy_data (filtered data is needed for loading)
   • load_galaxy_data → print_loaded_galaxies (data must be loaded before printing)

2. Run the pipeline by clicking Play next to the Dag.

   

   This opens a trigger dialog, allowing you to trigger a single run or a backfill to process a range of dates right from the UI. Dags can also have parameters that can be used within the implementation to keep certain parts of your pipeline configurable.

3. Select Single Run, keep the parameters at their defaults, and click Trigger.

   

   Your Dag starts, and under Latest Run in the Dags view you’ll see the current running instance of it.

4. Click that run date to go to the individual Dag run view.

   

   Watch how the Dag run finishes and explore the grid and graph views (buttons on the top left), two different representations of your pipeline.

5. After all tasks have finished successfully, open the grid view and click the print_loaded_galaxies task, the last step in your pipeline graph.

   

   This opens the logs of this task instance, showing a table of galaxies with their distance from the Milky Way and from the solar system, as well as the type of galaxy.

   

Take your time to explore the UI, trigger more runs, check the logs of other tasks, and make yourself familiar with the interface. Feel free to read the Airflow UI guide for a deep dive into its different views and functionality.

Step 3: Understand the basic concepts

After you’ve finished your exploration, switch back to the Astro IDE and have a look at the Python code inside example_etl_galaxies.py. The code contains a lot of comments explaining each step in detail. Here’s an overview before you dive into details.

The Python file contains the following key elements:

• Imports: All modules, classes, and functions needed for your implementation. Always use the Airflow Task SDK by importing from airflow.sdk, as this is the user-facing SDK.
• Constants: Any constants, like the connection string for the DuckDB instance.
• Dag definition: The data pipeline together with settings like its schedule.
• Tasks: The units of work. Tasks should be atomic and idempotent (producing the same result when run multiple times with the same inputs).
• Dependencies: How the tasks are connected, so Airflow knows how to construct the graph.

1
# imports
2
from airflow.sdk import Asset, chain, Param, dag, task
3
# ...
4
5
# constants
6
_DUCKDB_INSTANCE_NAME = os.getenv("DUCKDB_INSTANCE_NAME", "include/astronomy.db")
7
_DUCKDB_TABLE_NAME = os.getenv("DUCKDB_TABLE_NAME", "galaxy_data")
8
_DUCKDB_TABLE_URI = f"duckdb://{_DUCKDB_INSTANCE_NAME}/{_DUCKDB_TABLE_NAME}"
9
# ...
10
11
# Dag definition
12
@dag(...)
13
def example_etl_galaxies():
14
15
    # tasks
16
    @task(retries=2)
17
    def create_galaxy_table_in_duckdb(...):
18
        # ...
19
20
    @task
21
    def extract_galaxy_data(...):
22
        # ...
23
24
    @task
25
    def transform_galaxy_data(...):
26
        # ..
27
28
    @task
29
    def load_galaxy_data(...):
30
        # ...
31
32
    @task
33
    def print_loaded_galaxies(...):
34
        # ...
35
36
    # create task instances and define implicit dependencies
37
    create_galaxy_table_in_duckdb_obj = create_galaxy_table_in_duckdb()
38
    extract_galaxy_data_obj = extract_galaxy_data()
39
    transform_galaxy_data_obj = transform_galaxy_data(extract_galaxy_data_obj)
40
    load_galaxy_data_obj = load_galaxy_data(transform_galaxy_data_obj)
41
42
    # define explicit dependencies
43
    chain(
44
        create_galaxy_table_in_duckdb_obj, load_galaxy_data_obj, print_loaded_galaxies()
45
    )
46
47
# Instantiate the Dag
48
example_etl_galaxies()

Step 4: Extend the demo project

Now that you’ve run your first Dag, extend the project by adding a second Dag that builds on top of the first one.

You’ll create a galaxy_maintenance Dag that allows you to manually enter new galaxy data through an interactive form. The data is automatically added to the database and validated with automated quality checks.

What you’ll learn:

By the end of this section, you’ll have a powerful toolbox of concepts to explore Airflow further and confidently jump into your first real-world ETL/ELT project!

Step 4.1: Add provider packages

The example_etl_galaxies Dag currently connects directly to the DuckDB database using:

1
cursor = duckdb.connect(duckdb_instance_name)

While this works, Airflow offers a better approach: common SQL operators that execute queries using Airflow connections. This unifies and simplifies SQL workloads across your pipelines. The following steps set this up.

Airflow’s core functionality can be extended with provider packages for specific use cases. This tutorial uses two providers for the DuckDB connection.

1. Open the requirements.txt file in the Astro IDE.

2. Add the following lines at the bottom:

   apache-airflow-providers-common-sql==1.28.2
   airflow-provider-duckdb==0.2.0

3. Since you added new dependencies, sync the changes. Click Sync to Test and wait for the changes to be deployed.

Step 4.2: Set up a connection

An Airflow connection stores configuration details for connecting to external tools in your data ecosystem. Most hooks (what is a hook?) and operators that interact with external systems require a connection.

To create the connection:

1. Open Airflow and click Admin in the left navbar

2. Select Connections

3. Click Add Connection (top right)

4. Enter the following details:

   • Connection ID: duckdb_astronomy
   • Connection Type: DuckDB
   • Host: include/astronomy.db
   • Leave the remaining fields empty.

   

5. Save the connection and you’re now ready to connect! You can find the Airflow task that uses this connection in the example code in Step 4.4.

Step 4.3: Prepare test deployment for advanced usage

The test deployment is a fully functional but minimal Airflow setup. To enable advanced features like asset-aware scheduling (explained later), you need to apply a quick configuration change.

1. In the Astro IDE, click the dropdown menu next to Sync to Test (top right).
2. Select Test Deployment Details.
3. Navigate to the Environment tab, click Edit Deployment Variables, and remove AIRFLOW__SCHEDULER__USE_JOB_SCHEDULE by clicking the trash bin icon next to it.
4. Click Update Environment Variables (bottom right) and you’re ready to go! Head back to the Astro IDE.

Step 4.4: Implement Dag with human-in-the-loop

1. Within the Astro IDE, create a new file by right-clicking on the dags folder → New File… and name it galaxy_maintenance.py.

2. Paste the following content:

   1
   from airflow.sdk import chain, dag, Asset, Param
   2
   from airflow.providers.standard.operators.hitl import HITLEntryOperator
   3
   from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator, SQLColumnCheckOperator
   4
   5
   _DUCKDB_TABLE_URI = "duckdb://include/astronomy.db/galaxy_data"
   6
   _DUCKDB_CONN_ID = "duckdb_astronomy"
   7
   8
   galaxy_table_asset = Asset(_DUCKDB_TABLE_URI)
   9
   10
   @dag(schedule=galaxy_table_asset)
   11
   def galaxy_maintenance():
   12
   13
       _enter_galaxy_details = HITLEntryOperator(
   14
           task_id="enter_galaxy_details",
   15
           subject="Please provide required information: ",
   16
           params={
   17
               "name": Param("", type="string"),
   18
               "distance_from_milkyway": Param(10000, type="number"),
   19
               "distance_from_solarsystem": Param(10000, type="number"),
   20
               "type_of_galaxy": Param("Dwarf", type="string", enum=[
   21
                   "Dwarf Spheroidal",
   22
                   "Dwarf",
   23
                   "Irregular",
   24
                   "Spiral"
   25
               ]),
   26
               "characteristics": Param("", type="string")
   27
           }
   28
       )
   29
   30
       _insert_galaxy_details = SQLExecuteQueryOperator(
   31
           task_id="insert_galaxy_details",
   32
           conn_id=_DUCKDB_CONN_ID,
   33
           show_return_value_in_logs=True,
   34
           sql="""
   35
               -- in case db was removed due to sync
   36
               CREATE TABLE IF NOT EXISTS galaxy_data (
   37
                   name STRING PRIMARY KEY,
   38
                   distance_from_milkyway INT,
   39
                   distance_from_solarsystem INT,
   40
                   type_of_galaxy STRING,
   41
                   characteristics STRING
   42
               );
   43
               INSERT OR IGNORE INTO galaxy_data BY NAME
   44
               SELECT
   45
                   $name AS name,
   46
                   $distance_from_milkyway AS distance_from_milkyway,
   47
                   $distance_from_solarsystem AS distance_from_solarsystem,
   48
                   $type_of_galaxy AS type_of_galaxy,
   49
                   $characteristics AS characteristics
   50
           """,
   51
           parameters={
   52
               "name": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['name'] }}",
   53
               "distance_from_milkyway": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['distance_from_milkyway'] }}",
   54
               "distance_from_solarsystem": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['distance_from_solarsystem'] }}",
   55
               "type_of_galaxy": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['type_of_galaxy'] }}",
   56
               "characteristics": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['characteristics'] }}"
   57
           }
   58
       )
   59
   60
       _galaxy_dq_checks = SQLColumnCheckOperator(
   61
           task_id="dq_checks",
   62
           conn_id=_DUCKDB_CONN_ID,
   63
           table="galaxy_data",
   64
           column_mapping={
   65
               "distance_from_milkyway": {
   66
                   "min": {"geq_to": 10000},
   67
                   "max": {"leq_to": 900000},
   68
               },
   69
               "distance_from_solarsystem": {
   70
                   "min": {"geq_to": 10000},
   71
                   "max": {"leq_to": 900000},
   72
               },
   73
           },
   74
       )
   75
   76
       chain(_enter_galaxy_details, _insert_galaxy_details, _galaxy_dq_checks)
   77
   78
   galaxy_maintenance()

   This maintenance pipeline is triggered automatically whenever the galaxy data table is updated. It allows manual entry of new galaxy data through a human-in-the-loop interface, inserts the data into DuckDB, and runs data quality checks to ensure the values are within acceptable ranges.

   

   The Dag has these tasks:

   • enter_galaxy_details: Pauses the pipeline and prompts a user to manually enter galaxy information (name, distances, type, and characteristics) through a form interface.
   • insert_galaxy_details: Inserts the user-provided galaxy data into the DuckDB table using the values collected from the previous task.
   • dq_checks: Validates the data quality by checking that distance values are within acceptable ranges (between 10,000 and 900,000 light years).

   The tasks have these dependencies:

   • enter_galaxy_details → insert_galaxy_details (user input needed before insertion)
   • insert_galaxy_details → dq_checks (data must be inserted before validation)
   Airflow Concepts

   • Human-in-the-loop: Human-in-the-loop workflows are processes that require human intervention, for example, to approve or reject an AI generated output, or choose a branch in a Dag depending on the result of an upstream task.
   • SQL operators: The common SQL provider is a great place to start when looking for SQL-related operators. It includes the SQLExecuteQueryOperator operator, which is a generic operator that can be used with a variety of databases, including Snowflake and Postgres. It also comes with data quality related operators, like the SQLColumnCheckOperator.
   • Parameters: You can use parameters to have dynamic queries with placeholders. These will be handled on database-driver level.

3. Click Sync to Test (top right) to sync your changes to the test deployment.

4. After the sync process finishes, head back to the Airflow UI.

5. Open the Dags view, and a new Dag should appear in the list.

Notice how the schedule is set to be triggered whenever the asset named duckdb://include/astronomy.db/galaxy_data is updated.

The first Dag updates this asset when data is loaded to DuckDB by using the outlets parameter:

1
    @task(outlets=[Asset(_DUCKDB_TABLE_URI)])
2
    def load_galaxy_data(
3
        filtered_galaxy_df: pd.DataFrame,
4
        duckdb_instance_name: str = _DUCKDB_INSTANCE_NAME,
5
        table_name: str = _DUCKDB_TABLE_NAME,
6
    ):
7
    # ...

Step 4.5: Try your advanced Dag

Time to see asset-aware scheduling and your new Dag in action.

1. Trigger example_etl_galaxies and observe what happens.

   You’ll notice that galaxy_maintenance starts when example_etl_galaxies finishes. More precisely, when it updates the asset that triggers the other Dag.

2. While galaxy_maintenance is running, open the latest run and you’ll notice there’s a required action. This is part of the human-in-the-loop feature: your task is waiting for user input.

   

3. Take time to explore the Airflow UI and see where these required actions are visible!

4. Open the required action to see the form defined in the code, and enter the following details:

   • name: Astro
   • distance_from_milkyway: 10000
   • distance_from_solarsystem: 10000
   • type_of_galaxy: Dwarf
   • characteristics: Looks amazing

   

5. Click OK and observe how the pipeline proceeds. Pay close attention to the dq_checks task, which successfully validates the data.

6. Try it again by running galaxy_maintenance once more. This time, enter 42 as the distance and observe how the dq_checks task fails because the data quality check detected an issue with your galaxy data.

Conclusion and next steps

Congratulations 🎉! You’ve just built two interconnected data pipelines using Apache Airflow, and along the way you’ve learned the fundamental concepts that power modern data orchestration.

In this tutorial, you:

• Set up a complete Airflow development environment in minutes using Astro IDE.
• Built and ran your first ETL pipeline with extraction, transformation, and loading steps.
• Mastered core Airflow concepts: Dags, tasks, operators, and dependencies.
• Extended your project with provider packages and database connections.
• Implemented human-in-the-loop workflows for manual data entry.
• Added automated data quality checks to ensure data integrity.
• Used asset-aware scheduling to create a dependency between two Dags.

Ready to dive deeper?

Explore more guides:

• An introduction to Apache Airflow®
• Introduction to Dags
• An introduction to the Airflow UI
• Using Airflow to Execute SQL
• Assets and data-aware scheduling in Airflow
• Get started with Airflow using the Astro CLI
• Apache Airflow® GenAI Quickstart

Join the academy and get certified:

• Airflow 101 Learning Path
• Airflow Dag Authoring Learning Path
• Astro Onboarding Learning Path