IoT Health Pipeline with dbt

Build a tested IoT sensor health pipeline with dbt, Snowflake, and Airflow.

Introduction

30 Second Summary

Sensors often resend a reading when corrected data arrives after the original. A dashboard that trusts every row can overcount activity without showing why.

In this project, you will build an IoT sensor health pipeline that turns raw readings into hourly device metrics in Snowflake. You will use dbt Core to expose a duplicate before Apache Airflow runs the corrected workflow from seed to test.

What You'll Build

When you trigger the finished workflow, three Airflow tasks turn raw sensor readings into a Snowflake health table with trustworthy hourly metrics.

By the end of this project, you'll have:

  • A rerunnable hourly health table showing each device's reading count, average temperature, minimum battery level, and demo attention flag.
  • A visible duplicate-reading test that fails on the inflated metric before deterministic deduplication makes it pass.
  • An observable three-task workflow whose task states and logs show the pipeline progressing from seed to transform to test.
  • Secret Mission: a custom battery-range test that proves Airflow blocks out-of-range sensor readings.

Are there any prerequisites?

You need a Mac for the local setup. The guide walks you through every required tool plus the Snowflake trial account.

Before We Start

This is your moment to commit to building a local IoT sensor-health pipeline with dbt, Snowflake, and Airflow. Its goal is to remove duplicate readings before they distort hourly health metrics.

Install the Local Toolchain

Reliable sensor metrics depend on a reproducible local toolchain. dbt Core will transform the readings in later steps.

Apache Airflow will orchestrate those dbt commands. Separate Python environments prevent their dependencies from changing each other's runtime.

This step installs Visual Studio Code as your workspace. The uv tool manages Python 3.12.

In this step, get ready to:
  • Prepare Visual Studio Code with the project folder.
  • Install uv 0.12.13 with Python 3.12.
  • Build isolated environments for dbt Core 1.12.0 and Apache Airflow 3.3.1.
Install Visual Studio Code and open the project folder

Visual Studio Code keeps the project files beside an integrated terminal. The terminal starts inside the folder currently open in the editor.

You should now see Visual Studio Code inside your Applications folder.

  • Press Cmd+Space to open Spotlight.
  • Type Visual Studio Code.
  • Press Enter to open the editor.

You should see the Visual Studio Code welcome screen.

  • Press Cmd+Shift+P to open the Command Palette.
  • Type Shell Command: Install 'code' command in PATH.
  • Select Shell Command: Install 'code' command in PATH from the results.

The code command is now available to new terminal sessions.

  • Click File in the top menu bar.
  • Select Open Folder.
  • Choose your macOS home folder in the folder picker.
  • Select New Folder.
  • Type iot-sensor-pipeline as the folder name.
  • Press Enter to create the folder.
  • Select the new iot-sensor-pipeline folder.
  • Select Open.

The Explorer sidebar should now show iot-sensor-pipeline as the open folder.

  • Select View from the top menu.
  • Select Terminal to open the integrated terminal.
  • Verify the PATH command from the open project folder by running:
code .

What does this command do?

The code . command asks Visual Studio Code to open the terminal's current folder. The integrated terminal starts inside ~/iot-sensor-pipeline.

You should see iot-sensor-pipeline remain open without a terminal error. That confirms the project folder is ready and the code command is on your PATH.

Is the code command unavailable?

Return to the Command Palette and run Shell Command: Install 'code' command in PATH again. Close the current terminal before opening a new one from the View menu.

Still stuck? Help me make the Visual Studio Code command available in my macOS terminal.

Install uv and Python

The pinned uv release installs the shared Python runtime. Pinning uv to 0.12.13 keeps the setup commands consistent throughout the project.

  • Check whether uv is already available by running:
uv --version

What does this command do?

This command prints the installed uv version when the tool is available. Use the result to choose the matching tab below.

✔️ I see uv 0.12.13

Your pinned uv release is already available. Continue below to install Python 3.12.

ⓧ I see a different uv version

Install the pinned release so every later command uses the expected uv behavior.

  • Replace the current uv release by running this installer:
curl -LsSf https://astral.sh/uv/0.12.13/install.sh | sh

What does this command do?

This downloads the official uv installer for version 0.12.13. It installs the pinned command-line tool for your macOS user.

  • Select the trash can in the terminal panel to close the current shell.
  • Select View from the top menu.
  • Select Terminal to start a new shell.
  • Confirm the pinned release by running:
uv --version

What should I see?

The output should identify uv as version 0.12.13.

ⓧ Command not found

uv is absent from your PATH. Install the pinned release before continuing.

  • Install uv 0.12.13 by running this installer:
curl -LsSf https://astral.sh/uv/0.12.13/install.sh | sh

What does this command do?

This downloads the official uv installer for version 0.12.13. It installs the command-line tool for your macOS user.

  • Select the trash can in the terminal panel to close the current shell.
  • Select View from the top menu.
  • Select Terminal to start a new shell.
  • Confirm the installation by running:
uv --version

What should I see?

The output should identify uv as version 0.12.13.

Still cannot run uv?

Make sure you opened a new terminal after the installer completed. A terminal that stayed open during installation can retain the old PATH.

Need another pair of eyes? Help me troubleshoot my uv installation on macOS.

  • Install the shared Python 3.12 runtime by running:
uv python install 3.12

What does this command do?

uv downloads the latest available Python 3.12 patch release. The two project environments will both use this managed runtime.

The output should identify an installed Python 3.12 runtime.

Python download interrupted?

Run the same installation command again after checking your internet connection. uv can reuse completed download work.

Still blocked? Help me install Python 3.12 with uv on macOS.

Create isolated dbt and Airflow environments

Virtual environments give dbt and Airflow separate package directories. Each tool can keep its pinned dependencies without affecting the other environment.

  • Select the new-file control beside iot-sensor-pipeline in the Explorer sidebar.
  • Enter requirements.txt as the file name.

The new requirements.txt tab should open in the editor.

  • Fill requirements.txt with the pinned dbt packages by pasting:
dbt-core==1.12.0
dbt-snowflake==1.12.0

What does this file do?

The file pins dbt Core and the Snowflake adapter to compatible 1.12.0 releases. The environment sync step reads these exact package versions.

  • Save requirements.txt by pressing Cmd+S.
  • Confirm that requirements.txt appears in the Explorer sidebar.

