From 64c0adfab73530e6909ed84b74b100bcfd7bdfe7 Mon Sep 17 00:00:00 2001 From: Priyanka-Jammu Date: Tue, 17 Mar 2026 22:47:03 -0400 Subject: [PATCH 1/2] Add bronze-to-silver Snowflake pipeline template --- templates/bronze-silver-snowflake/.bruin.yml | 14 ++ templates/bronze-silver-snowflake/README.md | 70 ++++++++++ .../assets/bronze_raw_data.asset.yml | 52 ++++++++ .../assets/silver_aggregated.sql | 122 ++++++++++++++++++ .../bronze-silver-snowflake/pipeline.yml | 8 ++ 5 files changed, 266 insertions(+) create mode 100644 templates/bronze-silver-snowflake/.bruin.yml create mode 100644 templates/bronze-silver-snowflake/README.md create mode 100644 templates/bronze-silver-snowflake/assets/bronze_raw_data.asset.yml create mode 100644 templates/bronze-silver-snowflake/assets/silver_aggregated.sql create mode 100644 templates/bronze-silver-snowflake/pipeline.yml diff --git a/templates/bronze-silver-snowflake/.bruin.yml b/templates/bronze-silver-snowflake/.bruin.yml new file mode 100644 index 0000000000..463a41f2c9 --- /dev/null +++ b/templates/bronze-silver-snowflake/.bruin.yml @@ -0,0 +1,14 @@ +default_environment: default +environments: + default: + connections: + frankfurter: + - name: "frankfurter-default" + snowflake: + - name: "snowflake-default" + account: "ACCOUNT-NAME" + user: "USERNAME" + password: "PASSWORD" + database: "SNOWFLAKE_DATABASE" + warehouse: "COMPUTE_WAREHOUSE" + schema: "PUBLIC" diff --git a/templates/bronze-silver-snowflake/README.md b/templates/bronze-silver-snowflake/README.md new file mode 100644 index 0000000000..0e4480cdf1 --- /dev/null +++ b/templates/bronze-silver-snowflake/README.md @@ -0,0 +1,70 @@ +# Bronze to Silver Snowflake Pipeline + +## Overview + +This template demonstrates a **bronze-to-silver data pipeline** using Snowflake as the destination. + +It ingests exchange rate data from the Frankfurter API into a **bronze layer** (raw data) and performs transformations to generate aggregated insights in a **silver layer**. + +The pipeline showcases a complete **ingestion → transformation → validation workflow** using Bruin. + +--- + +## Architecture + +Frankfurter API +↓ +Bronze Layer (Raw Ingestion - Snowflake) +↓ +Silver Layer (Aggregations & Metrics - Snowflake) + +--- + +## Features + +- Ingestion from a public API (no credentials required) +- Bronze layer for raw data storage +- Silver layer with: + - Latest exchange rates + - 7-day and 30-day averages + - Observation counts +- Data quality checks: + - Not null checks + - Positive value checks + - Freshness validation + - Uniqueness constraints +- Snowflake-native SQL transformations + +--- + +## Project Structure +bronze-silver-snowflake/ +│ +├── pipeline.yml +├── .bruin.yml +├── README.md +└── assets/ + ├── bronze_raw_data.asset.yml + └── silver_aggregated.sql + + +--- + +## Snowflake Setup + +Before running the pipeline, configure your Snowflake connection in `.bruin.yml`: + +```bruin.yaml +snowflake: + - name: "snowflake-default" + account: "YOUR_ACCOUNT" + username: "YOUR_USERNAME" + password: "YOUR_PASSWORD" + database: "YOUR_DATABASE" + warehouse: "COMPUTE_WH" + schema: "PUBLIC" +``` +Replace placeholders with your Snowflake credentials for local testing. +⚠️ Do not commit real credentials to GitHub. + +--- \ No newline at end of file diff --git a/templates/bronze-silver-snowflake/assets/bronze_raw_data.asset.yml b/templates/bronze-silver-snowflake/assets/bronze_raw_data.asset.yml new file mode 100644 index 0000000000..2cff81218a --- /dev/null +++ b/templates/bronze-silver-snowflake/assets/bronze_raw_data.asset.yml @@ -0,0 +1,52 @@ +name: public.bronze_exchange_rates +type: ingestr + +description: > + Ingests daily FX rates from the Frankfurter API into Snowflake without transformations. + The dataset makes the raw bronze layer available for downstream silver modeling. + +columns: + - name: date + type: date + description: "Exchange rate date reported by the Frankfurter API." + checks: + - name: not_null + - name: base_currency + type: varchar + description: "Currency that the quote currency is priced against." + checks: + - name: not_null + - name: currency_code + type: varchar + description: "Quote currency code returned by Frankfurter." + checks: + - name: not_null + - name: rate + type: float + description: "Reported daily conversion rate between the base and quote currency." + checks: + - name: not_null + - name: positive + +custom_checks: + - name: unique currency per day + value: 0 + query: | + SELECT COUNT(*) + FROM ( + SELECT date, currency_code, base_currency + FROM public.bronze_exchange_rates + GROUP BY 1, 2, 3 + HAVING COUNT(*) > 1 + ) duplicates + - name: data freshness (3 day max lag) + value: 0 + query: | + SELECT CASE WHEN MAX(TO_DATE(date)) < DATEADD(day, -3, CURRENT_DATE()) THEN 1 ELSE 0 END AS stale_flag + FROM public.bronze_exchange_rates + +parameters: + source_connection: frankfurter-default + source_table: exchange_rates + destination: snowflake + destination_table: bronze_exchange_rates diff --git a/templates/bronze-silver-snowflake/assets/silver_aggregated.sql b/templates/bronze-silver-snowflake/assets/silver_aggregated.sql new file mode 100644 index 0000000000..df31676b0c --- /dev/null +++ b/templates/bronze-silver-snowflake/assets/silver_aggregated.sql @@ -0,0 +1,122 @@ +/* @bruin + +name: public.silver_exchange_rate_summary +type: sf.sql + +materialization: + type: table + +description: > + Aggregates the bronze Frankfurter exchange rate feed into a silver summary table with + fresh rates, rolling averages, and observation counts for downstream analytics. + +depends: + - public.bronze_exchange_rates + +columns: + - name: currency_code + type: varchar + description: "Quoted currency code." + checks: + - name: not_null + - name: base_currency + type: varchar + description: "Base currency that the quote is relative to." + checks: + - name: not_null + - name: latest_date + type: date + description: "Most recent exchange rate date captured in the bronze layer." + checks: + - name: not_null + - name: latest_rate + type: float + description: "Latest available exchange rate between the currency pair." + checks: + - name: not_null + - name: positive + - name: avg_rate_7d + type: float + description: "Seven day rolling average of the conversion rate." + checks: + - name: positive + - name: avg_rate_30d + type: float + description: "Thirty day rolling average of the conversion rate." + checks: + - name: positive + - name: observations_7d + type: integer + description: "Number of daily observations in the last seven days." + checks: + - name: non_negative + - name: observations_30d + type: integer + description: "Number of daily observations in the last thirty days." + checks: + - name: positive + +custom_checks: + - name: silver table populated + value: 0 + query: | + SELECT CASE WHEN COUNT(*) = 0 THEN 1 ELSE 0 END + FROM public.silver_exchange_rate_summary + - name: fresh latest date (<= 3 day lag) + value: 0 + query: | + SELECT COUNT(*) + FROM public.silver_exchange_rate_summary + WHERE latest_date < DATEADD(day, -3, CURRENT_DATE()) + +@bruin */ + +WITH bronze_data AS ( + SELECT + TO_DATE(date) AS rate_date, + base_currency, + currency_code, + CAST(rate AS FLOAT) AS rate + FROM public.bronze_exchange_rates +), +recent AS ( + SELECT + currency_code, + base_currency, + AVG(CASE WHEN rate_date >= DATEADD(day, -7, CURRENT_DATE()) THEN rate END) AS avg_rate_7d, + AVG(CASE WHEN rate_date >= DATEADD(day, -30, CURRENT_DATE()) THEN rate END) AS avg_rate_30d, + COUNT(CASE WHEN rate_date >= DATEADD(day, -7, CURRENT_DATE()) THEN 1 END) AS observations_7d, + COUNT(CASE WHEN rate_date >= DATEADD(day, -30, CURRENT_DATE()) THEN 1 END) AS observations_30d + FROM bronze_data + WHERE rate_date >= DATEADD(day, -30, CURRENT_DATE()) + GROUP BY currency_code, base_currency +), + +latest AS ( + SELECT + currency_code, + base_currency, + rate_date AS latest_date, + rate AS latest_rate + FROM bronze_data + QUALIFY ROW_NUMBER() OVER ( + PARTITION BY currency_code, base_currency + ORDER BY rate_date DESC + ) = 1 +) + +SELECT + r.currency_code, + r.base_currency, + l.latest_date, + l.latest_rate, + COALESCE(r.avg_rate_7d, r.avg_rate_30d) AS avg_rate_7d, + r.avg_rate_30d, + r.observations_7d, + r.observations_30d +FROM recent r +JOIN latest l + ON r.currency_code = l.currency_code + AND r.base_currency = l.base_currency +ORDER BY r.currency_code, r.base_currency; + diff --git a/templates/bronze-silver-snowflake/pipeline.yml b/templates/bronze-silver-snowflake/pipeline.yml new file mode 100644 index 0000000000..401fa614c6 --- /dev/null +++ b/templates/bronze-silver-snowflake/pipeline.yml @@ -0,0 +1,8 @@ +name: bronze-silver-snowflake +schedule: daily +start_date: "2024-01-01" +catchup: false + +default_connections: + frankfurter: "frankfurter-default" + snowflake: "snowflake-default" From f864fcfed09877f662ec2a7114d010b2a8f79779 Mon Sep 17 00:00:00 2001 From: Priyanka-Jammu Date: Wed, 18 Mar 2026 15:00:05 -0400 Subject: [PATCH 2/2] Fix README field and remove avg_rate_7d fallback --- templates/bronze-silver-snowflake/README.md | 4 ++-- .../bronze-silver-snowflake/assets/silver_aggregated.sql | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/templates/bronze-silver-snowflake/README.md b/templates/bronze-silver-snowflake/README.md index 0e4480cdf1..9d25966ea6 100644 --- a/templates/bronze-silver-snowflake/README.md +++ b/templates/bronze-silver-snowflake/README.md @@ -58,10 +58,10 @@ Before running the pipeline, configure your Snowflake connection in `.bruin.yml` snowflake: - name: "snowflake-default" account: "YOUR_ACCOUNT" - username: "YOUR_USERNAME" + user: "YOUR_USERNAME" password: "YOUR_PASSWORD" database: "YOUR_DATABASE" - warehouse: "COMPUTE_WH" + warehouse: "COMPUTE_WAREHOUSE" schema: "PUBLIC" ``` Replace placeholders with your Snowflake credentials for local testing. diff --git a/templates/bronze-silver-snowflake/assets/silver_aggregated.sql b/templates/bronze-silver-snowflake/assets/silver_aggregated.sql index df31676b0c..8e52371085 100644 --- a/templates/bronze-silver-snowflake/assets/silver_aggregated.sql +++ b/templates/bronze-silver-snowflake/assets/silver_aggregated.sql @@ -110,7 +110,7 @@ SELECT r.base_currency, l.latest_date, l.latest_rate, - COALESCE(r.avg_rate_7d, r.avg_rate_30d) AS avg_rate_7d, + r.avg_rate_7d, r.avg_rate_30d, r.observations_7d, r.observations_30d