Integrate a Database with Apache Airflow 3

Table of Contents

1. Challenge

   Build a Database Pipeline with Apache Airflow 3

   Build a Database Pipeline with Apache Airflow 3

   In this lab, you will build a complete extract, transform, and load pipeline with Apache Airflow 3.3.1 and SQLite. The pipeline begins with 10 employee records, applies two business transformations, and writes the processed records to a new relational table.

   Airflow will coordinate the work as one ordered DAG. You will define database queries, Python processing, task dependencies, and verification as connected operations that you can trigger and monitor from the Airflow interface.

   ---

   🟦 Why It Matters

   Database pipelines must read operational data, apply consistent rules, and store results that downstream systems can use. Integrating those operations with Airflow makes the workflow repeatable, observable, and easier to troubleshoot.

   • Centralize database access: Reuse one Airflow Connection instead of placing a database path throughout the DAG code.
   • Run SQL as a managed task: Include database operations in scheduling, dependency management, logging, and task-state tracking.
   • Separate pipeline stages: Test extraction, processing, and loading independently while preserving one end-to-end workflow.
   • Apply repeatable transformations: Use Pandas to apply the same business rules to every employee record.
   • Verify orchestration and data: Confirm both the Airflow run and the rows written to SQLite.

   ---

   🔍 What You Will Build

   The completed DAG contains five ordered tasks:

   | Pipeline stage | Airflow task | Result | |---|---|---| | Query | query_employee_count | Confirms that the source table contains 10 records | | Extract | extract_employee_data | Writes six employee columns to /tmp/extracted_employees.csv | | Transform | process_and_save_data | Updates salaries, categorizes experience, and writes /tmp/processed_employees.csv | | Prepare destination | create_processed_table | Recreates the seven-column processed_employees table | | Load | insert_processed_data | Inserts all 10 transformed records into SQLite |

   DAG order:

   query_employee_count → extract_employee_data → process_and_save_data → create_processed_table → insert_processed_data

   ---

   🔑 Key Concepts

   Airflow Connections

   • Purpose: Store the information that hooks and operators use to reach an external system.
   • Implementation: Startup configures sqlite_default to point to $HOME/airflow/airflow.db.
   • Use in this lab: Reuse the Connection with both SQLExecuteQueryOperator and SqliteHook.

   Running SQL with SQLExecuteQueryOperator

   • Purpose: Execute SQL as a managed Airflow task.
   • Implementation: Query the source row count and create the destination table through sqlite_default.
   • Use in this lab: Confirm the 10 source rows before extraction and prepare the destination before loading.

   Extracting and Loading with SqliteHook

   • Purpose: Work with SQLite from Python while using an Airflow-managed Connection.
   • Implementation: Retrieve query results as a Pandas DataFrame and obtain a database connection for a parameterized insert.
   • Use in this lab: Move employee data from the source table, through CSV processing, into the destination table.

   Transforming Data with Pandas

   • Purpose: Apply the same business rules to every extracted record.
   • Implementation: Increase each salary by 10 percent, round it to two decimal places, and add an experience category.
   • Use in this lab: Classify employees with fewer than 10 years of experience as Junior and all other employees as Senior.

   Orchestrating and Verifying the DAG

   • Purpose: Control execution order and make every stage observable.
   • Implementation: Define five tasks, connect their dependencies, trigger the DAG, and inspect its run state.
   • Use in this lab: Confirm that every task succeeds and that processed_employees contains the expected 10 transformed rows.

   ---

   🟩 Learning Objectives

   By the end of this lab, you will be able to:

   • Configure Airflow SQL tasks to use a SQLite Connection through conn_id.
   • Query a database table with SQLExecuteQueryOperator and expose the returned row count in the task log.
   • Extract employee data with SqliteHook and save the results in a structured CSV file.
   • Process extracted data in Python by applying salary and experience transformations with Pandas.
   • Create a destination table and write processed data to SQLite with parameterized database operations.
   • Define and run an Airflow 3 DAG that executes the complete database pipeline in the required order.
   • Verify the Airflow run and database output with task checks and representative processed records.

   ---

   🧰 Starting Environment

   The lab provides:

   • Apache Airflow: Version 3.3.1 at localhost:8081.
   • SQLite database: $HOME/airflow/airflow.db.
   • Airflow Connection: sqlite_default, prepared by startup for the learner code.
   • Source data: An employees table containing 10 synthetic employee records.
   • Starter file: dataextract.py.
   • Lab tools: Terminal, Filetree, code editor, and Web Browser.

   You will begin by creating the source-count operator in dataextract.py, then build the pipeline one code task at a time.

   info> If you get stuck, you can refer to the provided solution code for each task, available in the solution folder and as a link after your first attempt at each task.

   Select the Next Step arrow to query and extract the employee data.