Did the requirements file disappear?

Check that the file name is exactly requirements.txt inside iot-sensor-pipeline. Remove any extra file extension added to the name.

Need help? Help me create requirements.txt in the correct Visual Studio Code folder.

  • Select the new-file control beside iot-sensor-pipeline in the Explorer sidebar.
  • Enter .gitignore as the file name.

The new .gitignore tab should open in the editor.

  • Fill .gitignore with the project exclusions by pasting:
.venv-dbt/
.venv-airflow/
dbt/.env
dbt/target/
dbt/logs/
dbt/dbt_packages/

What does this file do?

These rules exclude both virtual environments from version control. They also exclude future dbt credentials and generated dbt output.

  • Save .gitignore by pressing Cmd+S.
  • Confirm that .gitignore appears in the Explorer sidebar.

Cannot see the gitignore file?

Check the Explorer sidebar for a file named exactly .gitignore. The leading dot is part of the file name.

Still missing? Help me create a .gitignore file in Visual Studio Code.

✔️ Awesome, I've got everything!

Great. Double check that requirements.txt and .gitignore are saved inside ~/iot-sensor-pipeline.

ⓧ I'd like to double check the full code

  • Compare your requirements.txt with this complete file:
dbt-core==1.12.0
dbt-snowflake==1.12.0

What should match?

Your file should contain these two package pins in this order. Each package uses version 1.12.0.

  • Compare your .gitignore with this complete file:
.venv-dbt/
.venv-airflow/
dbt/.env
dbt/target/
dbt/logs/
dbt/dbt_packages/

What should match here?

Your file should contain all six exclusions. The first two entries cover the environments created below.

  • Create the dbt environment and install its pinned packages by running:
uv venv --python 3.12 .venv-dbt
source .venv-dbt/bin/activate
uv pip sync requirements.txt
deactivate

What does this setup do?

  • The first line creates .venv-dbt with Python 3.12.
  • The activation line directs package installation into that environment.
  • The sync step installs the exact packages from requirements.txt.
  • The final line exits the environment after installation.
  • Verify the dbt environment by running:
./.venv-dbt/bin/dbt --version

What does this check prove?

This runs dbt directly from .venv-dbt. It avoids any dbt installation outside the project.

You should see dbt Core 1.12.0 with the Snowflake plugin at 1.12.0.

Did the dbt installation fail?

Confirm that the terminal is inside ~/iot-sensor-pipeline. Check that requirements.txt matches the complete file above.

Still blocked? Help me troubleshoot the isolated dbt installation.

The Airflow install has a larger dependency set. Expect the terminal to stay busy for several minutes while uv resolves and downloads the constrained packages.

  • Create the Airflow environment and install its pinned stack by running:
uv venv --python 3.12 .venv-airflow
source .venv-airflow/bin/activate
uv pip install "apache-airflow[standard]==3.3.1" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.3.1/constraints-3.12.txt"
deactivate

What does this setup do?

  • The first line creates .venv-airflow with Python 3.12.
  • The activation line directs package installation into that environment.
  • The constrained installation adds Apache Airflow 3.3.1 with the standard provider at 1.17.0.
  • The final line exits the environment after installation.

Before you run the final checks, which versions do you expect each tool to report?

  • Verify the complete local toolchain by running:
uv --version
./.venv-dbt/bin/dbt --version
./.venv-airflow/bin/airflow version

What do these checks prove?

The first check reads uv from your PATH. The remaining checks run dbt and Airflow directly from their isolated environments.

You should see uv 0.12.13. The dbt output should show Core 1.12.0 with the Snowflake plugin at 1.12.0.

The final command should report Apache Airflow 3.3.1. That is the local toolchain locked in for the rest of the pipeline.

Does one version check fail?

A missing dbt executable usually means the first environment installation did not complete. A missing Airflow executable usually means the constrained installation stopped before finishing.

Need help reading the result? Help me diagnose my uv, dbt, or Airflow version checks.

Your editor and isolated runtimes are ready. Next up, you'll connect dbt to the Snowflake warehouse that stores the sensor data.

Connect dbt to Snowflake

Your local toolchain is ready. Now you can give dbt Core a Snowflake destination for the sensor data.

The connection needs low-cost compute plus a target namespace. An authenticated profile points dbt at both.

In this step, get ready to:
  • Provision the Snowflake trial resources.
  • Store the dbt connection settings locally.
  • Verify that dbt can reach Snowflake.
Create the Snowflake warehouse

A Snowflake warehouse supplies the compute that runs your transformations. The database plus schema give every model a predictable destination.

Keep the trial free

Snowflake currently provides $400 in trial credits for up to 30 days. No payment information is required.

Adding a payment method converts the trial to paid usage. This project stays within the trial, so leave payment details unset.

  • Open the Snowflake trial signup page in your browser.
  • Complete the trial registration using a valid email address.
  • Follow the account verification link sent to your email.
  • Sign in to the Snowflake web interface.

You should now see the Snowflake web interface for your trial account. This confirms that your warehouse resources can be created.

The project keeps its setup statements in a reusable SQL file. This makes the Snowflake environment reproducible.

  • In the Visual Studio Code Explorer, select the New Folder button.
  • Enter sql as the folder name and press Enter.

You should see the new sql folder beneath iot-sensor-pipeline in the Explorer.

  • Select the sql folder in the Explorer.
  • Select the New File button.
  • Enter setup.sql as the file name and press Enter.

The empty setup.sql file opens in the editor. It will hold every statement needed to prepare Snowflake.

  • Add the Snowflake setup statements to setup.sql by pasting this code:
CREATE WAREHOUSE IF NOT EXISTS IOT_WH
  WAREHOUSE_SIZE = XSMALL
  AUTO_SUSPEND = 300
  AUTO_RESUME = TRUE
  INITIALLY_SUSPENDED = TRUE;

CREATE DATABASE IF NOT EXISTS IOT_DB;
CREATE SCHEMA IF NOT EXISTS IOT_DB.ANALYTICS;

What does this setup create?

  • The IOT_WH warehouse uses the XSMALL size to limit compute usage.
  • The AUTO_SUSPEND setting stops compute after 300 seconds of inactivity.
  • The INITIALLY_SUSPENDED setting keeps compute off until a query needs it.
  • The IOT_DB.ANALYTICS namespace becomes the destination for the seed plus dbt models.
  • Save setup.sql by pressing Cmd+S on macOS or Ctrl+S on Windows.
  • Copy all SQL from setup.sql.
  • Open a SQL worksheet in Snowflake.
  • Paste the copied SQL into the worksheet.
  • Execute the worksheet statements.

