> ## 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.
# 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.
No matter if you are an absolute Airflow beginner or already know about certain concepts, in 5 minutes from now, you will have your first data pipeline (a Dag) running in a fully functional Airflow environment.
Get a fully functional Airflow environment running in your browser with zero local setup using Astro IDE.
Create and run an ETL pipeline that processes galaxy data with extraction, transformation, and loading steps.
Learn Dags, tasks, operators, dependencies, and asset-aware scheduling through hands-on practice.
## Step 1: Set up your Astro trial and Astro IDE
1. The first step is to [start a free Astro trial](https://www.astronomer.io/lp/signup/?utm_source=website\&utm_medium=learn-guides\&utm_campaign=learn-intro-tutorial-11-25).
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](/docs/learn/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](https://duckdb.org/) 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](/docs/learn/rerunning-dags#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.
You just set up your Airflow development environment, started your first Airflow environment, and deployed and ran your first Dag. Take a moment to check the time and see how quickly you got there.
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](/docs/learn/airflow-ui) 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.
```python expandable wrap theme={null}
# imports
from airflow.sdk import Asset, chain, Param, dag, task
# ...
# constants
_DUCKDB_INSTANCE_NAME = os.getenv("DUCKDB_INSTANCE_NAME", "include/astronomy.db")
_DUCKDB_TABLE_NAME = os.getenv("DUCKDB_TABLE_NAME", "galaxy_data")
_DUCKDB_TABLE_URI = f"duckdb://{_DUCKDB_INSTANCE_NAME}/{_DUCKDB_TABLE_NAME}"
# ...
# Dag definition
@dag(...)
def example_etl_galaxies():
# tasks
@task(retries=2)
def create_galaxy_table_in_duckdb(...):
# ...
@task
def extract_galaxy_data(...):
# ...
@task
def transform_galaxy_data(...):
# ..
@task
def load_galaxy_data(...):
# ...
@task
def print_loaded_galaxies(...):
# ...
# create task instances and define implicit dependencies
create_galaxy_table_in_duckdb_obj = create_galaxy_table_in_duckdb()
extract_galaxy_data_obj = extract_galaxy_data()
transform_galaxy_data_obj = transform_galaxy_data(extract_galaxy_data_obj)
load_galaxy_data_obj = load_galaxy_data(transform_galaxy_data_obj)
# define explicit dependencies
chain(
create_galaxy_table_in_duckdb_obj, load_galaxy_data_obj, print_loaded_galaxies()
)
# Instantiate the Dag
example_etl_galaxies()
```
**Airflow Concepts**
* **[Dag](/docs/learn/dags)**: Your entire pipeline from start to finish, consisting of one or more tasks.
* **[Task](/docs/learn/intro-to-airflow#airflow-concepts)**: A unit of work within your pipeline.
* **[Operator/Decorator](/docs/learn/what-is-an-operator)**: The template/class that defines what work a task does, serving as the building blocks of pipelines.
* Traditional: `task = PythonOperator(...)` → returns operator directly.
* [TaskFlow API](/docs/learn/airflow-decorators): `@task def my_task(): ...` → creates operator, wrapped in `XComArg`.
* Many decorators available: `@task`, `@task.bash`, `@task.docker`, `@task.kubernetes`, etc.
* `XComArg`: Wrapper enabling automatic data passing and dependency inference.
## 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:
Add provider packages to extend Airflow with new operators and integrations for databases and external systems.
Set up proper Airflow connections to manage credentials and configurations for external tools.
Implement human-in-the-loop workflows that pause for manual data entry and human decision-making.
Use common SQL operators to run parameterized queries across different database systems.
Add automated data quality checks to ensure data integrity throughout your pipelines.
Trigger Dags based on asset-aware scheduling rather than time schedules for data-driven workflows.
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:
```python wrap theme={null}
cursor = duckdb.connect(duckdb_instance_name)
```
While this works, Airflow offers a better approach: common SQL operators that execute queries using [Airflow connections](/docs/learn/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](https://airflow.apache.org/docs/apache-airflow-providers/) 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:
```text wrap theme={null}
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**](/docs/learn/connections) stores configuration details for connecting to external tools in your data ecosystem. Most hooks ([what is a hook?](/docs/learn/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-4-implement-dag-with-human-in-the-loop).
**Astro Concepts**
You just added a connection to this deployment (a single Airflow instance). If you deployed your Dags to another environment or recreated the test deployment, you'd need to add the connection again. Astro offers a helpful solution: under **Environment** → **Connections** in the Astro platform, you can set up workspace-wide connections that are available across all your Airflow instances. See [Manage Airflow connections and variables](/docs/astro/manage-connections-variables) in the Astro documentation.
**Airflow Concepts**
* **Provider package**: Provider packages are installable modules that contain pre-built decorators, operators, hooks, and sensors for integrating with external services and extending Airflow functionality.
* **Connection**: Connections in Airflow are sets of configurations used to connect with other tools in the data ecosystem.
### 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](#step-4-4-implement-dag-with-human-in-the-loop)), 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:
```python expandable wrap theme={null}
from airflow.sdk import chain, dag, Asset, Param
from airflow.providers.standard.operators.hitl import HITLEntryOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator, SQLColumnCheckOperator
_DUCKDB_TABLE_URI = "duckdb://include/astronomy.db/galaxy_data"
_DUCKDB_CONN_ID = "duckdb_astronomy"
galaxy_table_asset = Asset(_DUCKDB_TABLE_URI)
@dag(schedule=galaxy_table_asset)
def galaxy_maintenance():
_enter_galaxy_details = HITLEntryOperator(
task_id="enter_galaxy_details",
subject="Please provide required information: ",
params={
"name": Param("", type="string"),
"distance_from_milkyway": Param(10000, type="number"),
"distance_from_solarsystem": Param(10000, type="number"),
"type_of_galaxy": Param("Dwarf", type="string", enum=[
"Dwarf Spheroidal",
"Dwarf",
"Irregular",
"Spiral"
]),
"characteristics": Param("", type="string")
}
)
_insert_galaxy_details = SQLExecuteQueryOperator(
task_id="insert_galaxy_details",
conn_id=_DUCKDB_CONN_ID,
show_return_value_in_logs=True,
sql="""
-- in case db was removed due to sync
CREATE TABLE IF NOT EXISTS galaxy_data (
name STRING PRIMARY KEY,
distance_from_milkyway INT,
distance_from_solarsystem INT,
type_of_galaxy STRING,
characteristics STRING
);
INSERT OR IGNORE INTO galaxy_data BY NAME
SELECT
$name AS name,
$distance_from_milkyway AS distance_from_milkyway,
$distance_from_solarsystem AS distance_from_solarsystem,
$type_of_galaxy AS type_of_galaxy,
$characteristics AS characteristics
""",
parameters={
"name": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['name'] }}",
"distance_from_milkyway": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['distance_from_milkyway'] }}",
"distance_from_solarsystem": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['distance_from_solarsystem'] }}",
"type_of_galaxy": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['type_of_galaxy'] }}",
"characteristics": "{{ task_instance.xcom_pull('enter_galaxy_details')['params_input']['characteristics'] }}"
}
)
_galaxy_dq_checks = SQLColumnCheckOperator(
task_id="dq_checks",
conn_id=_DUCKDB_CONN_ID,
table="galaxy_data",
column_mapping={
"distance_from_milkyway": {
"min": {"geq_to": 10000},
"max": {"leq_to": 900000},
},
"distance_from_solarsystem": {
"min": {"geq_to": 10000},
"max": {"leq_to": 900000},
},
},
)
chain(_enter_galaxy_details, _insert_galaxy_details, _galaxy_dq_checks)
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](/docs/learn/airflow-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](/docs/learn/airflow-branch-operator) in a Dag depending on the result of an upstream task.
* **SQL operators**: The [common SQL provider](https://airflow.apache.org/registry/providers/common-sql/) is a great place to start when looking for [SQL-related operators](/docs/learn/airflow-sql). It includes the [`SQLExecuteQueryOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-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`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-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:
```python wrap theme={null}
@task(outlets=[Asset(_DUCKDB_TABLE_URI)])
def load_galaxy_data(
filtered_galaxy_df: pd.DataFrame,
duckdb_instance_name: str = _DUCKDB_INSTANCE_NAME,
table_name: str = _DUCKDB_TABLE_NAME,
):
# ...
```
**Airflow Concepts**
* **Asset** (object): Logical representation of data (table, model, file) used to establish dependencies. Can be used imperatively (code-based) or declaratively (implicit definition via the [`@asset`](/docs/learn/airflow-datasets#asset-definition) decorator). It is an abstract representation of data.
* **Asset event**: Each time an asset is updated, the system creates an asset event object. This object includes the ID of the Dag that produced the update, the update timestamp, and optional custom information.
* **Asset-aware scheduling**: Set the `schedule` of a Dag to one or more assets, optionally with a logical expression using AND (`&`) and OR (`|`) operators, so that the Dag is triggered when these assets receive asset update events.
* **Producer task**: a task that produces updates to one or more assets provided to its `outlets` parameter, creating asset events when it completes successfully.
* **Materialize**: Running a producer task, which updates an asset.
* **@asset**: Declarative shortcut (Dag + task + asset(s) in one).
### Step 4.5: Try your advanced Dag
Time to see asset-aware scheduling and your new Dag in action.
This tutorial stores the DuckDB database in a project file (`include/astronomy.db`). The `example_etl_galaxies` Dag creates a table in this database, but the file isn't included in the auto-generated project repository. As a result, each time you sync changes to the deployment, the database file disappears. To handle this, the `insert_galaxy_details` task in the second Dag uses `CREATE TABLE IF NOT EXISTS` in case the database file was removed between runs. To improve this, you could use a persistent database service, for example Snowflake or BigQuery.
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®](/docs/learn/intro-to-airflow)
* [Introduction to Dags](/docs/learn/dags)
* [An introduction to the Airflow UI](/docs/learn/airflow-ui)
* [Using Airflow to Execute SQL](/docs/learn/airflow-sql)
* [Assets and data-aware scheduling in Airflow](/docs/learn/airflow-datasets)
* [Get started with Airflow using the Astro CLI](/docs/cli/v1.43/get-started-cli)
* [Apache Airflow® GenAI Quickstart](/docs/learn/airflow-quickstart-genai)
**Join the academy and get certified:**
* [Airflow 101 Learning Path](https://academy.astronomer.io/path/airflow-101)
* [Airflow Dag Authoring Learning Path](https://academy.astronomer.io/path/airflow-dag-authoring)
* [Astro Onboarding Learning Path](https://academy.astronomer.io/path/astro-onboarding)