2. Challenge

   Query and Extract Employee Data

   Prepare the Source Data for Processing

   In this part of the lab, you will use the supplied sqlite_default Connection in the DAG code, confirm the number of source records with SQLExecuteQueryOperator, and extract the employee rows to a CSV file with SqliteHook.

   By the end of this part, the DAG will contain two connected tasks. The first task will verify that the employees table contains 10 records. The second task will retrieve the six source columns and write /tmp/extracted_employees.csv for downstream processing.

   ---

   🟦 Why It Matters

   • Separate access details from task logic: Airflow Connections let operators and hooks reuse the same database configuration.
   • Validate the source early: A row-count query prevents an empty or unexpected dataset from moving silently through the pipeline.
   • Make the SQL operation observable: SQLExecuteQueryOperator records the query state and returned count in the Airflow task log.
   • Use managed Python access: SqliteHook retrieves database rows in a format that Pandas can process.
   • Create a clear pipeline handoff: A structured extraction file keeps database access separate from the transformations applied later.

   ---

   🔍 In Airflow, You Will

   • Use the Connection: Configure the query operator and extraction Hook with sqlite_default.
   • Query the source count: Create query_employee_count with SQLExecuteQueryOperator and run SELECT COUNT(*) FROM employees.
   • Expose the result: Write the returned count to the task log.
   • Extract the records: Complete extract_employee_data() with SqliteHook and a six-column SQL query.
   • Create the handoff file: Save all 10 records to /tmp/extracted_employees.csv without a DataFrame index.
   • Set the dependency: Run query_employee_count before extract_task.
   • Validate the result: Check both the DAG structure and the extracted CSV output.

   ---

   🗃 Source Dataset

   The employees table contains structured information about company employees, including department assignments, salary values, and years of experience.

   Employees Table Schema

   | Column | Type | Description | |---|---|---| | employee_id | Integer | Unique employee identifier | | name | Text | Full name of the employee | | age | Integer | Employee age | | department | Text | Department to which the employee belongs | | salary | Real | Current employee salary | | experience | Integer | Years of professional experience |

   The extraction query orders the rows by employee_id, giving the CSV a predictable order that the later checks can verify.

   ---

   ✅ Expected Result

   When this part is complete:

   • Source count: query_employee_count returns 10.
   • Row count: /tmp/extracted_employees.csv contains 10 data rows.
   • Output schema: The CSV contains exactly the six source columns shown above.
   • Record order: Alice Johnson is the first record and Jane Miller is the final record.
   • Task order: The query task runs before the extraction task.

   Complete Tasks 1.1 through 1.4 in order, then review the extraction result before continuing. ### 🔍 Observation: Trigger the DAG and Verify Extracted Data

   Now that the source-count query and extraction function are complete, trigger the Airflow DAG and inspect the extracted CSV. This checkpoint confirms that Airflow can query SQLite, retrieve all 10 employee records, and create the handoff file used by the processing tasks.

   🛠 Execute and Verify

   If Terminal does not recognize airflow, run source ~/.bashrc once, then retry the command.

   1. In Terminal, trigger the DAG:

      airflow dags trigger extract_employee_data
      

   2. In Web Browser, open localhost:8081, select Dags, and open extract_employee_data.

   3. Open the new DAG run and confirm that query_employee_count and extract_employee_data show Success.

   4. Open the query_employee_count task log and confirm that the operator output contains 10.

   5. Return to Terminal and display the extracted CSV:

      cat /tmp/extracted_employees.csv
      

   info> If you have trouble navigating the embedded Airflow interface, select the pop-out icon in the upper-right corner of Web Browser to open it in a new window.

   ✅ Expected Outcome

   The command displays the six-column extracted dataset:

   employee_id,name,age,department,salary,experience
   1,Alice Johnson,34,HR,65000.0,10
   2,Bob Smith,28,Engineering,85000.0,5
   3,Charlie Brown,45,Finance,92000.0,18
   4,Diana Green,29,Marketing,75000.0,7
   5,Ethan White,40,HR,60000.0,12
   6,Fiona Black,35,Engineering,97000.0,9
   7,George Clark,50,Finance,110000.0,25
   8,Hannah Blue,31,Marketing,72000.0,6
   9,Ian Gray,42,Engineering,102000.0,20
   10,Jane Miller,37,HR,70000.0,14
   

   If the file is missing, empty, or contains unexpected values:

   • Confirm that query_employee_count and extract_employee_data completed successfully.
   • Open the failed task instance in Airflow and inspect its log.
   • Review Tasks 1.1 through 1.4, then validate the affected task again.

   The extraction checkpoint is complete. /tmp/extracted_employees.csv contains all 10 source records, Alice Johnson is the first record, and Jane Miller is the final record.

   Select the Next Step arrow to transform the extracted records.