You should see all three statements complete successfully. Snowflake now has IOT_WH plus IOT_DB.ANALYTICS ready for dbt.

Snowflake setup did not complete?

Confirm that you are signed into the trial account you just created. Run the complete contents of setup.sql in one Snowflake SQL worksheet.

Still blocked? Help me troubleshoot my Snowflake setup statements.

A matching cleanup file records how to remove these resources later. You will keep it locally until the project cleanup section.

  • Keep the sql folder selected and choose the New File button.
  • Enter cleanup.sql as the file name and press Enter.
  • Add the cleanup statements to cleanup.sql by pasting this code:
ALTER WAREHOUSE IF EXISTS IOT_WH SUSPEND;
DROP WAREHOUSE IF EXISTS IOT_WH;
DROP DATABASE IF EXISTS IOT_DB;

What does this cleanup file do?

The first statement suspends warehouse compute. The remaining statements remove the warehouse plus its project database.

Keep this file for the cleanup section. Running it now would remove the Snowflake resources you just created.

  • Save cleanup.sql by pressing Cmd+S on macOS or Ctrl+S on Windows.

You should now see setup.sql plus cleanup.sql inside the sql folder.

Connect dbt with a local profile

A dbt profile names the warehouse plus namespace that commands should use. Local environment variables supply the account credentials without placing them in the profile.

  • Select the iot-sensor-pipeline project heading in the Explorer.
  • Select the New Folder button.
  • Enter dbt as the folder name and press Enter.

You should see the dbt folder beside sql in the Explorer.

  • Select the dbt folder in the Explorer.
  • Select the New File button.
  • Enter profiles.yml as the file name and press Enter.
  • Add the dbt connection profile to profiles.yml by pasting this code:
iot_sensor_pipeline:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: "{{ env_var('DBT_SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('DBT_SNOWFLAKE_USER') }}"
      password: "{{ env_var('DBT_ENV_SECRET_PASSWORD') }}"
      warehouse: IOT_WH
      database: IOT_DB
      schema: ANALYTICS
      threads: 4

How does the profile connect?

  • The dev output tells dbt to use Snowflake.
  • The account plus user and password values come from local environment variables.
  • The warehouse plus database and schema values match the resources created in Snowflake.
  • The threads value allows dbt to run up to four operations concurrently.
  • Save profiles.yml by pressing Cmd+S on macOS or Ctrl+S on Windows.

The Explorer should now list profiles.yml beneath the dbt folder.

Profile indentation looks wrong?

YAML uses spaces to represent nesting. Compare each level of profiles.yml with the reference before running the connection check.

Need another pair of eyes? Help me check my dbt profile indentation.

Keep credentials local

The existing .gitignore excludes dbt/.env from version control. Your live Snowflake password stays in that local file.

Avoid showing the completed file in screenshots. The template below contains plain placeholders only.

  • Keep the dbt folder selected and choose the New File button.
  • Enter .env as the file name and press Enter.
  • Add the local credential placeholders to .env by pasting this template:
DBT_SNOWFLAKE_ACCOUNT=your-account-identifier
DBT_SNOWFLAKE_USER=your-snowflake-username
DBT_ENV_SECRET_PASSWORD=your-snowflake-password

How does dbt load these values?

dbt Core loads .env from the directory where you run it. Each profile expression reads the matching value from that file.

The DBT_ENV_SECRET_PASSWORD prefix marks the password as sensitive so dbt scrubs it from logs.

  • In the Snowflake sidebar, select your name.
  • Hover over Account.
  • Select View account details.
  • Find the Account identifier value in Config file.
  • Replace your-account-identifier in .env with that account identifier.
  • Replace your-snowflake-username with the username used for your trial account.
  • Replace your-snowflake-password with the password used for your trial account.
  • Save .env by pressing Cmd+S on macOS or Ctrl+S on Windows.

The saved .env now gives dbt the credentials referenced by profiles.yml.

Environment values not loading?

Confirm that the file is named exactly .env inside the dbt folder. Check that each line contains one variable name plus one value.

Need help without sharing your password? Help me troubleshoot my dbt environment file safely.

✔️ Awesome, I've got everything!

Your Snowflake setup plus local dbt configuration now match the project template. Keep your completed .env private.

ⓧ I'd like to double check the full code

Compare each project file with the references below. The .gitignore and requirements.txt files remain unchanged from the previous step.

.venv-dbt/
.venv-airflow/
dbt/.env
dbt/target/
dbt/logs/
dbt/dbt_packages/

What does this file protect?

This file keeps the virtual environments plus generated dbt files out of version control. It also excludes the credential-bearing dbt/.env file.

DBT_SNOWFLAKE_ACCOUNT=your-account-identifier
DBT_SNOWFLAKE_USER=your-snowflake-username
DBT_ENV_SECRET_PASSWORD=your-snowflake-password

What should you compare here?

Compare the three variable names plus their order. Your local file should contain your real values in place of these placeholders.

iot_sensor_pipeline:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: "{{ env_var('DBT_SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('DBT_SNOWFLAKE_USER') }}"
      password: "{{ env_var('DBT_ENV_SECRET_PASSWORD') }}"
      warehouse: IOT_WH
      database: IOT_DB
      schema: ANALYTICS
      threads: 4

What should this profile target?

The profile targets IOT_WH plus IOT_DB.ANALYTICS. Its credential references match the names in .env.

dbt-core==1.12.0
dbt-snowflake==1.12.0

Why keep these pins?

These exact package versions keep dbt Core aligned with its Snowflake adapter. The isolated dbt environment already contains both.

ALTER WAREHOUSE IF EXISTS IOT_WH SUSPEND;
DROP WAREHOUSE IF EXISTS IOT_WH;
DROP DATABASE IF EXISTS IOT_DB;

When is this file used?

This file removes the project warehouse plus database during cleanup. Keep it saved without executing it now.

CREATE WAREHOUSE IF NOT EXISTS IOT_WH
  WAREHOUSE_SIZE = XSMALL
  AUTO_SUSPEND = 300
  AUTO_RESUME = TRUE
  INITIALLY_SUSPENDED = TRUE;

