ELT Pipeline with dbt and BigQuery

Load banking data into BigQuery and transform it with dbt models and tests.

Introduction

30 Second Summary

Every bank processes thousands of transactions daily. Before analysts can spot fraud or measure growth, someone has to clean and organize all that raw data into something useful.

In this project, you will build a complete ELT pipeline that loads banking transaction data into Google BigQuery and transforms it into clean, tested analytical models using dbt.

What You'll Build

A production-style ELT pipeline that extracts CSV data, loads it into BigQuery with Python, and transforms it into analytics-ready models using dbt.

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

  • A Python script that loads raw banking CSV data into BigQuery as raw tables.
  • A dbt project with staging models, a mart model, and automated data quality tests running against BigQuery.
  • Auto-generated documentation with a visual lineage DAG showing how data flows through your entire pipeline.
  • Secret Mission: Add an incremental model that processes only new transactions since the last run.

Are there any prerequisites?

You should be comfortable with basic SQL (SELECT, JOIN, GROUP BY) and basic Python. You also need a Google Cloud Platform account with BigQuery access. GCP offers free credits for new accounts, and BigQuery's free tier covers everything in this project.

Set Up Your Environment

Before you can build a data pipeline, you need the tools that connect your local machine to Google Cloud.

In this step, you will install the Google Cloud SDK, authenticate with your GCP account, set up a Python virtual environment with the required packages, and create two BigQuery datasets that will hold your data.

In this step, get ready to:
  • Install and configure the Google Cloud SDK.
  • Create a Python virtual environment and install dbt-bigquery.
  • Create two BigQuery datasets for your pipeline.
Install Google Cloud SDK and authenticate

The Google Cloud SDK gives you command-line tools to interact with GCP services. It includes gcloud for managing projects and authentication, and bq for working with BigQuery directly from your terminal.

  • Check if gcloud is already installed by running this command:
gcloud --version

✔️ I see a version number

The Google Cloud SDK is already installed. You can skip ahead to initialization below.

ⓧ Command not found

You need to install the Google Cloud SDK. Follow the official installer for your operating system.

  • Visit the official install page: cloud.google.com/sdk/docs/install-sdk
  • On macOS, download the archive, extract it, and run ./google-cloud-sdk/install.sh from the extracted folder.
  • On Windows, download and launch the installer. It is signed by Google LLC.
  • After the installer finishes, restart your terminal so the gcloud command is available on your PATH.

What does the installer include?

The Google Cloud SDK bundles gcloud, gsutil, and bq command-line tools. You will use gcloud for authentication and bq for creating datasets later in this step.

  • Confirm installation by running gcloud --version again. You should see a version number in the output.

Still seeing command not found?

Make sure you restarted your terminal after installation. On macOS, if you used the archive method, ensure you ran the install.sh script to add gcloud to your PATH.

Help me troubleshoot my gcloud CLI installation

  • Initialize the Google Cloud SDK by running this command:
gcloud init

What happens during gcloud init?

This command opens a browser window for you to sign in with your Google account. After signing in, it asks you to select a GCP project.

Choose the project where you want your BigQuery datasets to live. If you only have one project, it selects it automatically.

  • When prompted to log in, press Y and sign in with the Google account tied to your GCP project.
  • When prompted to select a project, choose the project you want to use for this pipeline.
  • When asked about a default Compute Engine region, you can press Enter to skip (not needed for this project).

Now set up application-default credentials. These allow both Python and dbt to authenticate with BigQuery without managing key files.

  • Generate application-default credentials by running this command:
gcloud auth application-default login --scopes=https://www.googleapis.com/auth/bigquery,https://www.googleapis.com/auth/drive.readonly,https://www.googleapis.com/auth/iam.test,https://www.googleapis.com/auth/cloud-platform

What does this command do?