3. Challenge

   Process Extracted Employee Data

   Transform the Extracted Employee Records

   In this part of the lab, you will turn the extracted employee records into a consistent processed dataset. The processing function will load /tmp/extracted_employees.csv into a Pandas DataFrame, retain the required source columns, apply two business rules, and save the result to /tmp/processed_employees.csv.

   By the end of this part, every employee record will include an adjusted salary and a standardized experience category. The processed CSV will contain seven columns and will be ready for the database-loading tasks.

   ---

   🟦 Why It Matters

   • Keep pipeline responsibilities clear: Separating processing from extraction makes each stage easier to understand and test.
   • Protect the output schema: Selecting the required source columns prevents unexpected CSV columns from entering the processed dataset.
   • Apply consistent business rules: One DataFrame calculation updates all 10 salary values in the same way.
   • Create an analysis-ready category: Converting years of experience into Junior or Senior creates a meaningful business grouping.
   • Preserve an auditable handoff: The processed CSV can be inspected before the records are loaded into SQLite.

   ---

   🔍 In Airflow, You Will

   • Load the extraction file: Read /tmp/extracted_employees.csv with pd.read_csv().
   • Preserve the source schema: Retain the six columns listed in EMPLOYEE_COLUMNS.
   • Update every salary: Increase each value by 10 percent and round the result to two decimal places.
   • Categorize experience: Create experience_category from each employee's years of experience.
   • Save the Processed Dataset: Write all seven output columns to /tmp/processed_employees.csv without a DataFrame index.
   • Validate the transformations: Check the complete output and representative processed values.

   ---

   🧮 Transformation Rules

   | Output value | Rule | Example | |---|---|---| | Updated salary | Current salary multiplied by 1.10, rounded to two decimals | 65000.00 becomes 71500.00 | | Junior | Fewer than 10 years of experience | Bob Smith, 5 years | | Senior | 10 years of experience or more | Alice Johnson, 10 years |

   The 10-year boundary belongs to the Senior category. This means an employee with exactly 10 years of experience must not be classified as Junior.

   ---

   🗃 Processed Dataset Schema

   The processed CSV retains all six source columns and adds one derived column:

   | Column | Source or derived | Description | |---|---|---| | employee_id | Source | Unique employee identifier | | name | Source | Full name of the employee | | age | Source | Employee age | | department | Source | Department assignment | | salary | Transformed | Salary after the 10 percent increase | | experience | Source | Years of professional experience | | experience_category | Derived | Junior or Senior classification |

   ---

   ✅ Expected Result

   When this part is complete:

   • Output size: /tmp/processed_employees.csv contains 10 rows and seven columns.
   • Senior boundary: Alice Johnson has salary 71500.00 and category Senior.
   • Junior example: Bob Smith has salary 93500.00 and category Junior.
   • Record integrity: No source employee record is removed or duplicated.

   Complete Tasks 2.1 through 2.4 in order, then review the processed output before loading it into SQLite. ### 🔍 Observation: Trigger the DAG and Verify Processed Data

   Now that the employee records have been transformed, trigger the Airflow DAG and inspect the processed CSV. This checkpoint confirms that the salary adjustment and experience categorization produce the expected values before the records are loaded into SQLite.

   🛠 Execute and Verify

   1. In Terminal, trigger the DAG:

      airflow dags trigger extract_employee_data
      

   2. In Web Browser, open localhost:8081, select Dags, and open extract_employee_data.

   3. Open the new DAG run and wait until process_and_save_data shows Success.

   4. Return to Terminal and display the processed CSV:

      cat /tmp/processed_employees.csv
      

   ✅ Expected Outcome

   The command displays the seven-column processed dataset:

   employee_id,name,age,department,salary,experience,experience_category
   1,Alice Johnson,34,HR,71500.0,10,Senior
   2,Bob Smith,28,Engineering,93500.0,5,Junior
   3,Charlie Brown,45,Finance,101200.0,18,Senior
   4,Diana Green,29,Marketing,82500.0,7,Junior
   5,Ethan White,40,HR,66000.0,12,Senior
   6,Fiona Black,35,Engineering,106700.0,9,Junior
   7,George Clark,50,Finance,121000.0,25,Senior
   8,Hannah Blue,31,Marketing,79200.0,6,Junior
   9,Ian Gray,42,Engineering,112200.0,20,Senior
   10,Jane Miller,37,HR,77000.0,14,Senior
   

   If the file is missing, empty, or contains unexpected values:

   • Confirm that extract_employee_data completed successfully.
   • Confirm that process_and_save_data completed successfully.
   • Open the failed task instance in Airflow and inspect its log.
   • Review Tasks 2.1 through 2.4, then validate the affected task again.

   The processing checkpoint is complete. /tmp/processed_employees.csv contains all 10 source records, the adjusted salary values, and the correct Junior or Senior category for every employee.

   Select the Next Step arrow to load the processed records into SQLite.