CREATE DATABASE IF NOT EXISTS IOT_DB;
CREATE SCHEMA IF NOT EXISTS IOT_DB.ANALYTICS;

What should this setup reproduce?

These statements reproduce the warehouse plus target namespace used by the dbt profile. Their names and settings must match exactly.

Verify the dbt connection

The final check validates both the local configuration and the remote connection. Running it from the dbt folder also lets dbt Core load the local .env file.

  • Right-click the dbt folder in the Explorer.
  • Select Open in Integrated Terminal.

Before you run the check, do you expect both the local project configuration and the Snowflake connection to succeed?

  • Test the dbt configuration plus Snowflake connection by running this command:
../.venv-dbt/bin/dbt debug

What does this check cover?

This check validates the dbt project environment plus connection profile. It then attempts to authenticate with Snowflake using the values loaded from .env.

You should see successful project and connection checks. The connection details identify IOT_WH plus IOT_DB and ANALYTICS.

That connection is live. dbt can now send the project seed plus transformations to the intended Snowflake namespace.

Authentication check failed?

  • Confirm that you are still using the Snowflake trial account created in this step.
  • Check that the account identifier in .env does not include a snowflakecomputing.com hostname.
  • Confirm that the username plus password match the credentials used to sign in to Snowflake.

Still seeing a failed connection? Help me troubleshoot my dbt Snowflake connection.

Your warehouse plus dbt profile now point to the same Snowflake environment. Next, you will load the sensor readings and build the first hourly health table.

Build the First Health Model

Your Snowflake connection is ready. Now your IoT readings need a path from source rows to metrics you can inspect.

This first dbt pass trusts every source row. Running it establishes a baseline that lets the data reveal whether that assumption is safe.

In this step, get ready to:
  • Configure the dbt project and load the sensor seed into Snowflake.
  • Build a staging view and an hourly device-health mart.
  • Query the mart and inspect the first health metrics.
Configure the dbt project and load the seed

A dbt project file maps your model folders to their warehouse materializations. A seed turns a version-controlled CSV file into a Snowflake table that later models can reference.

  • Populate a new dbt/dbt_project.yml file in Visual Studio Code with the configuration below:
name: iot_sensor_pipeline
version: "1.0.0"
config-version: 2
profile: iot_sensor_pipeline
require-dbt-version: "==1.12.0"

model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]

models:
  iot_sensor_pipeline:
    staging:
      +materialized: view
    marts:
      +materialized: table

What does this configuration do?

  • The project name links this configuration to the existing iot_sensor_pipeline profile.
  • The path settings tell dbt where to find models, seeds, and tests.
  • Models inside staging become views.
  • Models inside marts become tables.
  • Save dbt/dbt_project.yml.
  • Confirm dbt_project.yml appears inside the existing dbt folder.

Project file in the wrong place?

  • Check that dbt_project.yml sits directly inside dbt.
  • Confirm the filename ends with .yml.
  • Compare the indentation under models with the reference above.

Still stuck? Help me check why dbt is not reading my project configuration.

  • Populate a new dbt/seeds/sensor_readings.csv file with the sensor data below:
reading_id,device_id,recorded_at,ingested_at,temperature_c,battery_pct
r001,sensor_a,2026-09-11 09:00:00,2026-09-11 09:00:05,22.4,91
r002,sensor_a,2026-09-11 09:15:00,2026-09-11 09:15:04,23.1,88
r003,sensor_a,2026-09-11 09:30:00,2026-09-11 09:30:05,31.2,19
r003,sensor_a,2026-09-11 09:30:00,2026-09-11 09:32:00,29.8,18
r004,sensor_b,2026-09-11 09:00:00,2026-09-11 09:00:03,19.8,76
r005,sensor_b,2026-09-11 09:15:00,2026-09-11 09:15:03,20.3,73
r006,sensor_b,2026-09-11 09:30:00,2026-09-11 09:30:03,20.1,71

What is in the seed?

  • Each row captures a device reading plus the time that reading reached the pipeline.
  • The records cover sensor_a and sensor_b during the same hour.
  • The two r003 records share one reading ID.
  • The later r003 record represents a corrected reading.
  • Save dbt/seeds/sensor_readings.csv.
  • Load the seed into Snowflake by running this command from the Visual Studio Code terminal used for the connection check:
../.venv-dbt/bin/dbt seed

You should see the seed complete successfully. Snowflake now contains the sensor_readings table in IOT_DB.ANALYTICS.

Seed did not load?

  • Confirm the terminal is still using ~/iot-sensor-pipeline/dbt as its current folder.
  • Check that the seed file is saved at dbt/seeds/sensor_readings.csv.
  • Confirm every data row has six comma-separated values.

Need another pair of eyes? Help me troubleshoot why my dbt seed is not loading into Snowflake.

Build the staging view and health mart

The staging layer converts the timestamp strings into Snowflake timestamps. The mart then groups those staged rows into hourly device-health metrics through a dbt dependency.

  • Populate a new dbt/models/staging/stg_sensor_readings.sql file with the staging model below:
WITH typed AS (
    SELECT
        reading_id,
        device_id,
        TO_TIMESTAMP_NTZ(recorded_at) AS recorded_at,
        TO_TIMESTAMP_NTZ(ingested_at) AS ingested_at,
        temperature_c,
        battery_pct
    FROM {{ ref('sensor_readings') }}
)
SELECT * FROM typed

What does the staging model do?

  • The typed common table expression gives the timestamp fields warehouse-ready data types.
  • The ref('sensor_readings') call creates a dependency on the seed table.
  • The final selection retains every typed source row in this first version.
  • Save dbt/models/staging/stg_sensor_readings.sql.
  • Build the staging view by running this command:
../.venv-dbt/bin/dbt run

You should see the model run complete successfully. Snowflake now contains the stg_sensor_readings staging view.

Staging model did not build?

  • Check that the staging file is saved inside dbt/models/staging.
  • Confirm both timestamp expressions use matching parentheses.
  • Verify that the earlier seed completed before running the model.

Still blocked? Help me debug why the staging model cannot build from the sensor seed.

  • Populate a new dbt/models/marts/device_health_hourly.sql file with the mart model below:
SELECT
    device_id,
    DATE_TRUNC('hour', recorded_at) AS reading_hour,
    COUNT(*) AS reading_count,
    ROUND(AVG(temperature_c), 2) AS avg_temperature_c,
    MIN(battery_pct) AS min_battery_pct,
    CASE
        WHEN MAX(temperature_c) >= 30 OR MIN(battery_pct) <= 20 THEN TRUE
        ELSE FALSE
    END AS needs_attention
FROM {{ ref('stg_sensor_readings') }}
GROUP BY
    device_id,
    DATE_TRUNC('hour', recorded_at)

How does the health mart work?

  • The ref('stg_sensor_readings') call places the mart after the staging view in dbt's dependency graph.
  • The grouping produces one row for each device within each reading hour.
  • The aggregations calculate the reading count, average temperature, and minimum battery percentage.
  • The project-defined attention flag becomes true at a temperature of at least 30 or a battery percentage of at most 20.
  • Save dbt/models/marts/device_health_hourly.sql.
  • Build the staging view and mart table by running this command:
../.venv-dbt/bin/dbt run

You should see both models complete successfully. Snowflake now contains the stg_sensor_readings view and the DEVICE_HEALTH_HOURLY table.

Mart model did not build?

  • Confirm the mart file is saved inside dbt/models/marts.
  • Check that the referenced staging model is spelled stg_sensor_readings.
  • Compare the selected fields with the grouping fields in the model above.

Need help with the dependency? Help me troubleshoot why the health mart is not building from the staging view.

Inspect the first health metrics

A warehouse model becomes useful when you inspect its output. This query exposes the health metrics in device order so the first-pass assumption becomes visible.

  • Populate a new sql/verify.sql file with the verification query below:
SELECT
  DEVICE_ID,
  READING_HOUR,
  READING_COUNT,
  AVG_TEMPERATURE_C,
  MIN_BATTERY_PCT,
  NEEDS_ATTENTION
FROM IOT_DB.ANALYTICS.DEVICE_HEALTH_HOURLY
ORDER BY DEVICE_ID, READING_HOUR;

What does the verification query show?

  • The query reads the final mart table from IOT_DB.ANALYTICS.
  • The selected columns expose each metric produced by the mart.
  • The ordering keeps each device's hourly result easy to compare.
  • Save sql/verify.sql.
  • Rebuild the complete first-pass pipeline by running this command from ~/iot-sensor-pipeline/dbt:
../.venv-dbt/bin/dbt seed && ../.venv-dbt/bin/dbt run

You should see the seed and both models complete successfully. The warehouse now holds a fresh copy of the first-pass pipeline.

Before you query the mart, do you think the count reflects physical rows or distinct reading IDs?

  • Return to the Snowflake web interface from earlier.
  • Open a SQL worksheet.
  • Copy the saved query from sql/verify.sql into the worksheet.
  • Run the query.

You will see sensor_a with READING_COUNT = 4. Its source rows contain only three logical reading_id values.

That mismatch is the intended result. The first staging model keeps both copies of r003 before the mart counts its rows.

Not seeing the inflated count?

  • Confirm the query reads from IOT_DB.ANALYTICS.DEVICE_HEALTH_HOURLY.
  • Check that both r003 rows remain in sensor_readings.csv.
  • Rerun the seed and models before executing the verification query again.

Still seeing a different result? Help me trace why sensor_a does not show a reading count of 4.

✔️ Awesome, I've got everything!

Your seed, staging view, mart table, and verification query are saved. The first pipeline run now exposes the inflated sensor_a metric.

ⓧ I'd like to double check the full code

Compare each new file with the complete saved versions below:

name: iot_sensor_pipeline
version: "1.0.0"
config-version: 2
profile: iot_sensor_pipeline
require-dbt-version: "==1.12.0"

model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]

models:
  iot_sensor_pipeline:
    staging:
      +materialized: view
    marts:
      +materialized: table
reading_id,device_id,recorded_at,ingested_at,temperature_c,battery_pct
r001,sensor_a,2026-09-11 09:00:00,2026-09-11 09:00:05,22.4,91
r002,sensor_a,2026-09-11 09:15:00,2026-09-11 09:15:04,23.1,88
r003,sensor_a,2026-09-11 09:30:00,2026-09-11 09:30:05,31.2,19
r003,sensor_a,2026-09-11 09:30:00,2026-09-11 09:32:00,29.8,18
r004,sensor_b,2026-09-11 09:00:00,2026-09-11 09:00:03,19.8,76
r005,sensor_b,2026-09-11 09:15:00,2026-09-11 09:15:03,20.3,73
r006,sensor_b,2026-09-11 09:30:00,2026-09-11 09:30:03,20.1,71
WITH typed AS (
    SELECT
        reading_id,
        device_id,
        TO_TIMESTAMP_NTZ(recorded_at) AS recorded_at,
        TO_TIMESTAMP_NTZ(ingested_at) AS ingested_at,
        temperature_c,
        battery_pct
    FROM {{ ref('sensor_readings') }}
)
SELECT * FROM typed
SELECT
    device_id,
    DATE_TRUNC('hour', recorded_at) AS reading_hour,
    COUNT(*) AS reading_count,
    ROUND(AVG(temperature_c), 2) AS avg_temperature_c,
    MIN(battery_pct) AS min_battery_pct,
    CASE
        WHEN MAX(temperature_c) >= 30 OR MIN(battery_pct) <= 20 THEN TRUE
        ELSE FALSE
    END AS needs_attention
FROM {{ ref('stg_sensor_readings') }}
GROUP BY
    device_id,
    DATE_TRUNC('hour', recorded_at)
SELECT
  DEVICE_ID,
  READING_HOUR,
  READING_COUNT,
  AVG_TEMPERATURE_C,
  MIN_BATTERY_PCT,
  NEEDS_ATTENTION
FROM IOT_DB.ANALYTICS.DEVICE_HEALTH_HOURLY
ORDER BY DEVICE_ID, READING_HOUR;

Your first health model is running from seed to mart. Next, you will turn the hidden duplicate into a repeatable dbt test failure.

Expose Duplicate Readings

Your first dbt run produced an hourly health mart. However, sensor_a reports four readings even though it has three logical reading_id values.

A trustworthy pipeline needs an automated check that catches this hidden duplication. In this step, you will add model tests that turn the suspicious count into a repeatable failure.