This creates a local credentials file that any application running on your machine (like your Python script or dbt) can use to authenticate with BigQuery. The scopes flag grants access to BigQuery, Cloud Platform, and Drive (needed for dbt's full functionality).

  • A browser window opens. Sign in and click Allow to grant access.
  • Confirm your credentials are active by running:
gcloud auth list

You should see your Google account listed with an asterisk (*) next to it, indicating it is the active account.

Not seeing your account?

If gcloud auth list shows no accounts, re-run gcloud init and make sure you complete the browser sign-in step.

Help me fix my gcloud authentication

Create a Python virtual environment and install dependencies

A virtual environment keeps your project's packages isolated from the rest of your system. This prevents version conflicts between different Python projects.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir elt-pipeline
cd elt-pipeline
  • Create a Python virtual environment by running this command:
python3 -m venv dbt-env
  • Activate the virtual environment by running this command:

macOS

source dbt-env/bin/activate

Windows

dbt-env\Scripts\activate

You should see (dbt-env) at the start of your terminal prompt. This confirms the virtual environment is active.

Why a virtual environment?

dbt and its BigQuery adapter install many dependencies. A virtual environment ensures these packages stay contained in your project folder and do not interfere with other Python tools on your machine.

  • Upgrade pip before installing packages by running this command:
python -m pip install --upgrade pip wheel setuptools
  • Install dbt-bigquery and google-cloud-bigquery by running this command:
python -m pip install dbt-bigquery google-cloud-bigquery

What does this install?

The dbt-bigquery package installs dbt-core as a dependency, so you get both dbt and its BigQuery adapter in one command.

The google-cloud-bigquery library is what your Python loading script will use to send CSV data into BigQuery tables.

  • Verify dbt installed correctly by running:
dbt --version

You should see output listing dbt-core and dbt-bigquery with version numbers.

Seeing a Python version error?

The dbt-bigquery package requires Python 3.10 or newer (up to 3.13). If your installation fails, check your Python version with python3 --version. Python 3.14 is not yet supported.

Help me fix my dbt-bigquery installation error

Create BigQuery datasets for raw data and dbt models

Your pipeline needs two separate datasets in BigQuery. One will hold the raw data loaded by Python. The other will hold the clean, transformed models built by dbt.

  • Create the raw_banking dataset (for Python-loaded raw tables) by running this command:
bq mk --dataset raw_banking
  • Create the dbt_dev dataset (where dbt will build transformed models) by running this command:
bq mk --dataset dbt_dev

Why two separate datasets?

Separating raw and transformed data is a best practice in data engineering. The raw_banking dataset holds your source-of-truth data exactly as it arrived from CSV files.

The dbt_dev dataset holds the cleaned, tested, and joined models that analysts actually query. This separation makes it clear which data has been validated and which has not.

  • Verify both datasets exist by running:
bq ls

You should see both raw_banking and dbt_dev listed in the output.

Getting a permission denied error?

Make sure your gcloud init selected the correct project. Run gcloud config get-value project to check. If it shows the wrong project, run gcloud init again and select the correct one.

Help me fix my BigQuery dataset creation error

Your environment is fully configured. Next up, you will create sample banking data and load it into BigQuery using a Python script.

Create Sample Data and Load into BigQuery

Your environment is ready with authentication, packages, and empty BigQuery datasets waiting for data. Now it's time to fill those datasets with something real.

In a production ELT pipeline, the first two letters stand for Extract and Load. You pull raw data from a source and push it into your data warehouse before any transformations happen. In this step, you'll simulate that by creating sample banking CSV files and writing a Python script to load them into BigQuery.

In this step, get ready to:
  • Create three CSV data files representing customers, accounts, and transactions.
  • Write a Python script that loads each CSV into BigQuery using the client library.
  • Run the script and verify that all data landed in BigQuery.
Create three CSV files representing banking data

Real banking data comes from database exports and flat files. You'll create a small dataset that mirrors this pattern: customers who hold accounts, and transactions flowing through those accounts.

  • Create a data folder for your CSV files by running this command:
mkdir data
  • Confirm the folder exists by running:
ls

You should see data listed in the output.

  • In Cursor, create a new file called customers.csv inside the data folder.
  • Paste the following content into data/customers.csv:
customer_id,first_name,last_name,email,country,created_at
1,Raj,Sharma,raj.sharma@example.com,India,2024-01-15
2,Priya,Patel,priya.patel@example.com,India,2024-02-20
3,Amit,Kumar,amit.kumar@example.com,India,2024-03-10
4,Neha,Gupta,neha.gupta@example.com,India,2024-04-05
5,Vikram,Singh,vikram.singh@example.com,India,2024-05-12
6,Ananya,Reddy,ananya.reddy@example.com,India,2024-06-18
7,Suresh,Menon,suresh.menon@example.com,India,2024-07-22
8,Deepa,Nair,deepa.nair@example.com,India,2024-08-30
9,Arjun,Rao,arjun.rao@example.com,India,2024-09-14
10,Kavita,Joshi,kavita.joshi@example.com,India,2024-10-25

What does this file contain?

This file holds 10 bank customers with their basic profile information. Each row has a unique customer_id that other tables will reference to link accounts and transactions back to their owner.

  • Save data/customers.csv.
  • Create a second file called accounts.csv inside the data folder.
  • Paste the following content into data/accounts.csv:
account_id,customer_id,account_type,opened_date,status
101,1,savings,2024-01-16,active
102,1,current,2024-02-01,active
103,2,savings,2024-02-21,active
104,3,savings,2024-03-11,active
105,3,current,2024-03-15,active
106,4,savings,2024-04-06,active
107,5,savings,2024-05-13,active
108,6,current,2024-06-19,active
109,7,savings,2024-07-23,closed
110,8,savings,2024-08-31,active
111,9,current,2024-09-15,active
112,10,savings,2024-10-26,active

What does this file contain?

This file holds 12 bank accounts across the 10 customers. Notice that some customers have multiple accounts (customer 1 has both savings and current). The customer_id column links each account back to its owner.

  • Save data/accounts.csv.
  • Create a third file called transactions.csv inside the data folder.
  • Paste the following content into data/transactions.csv:
transaction_id,account_id,amount,transaction_type,transaction_date
1001,101,5000.00,credit,2024-06-01
1002,101,1500.00,debit,2024-06-05
1003,102,25000.00,credit,2024-06-10
1004,103,8000.00,credit,2024-06-12
1005,104,3000.00,debit,2024-06-15
1006,105,12000.00,credit,2024-06-18
1007,106,4500.00,credit,2024-06-20
1008,107,7000.00,debit,2024-06-22
1009,108,15000.00,credit,2024-06-25
1010,109,2000.00,debit,2024-06-28
1011,110,9500.00,credit,2024-07-01
1012,111,6000.00,credit,2024-07-05
1013,112,3500.00,debit,2024-07-08
1014,101,2000.00,credit,2024-07-10
1015,103,4000.00,debit,2024-07-12

What does this file contain?

This file holds 15 transactions across the accounts. Each transaction is either a credit (money in) or debit (money out), linked to an account via account_id.

These three tables form a classic relational data model: customers own accounts, and accounts have transactions.

  • Save data/transactions.csv.
  • Verify all three files exist by running:
ls data/

You should see accounts.csv, customers.csv, and transactions.csv listed.

Don't see all three files?

Make sure you created each file inside the data folder, not in the project root. If a file ended up in the wrong location, drag it into data/ using the Cursor sidebar.

Help me find my CSV files

Write a Python script to load CSVs into BigQuery

With your raw data ready, you need a script that reads each CSV and pushes it into a BigQuery table. The google-cloud-bigquery library handles the heavy lifting: schema detection, chunked uploads, and error handling.

  • Create a new file called load_to_bigquery.py in your project root (the same level as the data folder).
  • Add the imports and configuration at the top of load_to_bigquery.py:
from google.cloud import bigquery
import os

# Replace with your actual GCP project ID
PROJECT_ID = "[[GCP_PROJECT_ID=\"your-gcp-project-id-here\"]]"
DATASET_ID = "raw_banking"
DATA_DIR = "data"

# Create an authenticated BigQuery client
client = bigquery.Client(project=PROJECT_ID)

What does this code do?

  • PROJECT_ID holds your GCP project identifier. Replace the placeholder with the actual ID from your Google Cloud console.
  • DATASET_ID points to the raw_banking dataset you created in Step 1.
  • client creates an authenticated BigQuery connection using the application-default credentials you set up earlier.

Find your GCP project ID

Open the Google Cloud console. Your project ID appears in the project selector dropdown at the top of the page. It looks like my-project-123456 (not the project name, which can have spaces).

  • Below the client line, add the function that loads a single CSV file into BigQuery:
def load_csv_to_bigquery(file_name, table_name):
    # Build the fully qualified table path: project.dataset.table
    table_id = f"{PROJECT_ID}.{DATASET_ID}.{table_name}"
    file_path = os.path.join(DATA_DIR, file_name)

    # Configure the load job settings
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )

    # Upload the file and wait for the job to complete
    with open(file_path, "rb") as source_file:
        job = client.load_table_from_file(source_file, table_id, job_config=job_config)

    job.result()

    # Print confirmation with row count
    table = client.get_table(table_id)
    print(f"Loaded {table.num_rows} rows to {table_id}")