4. Challenge

   Load Processed Data into SQLite

   Complete and Verify the Database Pipeline

   In this final part of the lab, you will store the processed employee records in a structured SQLite table and run the complete Airflow pipeline. One task will recreate the destination table, and another will insert all 10 transformed records with a parameterized database operation.

   By the end of this part, the DAG will contain five tasks connected in the required order. You will trigger the DAG from the Airflow interface, confirm that each task succeeds, and verify that the destination table contains the expected processed values.

   ---

   🟦 Why It Matters

   • Preserve the processed schema: A dedicated destination table stores both source fields and the derived experience category.
   • Make database changes observable: SQLExecuteQueryOperator places table creation inside the managed DAG.
   • Keep repeated runs predictable: Dropping the earlier destination table prevents old rows from affecting the result.
   • Insert values safely: A parameterized executemany() operation keeps employee values separate from the SQL statement.
   • Protect data integrity: Explicit dependencies ensure processing finishes before the table is created and rows are inserted.
   • Verify orchestration and data: The final checks prove that a successful DAG run also produced the correct database result.

   ---

   🔍 In Airflow, You Will

   • Create the destination table: Define create_processed_table_task with SQLExecuteQueryOperator.
   • Rebuild the schema: Drop and recreate processed_employees with seven required columns.
   • Prepare the insert values: Read /tmp/processed_employees.csv and convert its DataFrame rows into tuples.
   • Load the records: Use a parameterized executemany() operation and commit all 10 rows to SQLite.
   • Complete the DAG: Connect the five tasks in one direct dependency chain.
   • Run the pipeline: Trigger extract_employee_data from the Airflow interface.
   • Verify the result: Confirm the latest DAG run, all five task states, and representative destination rows.

   ---

   🗃 Destination Table Schema

   | Column | Type | Description | |---|---|---| | employee_id | Integer, primary key | Unique employee identifier | | name | Text | Full name of the employee | | age | Integer | Employee age | | department | Text | Department assignment | | salary | Real | Processed salary value | | experience | Integer | Years of professional experience | | experience_category | Text | Junior or Senior classification |

   ---

   🔗 Complete DAG Order

   The final dependency chain is:

   query_employee_count → extract_employee_data → process_and_save_data → create_processed_table → insert_processed_data

   This order confirms the source before extraction, transforms the extracted rows before preparing the destination, and creates the destination before inserting records.

   ---

   ✅ Expected Result

   When the complete pipeline succeeds:

   • DAG state: The latest extract_employee_data run shows Success.
   • Task states: All five task instances show Success.
   • Destination count: processed_employees contains exactly 10 rows.
   • Senior sample: Alice Johnson has salary 71500.00 and category Senior.
   • Junior sample: Bob Smith has salary 93500.00 and category Junior.
   • Query output: The query_employee_count task log includes a returned count of 10.

   Complete Tasks 3.1 through 3.4 in order to finish the DAG code. Then run and verify the completed pipeline. ### 🔍 Observation: Trigger the DAG and Verify Stored Processed Data

   Now that the pipeline code is complete, trigger the Airflow DAG and verify that the transformed employee records are stored in SQLite. This observation confirms both sides of the workflow: Airflow successfully orchestrated all five tasks, and the database contains the expected processed values.

   🛠 Execute and Verify

   1. In Terminal, trigger the completed DAG:

      airflow dags trigger extract_employee_data
      

   2. In Web Browser, open localhost:8081, select Dags, and open extract_employee_data.

   3. Open the new DAG run and wait until all five task instances show Success.

   4. Open the query_employee_count task log and confirm that the operator output contains 10.

   5. Return to Terminal and query the destination table:

      python -c 'import sqlite3, pathlib; rows = sqlite3.connect(pathlib.Path.home() / "airflow/airflow.db").execute("SELECT * FROM processed_employees ORDER BY employee_id"); print("employee_id|name|age|department|salary|experience|experience_category"); print(*( "|".join(map(str, row)) for row in rows), sep="\n")'
      

   ✅ Expected Outcome

   The latest DAG run and all five task instances show Success. The database query displays these 10 rows:

   employee_id|name|age|department|salary|experience|experience_category
   1|Alice Johnson|34|HR|71500.0|10|Senior
   2|Bob Smith|28|Engineering|93500.0|5|Junior
   3|Charlie Brown|45|Finance|101200.0|18|Senior
   4|Diana Green|29|Marketing|82500.0|7|Junior
   5|Ethan White|40|HR|66000.0|12|Senior
   6|Fiona Black|35|Engineering|106700.0|9|Junior
   7|George Clark|50|Finance|121000.0|25|Senior
   8|Hannah Blue|31|Marketing|79200.0|6|Junior
   9|Ian Gray|42|Engineering|112200.0|20|Senior
   10|Jane Miller|37|HR|77000.0|14|Senior
   

   If the run fails or the records are missing:

   • Confirm that the latest DAG run was triggered after Task 3.4 was completed and validated.
   • Open the failed task instance and inspect its log.
   • Confirm that create_processed_table completed before insert_processed_data.
   • Review the affected task, then select Validate again.

   The complete pipeline has now extracted, transformed, and stored all 10 employee records.

   Select the Next Step arrow to review what you accomplished.