In this step, get ready to:
  • Define generic quality checks for the staging view and hourly mart.
  • Run the tests to expose the duplicated sensor reading.
  • Connect the failed uniqueness check to the inflated health metric.
Define the model tests

Generic data tests attach reusable assertions to model columns in schema.yml. These checks protect the reading identity plus the fields required by the mart.

  • Return to the iot-sensor-pipeline workspace in Visual Studio Code.
  • Select the dbt/models folder in the file sidebar.
  • Click the file creation icon beside the folder.
  • Name the file schema.yml.
  • Define the model tests by pasting this content into schema.yml:
version: 2

models:
  - name: stg_sensor_readings
    columns:
      - name: reading_id
        data_tests:
          - unique
          - not_null
      - name: device_id
        data_tests:
          - not_null
      - name: recorded_at
        data_tests:
          - not_null
  - name: device_health_hourly
    columns:
      - name: device_id
        data_tests:
          - not_null
      - name: reading_hour
        data_tests:
          - not_null

What Do These Tests Protect?

  • The version: 2 entry sets the schema version for these model definitions.
  • The reading_id column gets one unique check. It also gets one not_null check.
  • The remaining not_null checks cover fields required by the staging view and mart.
  • Each assertion runs independently. This makes the specific quality problem visible in the test results.
  • Save dbt/models/schema.yml.
  • Confirm schema.yml appears inside dbt/models in the file sidebar.

The file now registers six assertions across the two models.

Tests Not Appearing?

  • Check that the file path is exactly dbt/models/schema.yml.
  • Compare the indentation beneath models: with the reference below.

Still stuck? Help me check why dbt is not discovering the tests in my schema.yml file.

✔️ Awesome, I've got everything!

Your schema.yml file now defines the complete set of tests for this step.

ⓧ I'd like to double check the full code

Here is the complete dbt/models/schema.yml file. Compare it with yours line by line.

version: 2

models:
  - name: stg_sensor_readings
    columns:
      - name: reading_id
        data_tests:
          - unique
          - not_null
      - name: device_id
        data_tests:
          - not_null
      - name: recorded_at
        data_tests:
          - not_null
  - name: device_health_hourly
    columns:
      - name: device_id
        data_tests:
          - not_null
      - name: reading_hour
        data_tests:
          - not_null
Run the failing test

The dbt test command evaluates every assertion against the current warehouse relations. It reports each check independently.

  • Switch back to the integrated terminal from earlier.
  • Confirm the terminal is still working inside ~/iot-sensor-pipeline/dbt.

Before you run the tests, which assumption do you think the repeated r003 rows will challenge?

  • Run all configured tests by running this command:
../.venv-dbt/bin/dbt test

This Failure Is Expected

The command executes all six assertions against the current models. The unique check returns the repeated r003 value.

dbt marks that test as failed. Each not_null check passes because the required fields contain values.

That failed exit status is the pipeline guard doing its job.

  • Locate the failed unique test for stg_sensor_readings.reading_id in the terminal output.
  • Confirm each configured not_null test reports a pass.

Strong work. The duplicate now appears as a visible pipeline failure.

The tests only inspect the current relations. sensor_a therefore remains at READING_COUNT = 4 until the staging logic changes.

Unexpected Test Result?

  • If no tests run, confirm the file is saved at dbt/models/schema.yml.
  • If the uniqueness test passes, compare the end of dbt/models/staging/stg_sensor_readings.sql with SELECT * FROM typed.
  • If the command cannot reach the warehouse, confirm the Snowflake connection from the earlier setup still works.

Need another pair of eyes? Help me diagnose my unexpected dbt test result.

Your quality check now catches the duplicate consistently. Next, you will keep the latest correction before Apache Airflow runs the workflow from seed through test.

Deduplicate and Orchestrate the Pipeline

The failing dbt uniqueness test has traced the inflated metric to the duplicate r003 reading. That duplicate still makes sensor_a report four readings in Snowflake.

You will keep the latest copy of each reading with a deterministic window function. You will also turn the separate dbt commands into an observable workflow with Apache Airflow.

In this step, get ready to:
  • Deduplicate the staging model with deterministic row ranking.
  • Build an Airflow DAG for the dbt seed, model, and test commands.
  • Trigger the DAG and inspect its successful task logs.
Deduplicate the staging model

Each reading_id represents one logical sensor event. Ranking rows by ingested_at lets the staging model keep the latest correction for that event.

  • Return to dbt/models/staging/stg_sensor_readings.sql in Visual Studio Code.
  • Replace the entire file with this completed staging model:
WITH typed AS (
    SELECT
        reading_id,
        device_id,
        TO_TIMESTAMP_NTZ(recorded_at) AS recorded_at,
        TO_TIMESTAMP_NTZ(ingested_at) AS ingested_at,
        temperature_c,
        battery_pct
    FROM {{ ref('sensor_readings') }}
),
ranked AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY reading_id
            ORDER BY ingested_at DESC
        ) AS row_num
    FROM typed
)
SELECT
    reading_id,
    device_id,
    recorded_at,
    ingested_at,
    temperature_c,
    battery_pct
FROM ranked
WHERE row_num = 1

What does this model do?

  • The typed CTE converts the two timestamp fields before they are compared.
  • The ranked CTE assigns a row number within each reading_id group.
  • The descending ingestion order gives the latest copy a row_num value of 1.
  • The final filter keeps that latest copy for the downstream health mart.
  • Save dbt/models/staging/stg_sensor_readings.sql.

Before you run the corrected model, predict whether the uniqueness test will still find two copies of r003.

  • Switch back to the VS Code terminal already positioned in ~/iot-sensor-pipeline/dbt.
  • Rebuild the models and rerun every configured test with this command:
../.venv-dbt/bin/dbt run && ../.venv-dbt/bin/dbt test

What does this command prove?

The first command rebuilds the staging view and hourly mart with the deduplication logic. The second command checks the rebuilt data against every test in dbt/models/schema.yml.

The chained commands stop if the model run fails. A completed sequence proves the transformation runs before its quality checks.

You should see the model run complete successfully. Every configured dbt test should now pass, including the uniqueness test for reading_id.

  • Return to the Snowflake SQL worksheet from earlier.
  • Replace the worksheet contents with this verification query:
SELECT
  DEVICE_ID,
  READING_HOUR,
  READING_COUNT,
  AVG_TEMPERATURE_C,
  MIN_BATTERY_PCT,
  NEEDS_ATTENTION