What does this function do?

  • table_id builds the fully qualified BigQuery table path: project.dataset.table.
  • autodetect=True tells BigQuery to infer column names and types from the CSV headers automatically.
  • WRITE_TRUNCATE means the table gets overwritten on each run. This makes the script safe to re-run without creating duplicate rows.
  • job.result() blocks until the upload finishes, then the function prints a confirmation with the row count.
  • Below the load_csv_to_bigquery function, add the main function and entry point:
def main():
    # Map each CSV filename to its target BigQuery table name
    tables = {
        "customers.csv": "customers",
        "accounts.csv": "accounts",
        "transactions.csv": "transactions",
    }

    # Load each CSV in sequence
    for file_name, table_name in tables.items():
        load_csv_to_bigquery(file_name, table_name)

    print("\nAll tables loaded successfully!")


if __name__ == "__main__":
    main()

What does this code do?

  • The tables dictionary maps each CSV filename to the BigQuery table name it should create.
  • The loop calls load_csv_to_bigquery for each file, loading all three tables in sequence.
  • The if __name__ guard ensures the script only runs when executed directly, not when imported.
  • Save load_to_bigquery.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved load_to_bigquery.py and let's run it.

ⓧ I'd like to double check the full code

Here is the complete load_to_bigquery.py file:

from google.cloud import bigquery
import os

PROJECT_ID = "[[GCP_PROJECT_ID=\"your-gcp-project-id-here\"]]"
DATASET_ID = "raw_banking"
DATA_DIR = "data"

client = bigquery.Client(project=PROJECT_ID)