5. Challenge

   Pipeline Complete

   ✅ Pipeline Complete

   You built, ran, and verified a five-task database pipeline with Apache Airflow 3. The successful DAG run moved 10 employee records from the source table, through repeatable transformations, into the processed_employees table.

   🟩 What You Accomplished

   1. Configured Airflow operators and Hooks to use the supplied sqlite_default Connection.
   2. Queried the employees table with SQLExecuteQueryOperator and confirmed its 10 source records.
   3. Extracted six employee columns to /tmp/extracted_employees.csv with SqliteHook.
   4. Increased each salary by 10 percent and rounded the result to two decimal places with Pandas.
   5. Categorized employees as Junior or Senior according to their years of experience.
   6. Recreated the seven-column processed_employees table and inserted all transformed records with a parameterized operation.
   7. Connected the five Airflow tasks in the required query, extract, transform, prepare, and load order.
   8. Triggered the DAG and verified both the successful Airflow run and the final SQLite output.

   🔍 Verified Result

   • DAG state: The latest extract_employee_data run is successful.
   • Task states: All five task instances are successful.
   • Database result: processed_employees contains 10 rows.
   • Representative values: Alice Johnson has a salary of 71500.00 and a Senior category. Bob Smith has a salary of 93500.00 and a Junior category.

   The completed workflow demonstrates how Airflow can coordinate database operations, Python processing, dependencies, logs, and verification as one observable pipeline.

   🎉 Congratulations, You Have Completed the Lab!

   info> Select the Next Step arrow one more time to complete the lab and mark your progress as 100%.