FROM IOT_DB.ANALYTICS.DEVICE_HEALTH_HOURLY
ORDER BY DEVICE_ID, READING_HOUR;

What does this query check?

The query reads the final hourly mart in IOT_DB.ANALYTICS. Its ordered output makes the corrected sensor_a metric easy to compare with the earlier result.

  • Execute the query in the Snowflake worksheet.

You should now see sensor_a with a READING_COUNT value of 3. That is the visible proof that the duplicate correction no longer inflates the mart.

Still seeing four readings?

  • Confirm that ORDER BY ingested_at DESC appears inside the row-number window.
  • Confirm that the final query ends with WHERE row_num = 1.
  • Rerun the dbt model command before executing sql/verify.sql again.

If the count remains inflated, ask for help checking the staging model.

Build the Airflow DAG

An Airflow DAG records each dbt command as a task. Explicit dependencies ensure the seed loads before the models run.

Why keep dbt in a separate environment?

Airflow and dbt each have large dependency graphs. The DAG calls the dbt executable inside .venv-dbt while Airflow continues running from .venv-airflow.

Each task also uses the dbt folder as its working directory. That lets dbt load the existing local .env file.

  • Return to the Explorer sidebar in Visual Studio Code.
  • Use the new-file control to create airflow/dags/iot_sensor_health.py inside ~/iot-sensor-pipeline.

You should see the new file nested under the airflow/dags folders in the Explorer sidebar.

  • Add the imports, project paths, and dbt command helper to airflow/dags/iot_sensor_health.py with this code:
from datetime import datetime
from pathlib import Path

from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

PROJECT_ROOT = Path.home() / "iot-sensor-pipeline"
DBT_PROJECT_DIR = PROJECT_ROOT / "dbt"
DBT_EXECUTABLE = PROJECT_ROOT / ".venv-dbt" / "bin" / "dbt"


def dbt_command(subcommand: str) -> str:
    return f'set -e; "{DBT_EXECUTABLE}" {subcommand}'

How does the helper work?

  • The path constants locate the project folder and its isolated dbt executable.
  • The dbt_command() helper turns a dbt subcommand into the shell command used by each task.
  • The set -e setting makes a failed dbt command fail its Airflow task.
  • Save airflow/dags/iot_sensor_health.py.
  • Confirm that the file remains listed under airflow/dags in the Explorer sidebar.

Seeing unresolved Airflow imports?

Visual Studio Code may be using a different Python interpreter for editor analysis. The DAG still runs through the isolated Airflow executable in the next substep.

If the file has a Python syntax warning, ask for help checking the imports and path definitions.

  • Place the DAG definition directly below dbt_command() with this code:
with DAG(
    dag_id="iot_sensor_health",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
    tags=["dbt", "snowflake", "iot"],
) as dag:
    seed_sensor_readings = BashOperator(
        task_id="seed_sensor_readings",
        bash_command=dbt_command("seed"),
        cwd=str(DBT_PROJECT_DIR),
    )

    run_models = BashOperator(
        task_id="run_models",
        bash_command=dbt_command("run"),
        cwd=str(DBT_PROJECT_DIR),
    )

    test_models = BashOperator(
        task_id="test_models",
        bash_command=dbt_command("test"),
        cwd=str(DBT_PROJECT_DIR),
    )

    seed_sensor_readings >> run_models >> test_models

What does the DAG define?

  • The iot_sensor_health DAG runs only when you trigger it.
  • The seed_sensor_readings task loads the version-controlled sensor CSV.
  • The run_models task rebuilds the staging view and hourly mart.
  • The test_models task runs the configured dbt data tests.
  • The final dependency expression forces those three tasks to run in sequence.
  • Save airflow/dags/iot_sensor_health.py.
  • Confirm that the unsaved indicator disappears from the file tab.

DAG file showing a syntax warning?

  • Confirm that all three task definitions remain indented inside the with DAG block.
  • Confirm that the dependency line uses the exact task variable names from the definitions above.

If the warning remains, ask for help comparing the complete DAG.

✔️ Awesome, I've got everything!

Great. Double check that airflow/dags/iot_sensor_health.py is saved inside the existing project.

ⓧ I'd like to double check the full code

Compare your complete airflow/dags/iot_sensor_health.py file with this version line by line.

from datetime import datetime
from pathlib import Path

from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

PROJECT_ROOT = Path.home() / "iot-sensor-pipeline"
DBT_PROJECT_DIR = PROJECT_ROOT / "dbt"
DBT_EXECUTABLE = PROJECT_ROOT / ".venv-dbt" / "bin" / "dbt"


def dbt_command(subcommand: str) -> str:
    return f'set -e; "{DBT_EXECUTABLE}" {subcommand}'


with DAG(
    dag_id="iot_sensor_health",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
    tags=["dbt", "snowflake", "iot"],
) as dag:
    seed_sensor_readings = BashOperator(
        task_id="seed_sensor_readings",
        bash_command=dbt_command("seed"),
        cwd=str(DBT_PROJECT_DIR),
    )

    run_models = BashOperator(
        task_id="run_models",
        bash_command=dbt_command("run"),
        cwd=str(DBT_PROJECT_DIR),
    )

    test_models = BashOperator(
        task_id="test_models",
        bash_command=dbt_command("test"),
        cwd=str(DBT_PROJECT_DIR),
    )

    seed_sensor_readings >> run_models >> test_models

How the complete file fits together

The first section locates dbt and builds reusable commands. The second section defines the three Airflow tasks in their required execution order.

Run the orchestrated pipeline

Airflow standalone initializes the local metadata database and starts the services needed for the web interface. The AIRFLOW_HOME value keeps its DAGs and logs inside ~/iot-sensor-pipeline/airflow.

The first start can take a few minutes while Airflow prepares its local database. Leave this terminal running after the credentials appear.

  • Start Airflow from the VS Code terminal from earlier by running these commands:
cd "$HOME/iot-sensor-pipeline"
AIRFLOW_HOME="$HOME/iot-sensor-pipeline/airflow" ./.venv-airflow/bin/airflow standalone

What does this command start?

The first line moves the terminal to the existing project folder. The second line starts Airflow with the project-specific home directory.

Standalone mode initializes the database and creates a local user. It also serves the Airflow interface at localhost:8080.