def load_csv_to_bigquery(file_name, table_name):
    table_id = f"{PROJECT_ID}.{DATASET_ID}.{table_name}"
    file_path = os.path.join(DATA_DIR, file_name)

    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )

    with open(file_path, "rb") as source_file:
        job = client.load_table_from_file(source_file, table_id, job_config=job_config)

    job.result()

    table = client.get_table(table_id)
    print(f"Loaded {table.num_rows} rows to {table_id}")


def main():
    tables = {
        "customers.csv": "customers",
        "accounts.csv": "accounts",
        "transactions.csv": "transactions",
    }

    for file_name, table_name in tables.items():
        load_csv_to_bigquery(file_name, table_name)

    print("\nAll tables loaded successfully!")


if __name__ == "__main__":
    main()

Seeing a red underline on the import?

Make sure your virtual environment is activated in Cursor's terminal. If you see a warning about google.cloud not being found, your editor may be using the system Python instead of the venv. Select the correct interpreter by pressing Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) and choosing Python: Select Interpreter.

Help me troubleshoot this import error

Run the script and verify data in BigQuery

The script is ready. Time to execute it and confirm the data arrives safely in your raw_banking dataset.

  • Run the load script by executing this command:
python load_to_bigquery.py

You should see output confirming each table was loaded with its row count:

Loaded 10 rows to your-gcp-project-id-here.raw_banking.customers
Loaded 12 rows to your-gcp-project-id-here.raw_banking.accounts
Loaded 15 rows to your-gcp-project-id-here.raw_banking.transactions

All tables loaded successfully!

Seeing a permissions error?

If you see a 403 or "Access Denied" error, your application-default credentials may have expired. Re-run the authentication command from Step 1 to refresh them.

If you see "Not found: Dataset", double check that PROJECT_ID in your script matches your actual GCP project ID and that the raw_banking dataset exists in that project.

Help me fix this BigQuery error

  • Verify the data by querying one of the tables with the bq CLI:
bq query --nouse_legacy_sql 'SELECT COUNT(*) AS total_rows FROM `raw_banking.customers`'

You should see a result showing 10 total rows.

What does this command do?

The bq query command runs a SQL query against BigQuery from your terminal. The --nouse_legacy_sql flag tells it to use GoogleSQL syntax (the modern default). COUNT(*) returns the total number of rows in the table.

Your raw data is now sitting in BigQuery, ready to be transformed. Next up, you'll initialize a dbt project and build staging models that clean and standardize this data.

Initialize dbt and Build Staging Models

Your raw banking data is safely loaded into BigQuery. Now it's time to build the transformation layer that turns those raw tables into clean, analytics-ready models.

dbt (data build tool) handles the T in ELT. You'll initialize a dbt project, connect it to BigQuery using OAuth, then build staging models that standardize your raw data into a consistent format.

In this step, get ready to:
  • Initialize a dbt project and validate its connection to BigQuery.
  • Declare raw tables as dbt sources and build staging models.
  • Run your first dbt transformation and verify the results in BigQuery.
Initialize the dbt project and connect to BigQuery

The dbt init command scaffolds a new project with all the folders and config files dbt needs. You'll skip the interactive profile setup and configure the BigQuery connection manually for OAuth.

  • In your terminal, make sure you are in your project root directory (where load_to_bigquery.py lives).
  • Initialize the dbt project by running this command:
dbt init banking_pipeline --skip-profile-setup

What does this command do?

dbt init banking_pipeline creates a new folder called banking_pipeline with sample model files, a dbt_project.yml config, and standard directory structure.

The --skip-profile-setup flag tells dbt not to prompt you for connection details interactively. You'll configure the connection manually in the next chunk.

  • Confirm the project folder was created by running:
ls banking_pipeline

You should see folders like models, macros, tests, and a dbt_project.yml file.

Now you need to tell dbt how to connect to BigQuery. dbt stores connection profiles in a file at ~/.dbt/profiles.yml.

  • Create the .dbt directory and profiles.yml file by running these commands:
mkdir -p ~/.dbt
touch ~/.dbt/profiles.yml
  • Open ~/.dbt/profiles.yml in Cursor and paste the following configuration:
banking_pipeline:
  target: dev
  outputs:
    dev:
      type: bigquery
      method: oauth
      project: [[GCP_PROJECT_ID="your-gcp-project-id"]]
      dataset: dbt_dev
      threads: 4

What does this configuration do?

  • banking_pipeline matches the profile name in dbt_project.yml. This is how dbt knows which connection to use.
  • method: oauth tells dbt to use the application-default credentials you set up with gcloud auth application-default login in Step 1.
  • dataset: dbt_dev points dbt to the empty dataset you created for transformed models.
  • threads: 4 lets dbt run up to 4 models in parallel.
  • Save profiles.yml.
  • Move into the dbt project folder and validate the connection by running these commands:
cd banking_pipeline
dbt debug

You should see output ending with Connection test: OK connection ok. This confirms dbt can reach BigQuery using your OAuth credentials.