You should see local admin credentials printed in the terminal. Keep those credentials private because they unlock your local Airflow interface.

Airflow did not finish starting?

  • Confirm that you launched the command from ~/iot-sensor-pipeline.
  • Confirm that the standalone command uses the executable inside .venv-airflow.
  • Leave the terminal running because stopping it also stops the local Airflow services.

If startup still fails, ask for help diagnosing the local Airflow process.

  • Create a second integrated terminal with the plus icon in the VS Code terminal panel.

Before you trigger the workflow, predict whether all three tasks start together or follow the dependency chain.

  • List the DAGs, unpause the sensor pipeline, and trigger a run by executing these commands:
cd "$HOME/iot-sensor-pipeline"
AIRFLOW_HOME="$HOME/iot-sensor-pipeline/airflow" ./.venv-airflow/bin/airflow dags list
AIRFLOW_HOME="$HOME/iot-sensor-pipeline/airflow" ./.venv-airflow/bin/airflow dags unpause iot_sensor_health
AIRFLOW_HOME="$HOME/iot-sensor-pipeline/airflow" ./.venv-airflow/bin/airflow dags trigger iot_sensor_health

What do these commands do?

  • The list command confirms that Airflow discovered iot_sensor_health.
  • The unpause command allows the DAG to run.
  • The trigger command creates a new manual DAG run.

The credentials only unlock the local interface on your Mac. Keep the terminal containing them out of any screenshots.

  • Open http://localhost:8080 in your browser.
  • Sign in with the admin credentials printed by the standalone process.
  • Select iot_sensor_health from the DAG list.
  • Open the newest run to inspect its task progression.
  • Inspect the logs for each completed task.

You should see seed_sensor_readings finish before run_models starts. You should then see test_models finish successfully.

That completes the full workflow. Your corrected sensor data now moves through seed, transformation, and testing as one observable pipeline.

DAG missing or task failed?

  • Confirm that iot_sensor_health.py is inside airflow/dags.
  • Confirm that the standalone process uses ~/iot-sensor-pipeline/airflow as its Airflow home.
  • Inspect the failed task log to identify whether the seed, model, or test command stopped.

If the run remains unsuccessful, ask for help tracing the failed Airflow task.

Secret mission

Add a Battery-Range Guard

A duplicate-free dataset can still contain impossible sensor readings. Add a custom battery-range test that blocks an invalid percentage in Airflow before accepting the corrected value.

Clean Up Your Resources

Clean Up Your Resources

Your Snowflake trial consumes credits while IOT_WH runs. Local files do not create ongoing charges.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Cost warning

Snowflake currently gives trial accounts $400 in credits for up to 30 days. The trial can end sooner when the balance runs out.

Adding a payment method converts the account to paid usage. The Pause and Delete options below stop warehouse compute.

Resources you used:

  • Your IOT_WH Snowflake warehouse.
  • Your IOT_DB database containing every object in IOT_DB.ANALYTICS.
  • Your local Apache Airflow process.
  • Your local ~/iot-sensor-pipeline folder containing both Python virtual environments plus all dbt and Airflow files.

Keep everything running

No action needed. Choose this if you plan to keep working with the pipeline.

  • Leave IOT_WH configured with its 300-second auto-suspend setting.
  • Keep IOT_DB available for future dbt runs.
  • Leave Airflow running only while you are reviewing DAG runs.
  • Keep the ~/iot-sensor-pipeline folder on your Mac.

Pause - I'll come back to this later

Pausing stops the local Airflow process plus Snowflake compute. Your database objects and local files stay ready for next time.

  • Switch back to the terminal running Airflow.
  • Press Control+C to stop Airflow.
  • Return to the Snowflake SQL worksheet from earlier.
  • Suspend IOT_WH by running this statement:
ALTER WAREHOUSE IF EXISTS IOT_WH SUSPEND;

What Does This Statement Do?

  • SUSPEND stops the warehouse compute after active statements finish.
  • IOT_DB stays available with its schema plus transformed data.
  • Your local project files remain unchanged.
  • Confirm the worksheet reports that the warehouse suspension completed.

Warehouse Would Not Suspend?

  • Check that your current Snowflake role can manage IOT_WH.
  • Help me diagnose the warehouse suspension issue.

Your project is safely paused. The warehouse now uses no compute until a later query resumes it.

Delete - I don't want to use this again

Remove the project resources when you want a complete reset. Warehouse deletion cannot be undone.

Your installed development tools stay available. Your Snowflake account also stays open.

  • Switch back to the terminal running Airflow.
  • Press Control+C to stop Airflow.
  • Return to the Snowflake SQL worksheet from earlier.
  • Remove the project warehouse plus database by running sql/cleanup.sql:
ALTER WAREHOUSE IF EXISTS IOT_WH SUSPEND;
DROP WAREHOUSE IF EXISTS IOT_WH;
DROP DATABASE IF EXISTS IOT_DB;

What Does This Cleanup Remove?

  • SUSPEND stops warehouse compute after active statements finish.
  • DROP WAREHOUSE removes IOT_WH.
  • DROP DATABASE removes IOT_DB with every object in IOT_DB.ANALYTICS.
  • Confirm the worksheet reports that all three statements completed.

Cleanup Did Not Complete?

  • Check that your current Snowflake role can manage the project warehouse plus database.
  • Rerun the full cleanup script after correcting the role permissions.
  • Help me diagnose the Snowflake cleanup failure.
  • Delete the ~/iot-sensor-pipeline folder in Finder.
  • Confirm the folder no longer appears at that path.

Your project resources are cleared. Your Mac retains the development tools for another project.

Nice Work!

Nice Work!

You did it! You built a rerunnable IoT sensor health pipeline that transforms warehouse data, catches duplicate readings, and exposes each workflow stage in Airflow.

You've learned how to:

  • Built a layered dbt pipeline that loads version-controlled sensor readings into Snowflake. The pipeline transforms them into hourly device-health metrics.
  • Exposed an inflated sensor metric with a uniqueness test. You then applied deterministic latest-record deduplication to restore the correct reading count.
  • Orchestrated seed, model, and test tasks through an Airflow DAG. You also inspected task states and logs to confirm successful failure propagation.
  • Secret Mission: Added a custom battery-range test that blocks invalid sensor data. You proved the guard works by comparing a failed Airflow run with a successful corrected run.

Ready to quiz yourself?