Connection test failed?

  • Check that your GCP project ID in profiles.yml matches the one you used in load_to_bigquery.py. Open BigQuery console to confirm the exact project ID.
  • Make sure you ran gcloud auth application-default login in Step 1. If your session expired, run it again.
  • Verify that the dbt_dev dataset exists in your BigQuery project.

Help me debug my dbt debug connection failure.

Declare sources and create staging models

Before dbt can transform your raw data, it needs to know where that data lives. You'll declare your BigQuery tables as sources, then build staging models that clean and standardize each table.

What is the staging/marts pattern?

dbt best practices recommend organizing models into two layers. Staging models sit directly on top of raw sources and handle type casting, renaming, and basic cleanup.

Mart models join and aggregate staged data into business-ready tables. This separation keeps transformations modular and testable.

First, update the project configuration to define how each layer should be materialized.

  • Open banking_pipeline/dbt_project.yml in Cursor and replace its entire contents with:
name: banking_pipeline
version: "1.0.0"

profile: banking_pipeline

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

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

What does the models section do?

The models block tells dbt how to materialize each folder. Staging models become views (lightweight SQL definitions that don't store data). Mart models become tables (physical copies that query faster).

  • Save dbt_project.yml.

Now create the staging folder structure and declare your raw tables as dbt sources.

  • Create the staging models directory by running:
mkdir -p models/staging
  • Delete the sample models that dbt init created by running:
rm -rf models/example
  • Create a new file models/staging/_sources.yml and paste the following:
version: 2

sources:
  - name: raw_banking
    database: "{{ var('project_id', target.database) }}"
    schema: raw_banking
    tables:
      - name: customers
      - name: accounts
      - name: transactions

What is a dbt source?

A source declaration tells dbt about tables that already exist in your warehouse (the raw tables you loaded with Python). Instead of hardcoding table names in SQL, you reference them with {{ source('raw_banking', 'customers') }}.

This gives dbt full visibility into your data lineage and lets it track where transformations start.

  • Save _sources.yml.

Now create the three staging models. Each one selects from a raw source, casts columns to proper types, and outputs a clean version.

  • Create a new file models/staging/stg_customers.sql and paste the following:
with source as (
    select * from {{ source('raw_banking', 'customers') }}
)

select
    cast(customer_id as int64) as customer_id,
    first_name,
    last_name,
    email,
    country,
    cast(created_at as date) as created_at
from source

What does this model do?

  • The with source CTE pulls all columns from the raw customers table using the source declaration you just created.
  • cast(customer_id as int64) ensures numeric IDs are stored as integers, not strings (BigQuery's autodetect sometimes infers the wrong type).
  • cast(created_at as date) converts the date string into a proper DATE type for accurate date operations.
  • Save stg_customers.sql.
  • Create a new file models/staging/stg_accounts.sql and paste the following:
with source as (
    select * from {{ source('raw_banking', 'accounts') }}
)

select
    cast(account_id as int64) as account_id,
    cast(customer_id as int64) as customer_id,
    account_type,
    cast(opened_date as date) as opened_date,
    status
from source

What does this model do?

Same pattern as stg_customers. It casts account_id and customer_id to integers, and opened_date to a proper DATE type. String columns like account_type and status pass through unchanged.

  • Save stg_accounts.sql.
  • Create a new file models/staging/stg_transactions.sql and paste the following:
with source as (
    select * from {{ source('raw_banking', 'transactions') }}
)

select
    cast(transaction_id as int64) as transaction_id,
    cast(account_id as int64) as account_id,
    cast(amount as numeric) as amount,
    transaction_type,
    cast(transaction_date as date) as transaction_date
from source

What does this model do?

This model casts amount to NUMERIC (a precision type for financial data) instead of FLOAT64, which avoids rounding errors in monetary calculations. The ID and date columns get the same treatment as the other staging models.

  • Save stg_transactions.sql.

Not sure if your files are in the right place?

Your models/staging/ folder should contain exactly four files: _sources.yml, stg_customers.sql, stg_accounts.sql, and stg_transactions.sql. Run ls models/staging/ to confirm.

Help me check my dbt staging model file structure.

✔️ Awesome, I've got everything!

Great. Make sure all four files in models/staging/ are saved.

ⓧ I'd like to double check the full code

Here is the complete dbt_project.yml:

name: banking_pipeline
version: "1.0.0"

profile: banking_pipeline

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

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

Here is models/staging/_sources.yml:

version: 2

sources:
  - name: raw_banking
    database: "{{ var('project_id', target.database) }}"
    schema: raw_banking
    tables:
      - name: customers
      - name: accounts
      - name: transactions

Here is models/staging/stg_customers.sql:

with source as (
    select * from {{ source('raw_banking', 'customers') }}
)

select
    cast(customer_id as int64) as customer_id,
    first_name,
    last_name,
    email,
    country,
    cast(created_at as date) as created_at
from source

Here is models/staging/stg_accounts.sql:

with source as (
    select * from {{ source('raw_banking', 'accounts') }}
)

select
    cast(account_id as int64) as account_id,
    cast(customer_id as int64) as customer_id,
    account_type,
    cast(opened_date as date) as opened_date,
    status
from source

Here is models/staging/stg_transactions.sql:

with source as (
    select * from {{ source('raw_banking', 'transactions') }}
)

select
    cast(transaction_id as int64) as transaction_id,
    cast(account_id as int64) as account_id,
    cast(amount as numeric) as amount,
    transaction_type,
    cast(transaction_date as date) as transaction_date
from source
Run dbt and verify your staging views

With your sources declared and staging models written, it's time to let dbt compile and execute the SQL against BigQuery.

  • Make sure you are inside the banking_pipeline directory in your terminal.
  • Run the dbt models by running this command:
dbt run

What does dbt run do?

dbt run compiles each .sql model file into raw SQL (resolving {{ source() }} references), then executes the SQL against BigQuery. Because staging models are configured as views, dbt runs a CREATE VIEW statement for each one.

You should see output showing all three models completed successfully, similar to:

Completed successfully

Done. PASS=3 WARN=0 ERROR=0 SKIP=0 TOTAL=3
  • Verify the views exist by opening the BigQuery console in your browser.
  • Expand your project in the left sidebar, then expand the dbt_dev dataset.
  • You should see three views listed: stg_customers, stg_accounts, and stg_transactions.
  • Click on stg_customers and select the Preview tab to confirm the data looks correct.

dbt run shows errors?

  • If you see a "source not found" error, check that _sources.yml is saved inside models/staging/ (not in the project root).
  • If you see a "dataset not found" error, confirm the raw_banking dataset exists and that your project ID in profiles.yml is correct.
  • If you see a permissions error, re-run gcloud auth application-default login and try dbt run again.

Help me debug my dbt run errors.

Your staging layer is live. You now have three clean, type-safe views sitting on top of your raw data. Next up, you'll build a mart model that joins these staging views into a powerful customer summary table and add automated tests to catch data quality issues.

Build a Mart Model and Add Tests

Your staging models are cleaning and standardizing raw data. Now it's time to combine them into something an analyst can query directly.

In this step, you'll create a mart model that joins customers, accounts, and transactions into a single summary table. You'll also add automated tests that catch data quality issues before they reach production, then generate documentation with a visual lineage DAG.

In this step, get ready to:
  • Create a mart model that produces a per-customer summary.
  • Add automated tests validating data quality across all models.
  • Generate documentation with a visual lineage DAG.
Create the customer_summary mart model

A mart model is the business-facing layer of your dbt project. It joins cleaned staging data into a shape that answers real questions.

Your customer_summary mart will produce one row per customer with their total accounts, credits, debits, net balance, and last transaction date.

  • Create the models/marts/ directory inside your banking_pipeline project by running this command:
mkdir models/marts
  • Create a new file called customer_summary.sql inside models/marts/ and paste the following code:
with customers as (
    select * from {{ ref('stg_customers') }}
),

accounts as (
    select * from {{ ref('stg_accounts') }}
),

transactions as (
    select * from {{ ref('stg_transactions') }}
),

account_transactions as (
    select
        accounts.customer_id,
        transactions.amount,
        transactions.transaction_type,
        transactions.transaction_date
    from transactions
    inner join accounts on transactions.account_id = accounts.account_id
),

customer_summary as (
    select
        customers.customer_id,
        customers.first_name,
        customers.last_name,
        customers.email,
        count(distinct a.account_id) as total_accounts,
        coalesce(sum(case when at.transaction_type = 'credit' then at.amount else 0 end), 0) as total_credits,
        coalesce(sum(case when at.transaction_type = 'debit' then at.amount else 0 end), 0) as total_debits,
        coalesce(sum(case when at.transaction_type = 'credit' then at.amount else -at.amount end), 0) as net_balance,
        max(at.transaction_date) as last_transaction_date
    from customers
    left join accounts as a on customers.customer_id = a.customer_id
    left join account_transactions as at on customers.customer_id = at.customer_id
    group by
        customers.customer_id,
        customers.first_name,
        customers.last_name,
        customers.email
)

select * from customer_summary

What does this code do?

This model uses five CTEs (Common Table Expressions) to build up the final result step by step.

  • The first three CTEs pull in each staging model using {{ ref('stg_customers') }}. The ref() function tells dbt about the dependency between models so it builds them in the correct order.
  • The account_transactions CTE joins transactions to accounts so you can trace each transaction back to a customer.
  • The final customer_summary CTE aggregates everything per customer: counting accounts, summing credits and debits, calculating net balance, and finding the most recent transaction date.
  • Save customer_summary.sql.
  • You won't see a change yet. The next substep adds tests, then you'll run everything together.

Why left join instead of inner join?

A left join from customers ensures every customer appears in the summary, even if they have no accounts or transactions yet. An inner join would silently drop customers with no activity.

✔️ Awesome, I've got everything!

Great. Double check you've saved customer_summary.sql in the models/marts/ directory.

ⓧ I'd like to double check the full code

with customers as (
    select * from {{ ref('stg_customers') }}
),

accounts as (
    select * from {{ ref('stg_accounts') }}
),

transactions as (
    select * from {{ ref('stg_transactions') }}
),

account_transactions as (
    select
        accounts.customer_id,
        transactions.amount,
        transactions.transaction_type,
        transactions.transaction_date
    from transactions
    inner join accounts on transactions.account_id = accounts.account_id
),

customer_summary as (
    select
        customers.customer_id,
        customers.first_name,
        customers.last_name,
        customers.email,
        count(distinct a.account_id) as total_accounts,
        coalesce(sum(case when at.transaction_type = 'credit' then at.amount else 0 end), 0) as total_credits,
        coalesce(sum(case when at.transaction_type = 'debit' then at.amount else 0 end), 0) as total_debits,
        coalesce(sum(case when at.transaction_type = 'credit' then at.amount else -at.amount end), 0) as net_balance,
        max(at.transaction_date) as last_transaction_date
    from customers
    left join accounts as a on customers.customer_id = a.customer_id
    left join account_transactions as at on customers.customer_id = at.customer_id
    group by
        customers.customer_id,
        customers.first_name,
        customers.last_name,
        customers.email
)

select * from customer_summary
Add dbt tests in properties files

dbt tests are assertions you declare in YAML. Each test becomes a SQL query that checks for failing rows. If the query returns zero rows, the test passes.

You'll add tests to both your staging models and your new mart model.

  • Create a new file called _models.yml inside models/staging/ and paste the following:
version: 2

models:
  - name: stg_customers
    description: Cleaned customer data from raw source
    columns:
      - name: customer_id
        description: Primary key for customers
        tests:
          - unique
          - not_null

  - name: stg_accounts
    description: Cleaned account data from raw source
    columns:
      - name: account_id
        description: Primary key for accounts
        tests:
          - unique
          - not_null
      - name: customer_id
        description: Foreign key to customers
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id

  - name: stg_transactions
    description: Cleaned transaction data from raw source
    columns:
      - name: transaction_id
        description: Primary key for transactions
        tests:
          - unique
          - not_null
      - name: account_id
        description: Foreign key to accounts
        tests:
          - not_null

What do these tests check?

  • unique ensures no duplicate values exist in a column. dbt checks this by looking for any value that appears more than once.
  • not_null ensures every row has a value in that column. No blanks allowed.
  • relationships checks referential integrity. Every customer_id in stg_accounts must have a matching customer_id in stg_customers.
  • Save models/staging/_models.yml.

✔️ Awesome, I've got everything!

Great. Make sure the indentation matches exactly. YAML is whitespace-sensitive.

ⓧ I'd like to double check the full code

version: 2

models:
  - name: stg_customers
    description: Cleaned customer data from raw source
    columns:
      - name: customer_id
        description: Primary key for customers
        tests:
          - unique
          - not_null

  - name: stg_accounts
    description: Cleaned account data from raw source
    columns:
      - name: account_id
        description: Primary key for accounts
        tests:
          - unique
          - not_null
      - name: customer_id
        description: Foreign key to customers
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id

  - name: stg_transactions
    description: Cleaned transaction data from raw source
    columns:
      - name: transaction_id
        description: Primary key for transactions
        tests:
          - unique
          - not_null
      - name: account_id
        description: Foreign key to accounts
        tests:
          - not_null
  • Now create a second _models.yml file inside models/marts/ and paste the following:
version: 2

models:
  - name: customer_summary
    description: One row per customer with aggregated account and transaction metrics
    columns:
      - name: customer_id
        description: Primary key - unique customer identifier
        tests:
          - unique
          - not_null
      - name: total_accounts
        description: Number of accounts owned by this customer
      - name: net_balance
        description: Total credits minus total debits across all accounts
      - name: last_transaction_date
        description: Date of the most recent transaction

What does this file do?

This properties file defines tests for your mart model. It ensures customer_id is unique and never null in the final summary table. The descriptions also feed into the auto-generated documentation you'll view shortly.

  • Save models/marts/_models.yml.

✔️ Awesome, I've got everything!

Great. You now have test definitions for both staging and mart models.

ⓧ I'd like to double check the full code

version: 2

models:
  - name: customer_summary
    description: One row per customer with aggregated account and transaction metrics
    columns:
      - name: customer_id
        description: Primary key - unique customer identifier
        tests:
          - unique
          - not_null
      - name: total_accounts
        description: Number of accounts owned by this customer
      - name: net_balance
        description: Total credits minus total debits across all accounts
      - name: last_transaction_date
        description: Date of the most recent transaction
Run tests and generate documentation

With the model and tests in place, it's time to build everything, validate the data, and generate a documentation site that visualizes your entire pipeline.

  • Build all models (including your new mart) by running:
dbt run

What does this command do?

Running dbt run compiles your SQL models and executes them against BigQuery. Your staging models rebuild as views, and customer_summary materializes as a table in the dbt_dev dataset (because your dbt_project.yml sets marts: +materialized: table).

  • You should see output confirming all 4 models completed successfully (3 staging views and 1 mart table).

Seeing a compilation error?

  • Check that your customer_summary.sql file is inside models/marts/ and not models/staging/.
  • Verify the model names inside ref() match your staging filenames exactly: stg_customers, stg_accounts, stg_transactions.
  • Make sure you are running the command from inside the banking_pipeline directory.

Why is my dbt run failing with a compilation error?

  • Run all your tests by running:
dbt test
  • You should see each test listed with a PASS status, followed by a summary line showing all tests passed with zero errors.

Seeing test failures?

  • A failing relationships test means a customer_id in stg_accounts does not exist in stg_customers. Check that your CSV data is consistent.
  • A failing unique test means duplicate values exist. Re-check your source CSVs for duplicate IDs.
  • Make sure you ran dbt run before dbt test. Tests require the models to be materialized first.

Why are my dbt tests failing?

  • Generate the documentation catalog by running:
dbt docs generate
  • Serve the documentation site locally by running:
dbt docs serve
  • Your browser should open automatically to a documentation site on port 8080.
  • Click on the lineage graph icon in the bottom-right corner to view the full DAG of your pipeline.
  • You should see your raw sources flowing into staging models, which then feed into the customer_summary mart.

What is the lineage DAG?

The lineage DAG (Directed Acyclic Graph) is a visual map of how data flows through your project. Each node represents a model or source, and the arrows show dependencies. This is how dbt knows the correct build order.

In production, teams use this to understand impact analysis. If a source table changes, you can instantly see which downstream models are affected.

Stop the docs server

Press Ctrl+C in your terminal to stop the docs server when you're done exploring.

Your pipeline is fully tested and documented. Next up, you'll tackle the Secret Mission and build an incremental model that processes only new data.

Secret mission

Add an Incremental Model

Your current dbt models rebuild every table from scratch on each run. In production, transaction tables grow by millions of rows daily. Can you build a model that only processes new data since the last run?

Clean Up Your Resources

Clean Up Your Resources

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

This project uses BigQuery's free tier, which includes 10 GiB of storage per month. Your datasets are well within this limit, so there are no ongoing costs if you leave them idle.

Resources you used:

  • BigQuery dataset: raw_banking (3 raw tables: customers, accounts, transactions).
  • BigQuery dataset: dbt_dev (3 staging views, 2 mart tables).
  • Local project folder on your Desktop (CSV data files, Python load script, dbt project, virtual environment).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with dbt models or adding more data.

  • Your BigQuery datasets stay well within the free tier at this size.
  • Your local project files stay on your Desktop, ready for you to pick up where you left off.
  • Your Python virtual environment remains in the project folder. Activate it again with source dbt-env/bin/activate (macOS/Linux) or dbt-env\Scripts\activate (Windows) when you return.

Pause - I'll come back to this later

Free up local resources while keeping your BigQuery data intact for later.

  • Deactivate your Python virtual environment by running:
deactivate
  • Your BigQuery datasets remain at no cost. The free tier covers idle storage for datasets this small.
  • When you return, navigate back to your project folder and reactivate the virtual environment.

Delete - I don't want to use this again

Remove all project resources and start fresh.

  • Delete both BigQuery datasets by running these commands in your terminal:
bq rm -r -f [[GCP_PROJECT_ID="your-gcp-project-id"]]:raw_banking
bq rm -r -f [[GCP_PROJECT_ID="your-gcp-project-id"]]:dbt_dev

What do these flags do?

  • -r removes all tables and views inside the dataset recursively.
  • -f forces the deletion without asking for confirmation.
  • Delete your local project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf banking_pipeline data load_to_bigquery.py
  • On Windows, run:
cd ~\Desktop
Remove-Item -Recurse -Force banking_pipeline, data, load_to_bigquery.py
  • If you created a virtual environment folder (e.g. dbt-env), delete that too:
rm -rf dbt-env
  • Confirm the datasets are gone by visiting the BigQuery console. You should no longer see raw_banking or dbt_dev listed under your project.

Nice Work!

Nice Work!

You just built a production-style ELT pipeline from scratch using the modern data stack. Your pipeline loads raw banking data into BigQuery, transforms it through dbt staging and mart models, validates quality with automated tests, and documents everything with a visual lineage DAG.

You've learned how to:

  • Loaded raw banking CSV data into BigQuery using a Python script with the google-cloud-bigquery library.
  • Built a dbt project with source declarations, staging models, and a mart model following the industry-standard staging/marts pattern.
  • Added automated data quality tests (unique, not_null, relationships) and generated documentation with a visual lineage DAG.
  • (Secret Mission) Implemented an incremental model that efficiently processes only new data instead of scanning the full table on every run.

Ready to quiz yourself?