Tutorial: dbt + Snowflake Identity Pipeline

Build a Customer 360 identity pipeline using dbt-kanoniv macros on Snowflake. You will normalize raw CRM and billing data with dbt, write a Kanoniv spec, reconcile with the Python SDK, and materialize golden records as dbt mart tables.

Time: 30 minutes Prerequisites: dbt 1.3+, Snowflake account, Python 3.9+, pip install kanoniv

What You Will Build

Configure a dbt project with the dbt-kanoniv package
Define Snowflake sources for raw CRM and Stripe tables
Write staging models that normalize names, emails, and phones with dbt-kanoniv macros
Author a Kanoniv spec with matching rules and survivorship
Reconcile records locally with the Python SDK
Enable dbt-kanoniv mart models for analytics

Step 1: Project Setup

Initialize a new dbt project and install the dbt-kanoniv package.

bash

dbt init customer_identity
cd customer_identity

Add dbt-kanoniv as a dependency:

packages.yml

packages: - git: "https://github.com/dreynow/dbt-kanoniv.git"revision: main

Install the package:

bash

dbt deps

Configure your dbt_project.yml to set the Snowflake profile and Kanoniv variables:

dbt_project.yml

name: customer_identityversion: '1.0.0'config-version: 2profile: snowflake_identity
model-paths: ["models"] macro-paths: ["macros"]
vars: kanoniv_default_country: 'US'kanoniv_email_normalize_gmail: truekanoniv_enable_models: falsekanoniv_schema: 'identity'

Make sure your ~/.dbt/profiles.yml has a Snowflake profile:

yaml

snowflake_identity:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
      database: ANALYTICS
      warehouse: TRANSFORM_WH
      schema: IDENTITY
      role: TRANSFORMER

Step 2: Define Sources

Assume you have two raw tables in Snowflake loaded by your existing data pipeline:

RAW.CRM_CONTACTS — CRM export with name, email, phone, company
RAW.STRIPE_CUSTOMERS — Stripe billing data with name, email, created date

Create a sources file so dbt can reference them:

sources.yml

version: 2sources:
name: rawdatabase: ANALYTICSschema: RAWtables: name: crm_contactsdescription: CRM contact export with name, email, phonecolumns:
name: iddescription: CRM contact primary key
name: name
name: email
name: phone
name: company
name: stripe_customersdescription: Stripe billing customer recordscolumns:
name: customer_iddescription: Stripe customer primary key
name: full_name
name: email
name: created_at

Verify the connection:

bash

dbt debug

Step 3: Staging Models

Create staging models that normalize raw data using dbt-kanoniv macros. These macros standardize names, emails, and phone numbers so that matching rules work consistently.

CRM staging model

stg_crm_contacts.sql

with source as ( <span class="hl-fn">select</span> * <span class="hl-fn">from</span> &#123;&#123; <span class="hl-fn">source</span>(<span class="hl-str">'raw'</span>, <span class="hl-str">'crm_contacts'</span>) &#125;&#125;
)
select id as contact_id, {{ dbt_kanoniv.normalize_name('name') }} as normalized_name, {{ dbt_kanoniv.normalize_email('email') }} as normalized_email, {{ dbt_kanoniv.normalize_phone('phone') }} as normalized_phone, {{ dbt_kanoniv.email_domain('email') }} as email_domain, {{ dbt_kanoniv.blocking_key(['email_domain', 'normalized_name']) }} as block_key, company, 'crm' as source_system
from source where email is not null

What each macro does:

normalize_name('name') — uppercases, trims, strips titles (Dr., Mr.) and suffixes (Jr., III)
normalize_email('email') — lowercases, trims, normalizes Gmail dot-addressing
normalize_phone('phone') — strips non-numeric characters, formats to E.164
email_domain('email') — extracts the domain portion after @
blocking_key([...]) — concatenates fields into a composite key to reduce comparison space

Stripe staging model

stg_stripe_customers.sql

with source as ( <span class="hl-fn">select</span> * <span class="hl-fn">from</span> &#123;&#123; <span class="hl-fn">source</span>(<span class="hl-str">'raw'</span>, <span class="hl-str">'stripe_customers'</span>) &#125;&#125;
)
select customer_id as contact_id, {{ dbt_kanoniv.normalize_name('full_name') }} as normalized_name, {{ dbt_kanoniv.normalize_email('email') }} as normalized_email, null as normalized_phone, {{ dbt_kanoniv.email_domain('email') }} as email_domain, {{ dbt_kanoniv.blocking_key(['email_domain', 'normalized_name']) }} as block_key, null as company, 'stripe' as source_system
from source where email is not null

Both models produce a uniform schema: contact_id, normalized_name, normalized_email, normalized_phone, email_domain, block_key, company, source_system. This consistency is what makes the Kanoniv spec work — all sources share the same field names.

Build the staging models:

bash

dbt run --select staging

Step 4: Write the Kanoniv Spec

Create a spec that tells Kanoniv how to match records across the two staged sources. The spec uses the dbt adapter which reads from dbt's manifest.json to locate the staging tables.

customer-spec.yaml

api_version: "kanoniv/v2"identity_version: "customer_360_v1"metadata: name: customer-360-snowflakedescription: Unify customer identities from CRM and Stripe via dbt staging modelsentity: name: customer
sources:
name: crmadapter: dbtlocation: ref('stg_crm_contacts')primary_key: contact_idattributes: name: normalized_nameemail: normalized_emailphone: normalized_phonecompany: company
name: stripeadapter: dbtlocation: ref('stg_stripe_customers')primary_key: contact_idattributes: name: normalized_nameemail: normalized_email
rules:
name: email_exacttype: exactfield: emailweight: 1.0
name: name_fuzzytype: similarityfield: namealgorithm: jaro_winklerthreshold: 0.85weight: 0.6
survivorship: default: source_priorityoverrides: - field: namestrategy: source_prioritypriority: [crm, stripe] - field: emailstrategy: source_prioritypriority: [crm, stripe]
decision: thresholds: match: 0.85review: 0.60

Key points:

adapter: dbt — sources read from dbt-materialized tables via location: ref('model_name'). No CSV paths needed.
attributes — map canonical field names to the normalized column names from your staging models.
Two rules — exact email (weight 1.0) catches identical emails; fuzzy name with Jaro-Winkler (weight 0.6, threshold 0.85) catches variations like "Bob Johnson" vs "Robert Johnson".
source_priority survivorship — the default strategy is source_priority, with per-field overrides that prioritize CRM values, with Stripe filling in gaps.
Decision thresholds — 0.85+ is an automatic match; 0.60–0.85 goes to a review queue.

Step 5: Reconcile with the Python SDK

With staging models built and the spec written, run reconciliation locally using the Kanoniv Python SDK. The Source.from_dbt() adapter reads directly from the dbt manifest.

Validate the spec

python

from kanoniv import Spec, validate

spec = Spec.from_file("specs/customer-spec.yaml")

result = validate(spec)
result.raise_on_error()

print("Spec is valid!")
print(f"  Sources: {len(spec.sources)}")
print(f"  Rules: {len(spec.rules)}")

Load sources from dbt

python

from kanoniv import Source, reconcile

sources = [
    Source.from_dbt("crm", manifest="target/manifest.json", model="stg_crm_contacts"),
    Source.from_dbt("stripe", manifest="target/manifest.json", model="stg_stripe_customers"),
]

Source.from_dbt() reads the compiled manifest to locate each model's warehouse table, then queries Snowflake to pull the staged records into the local reconciliation engine.

Run reconciliation

python

result = reconcile(sources, spec)

print(f"Records processed: {result.total_records}")
print(f"Clusters found: {len(result.clusters)}")
print(f"Golden records: {len(result.golden_records)}")
print(f"Merge rate: {result.merge_rate:.1%}")

Inspect clusters

python

for cluster in result.clusters:
    if len(cluster.members) > 1:
        print(f"Cluster {cluster.id} ({len(cluster.members)} records):")
        for member in cluster.members:
            print(f"  [{member.source}] {member.fields.get('name', 'N/A')} "
                  f"<{member.fields.get('email', 'N/A')}>")
        print()

Export to DataFrame

python

df = result.to_pandas()
print(df[["golden_id", "name", "email", "company", "source_count"]].to_string(index=False))

The golden records now contain one canonical row per identity cluster, with CRM values winning via the source_priority survivorship default with per-field priority overrides.

Step 6: Analytics with dbt Marts

Once you have verified the reconciliation results, enable the built-in dbt-kanoniv mart models. These models read from the Kanoniv output tables and provide analytics-ready views.

Update dbt_project.yml to enable the models and point them at the correct schema:

dbt_project.yml

name: customer_identityversion: '1.0.0'config-version: 2profile: snowflake_identity
model-paths: ["models"] macro-paths: ["macros"]
vars: kanoniv_default_country: 'US'kanoniv_email_normalize_gmail: truekanoniv_enable_models: true # <-- enabledkanoniv_schema: 'identity'kanoniv_database: 'ANALYTICS'

Build everything — your staging models plus the Kanoniv mart models:

bash

dbt build

The dbt build command runs your staging models first, then the Kanoniv package models. After it completes, you can query the analytics tables directly in Snowflake:

query_overview.sql

-- Canonical overview: one row per golden recordselect canonical_id, canonical_name, canonical_email, source_count, first_seen_at, last_seen_at from {{ ref('kanoniv__canonical_overview') }} order by source_count desclimit 20;

query_duplicates.sql

-- Find records that were merged across sourcesselect canonical_id, source_system, source_id, normalized_name, normalized_email, match_score from {{ ref('kanoniv__identity_links') }} where match_score >= 0.85order by canonical_id, source_system;

Production Workflow

In production, schedule dbt build on a cron (via dbt Cloud, Airflow, or Dagster). Each run picks up new raw records, normalizes them through staging, and the Kanoniv models keep the golden records up to date.

Summary

You built a complete identity resolution pipeline:

Layer	Tool	What It Does
Data loading	Snowflake / Fivetran	Loads raw CRM and Stripe data
Normalization	dbt + dbt-kanoniv macros	Standardizes names, emails, phones
Matching	Kanoniv spec	Declares exact and fuzzy rules
Reconciliation	Kanoniv Python SDK	Clusters records, produces golden records
Analytics	dbt-kanoniv mart models	Materialized tables for BI dashboards

Next Steps

Customer 360 Tutorial — Learn the basics with CSV files before adding dbt
Spec Reference — Full reference for all spec options
Python SDK — Complete API documentation for Source.from_dbt() and reconcile()
Source Adapters — All available adapters: CSV, pandas, warehouse, dbt

Tutorial: dbt + Snowflake Identity Pipeline ​

What You Will Build ​

Step 1: Project Setup ​

Step 2: Define Sources ​

Step 3: Staging Models ​

CRM staging model ​

Stripe staging model ​

Step 4: Write the Kanoniv Spec ​

Step 5: Reconcile with the Python SDK ​

Validate the spec ​

Load sources from dbt ​

Run reconciliation ​

Inspect clusters ​

Export to DataFrame ​

Step 6: Analytics with dbt Marts ​

Summary ​

Next Steps ​

Tutorial: dbt + Snowflake Identity Pipeline

What You Will Build

Step 1: Project Setup

Step 2: Define Sources

Step 3: Staging Models

CRM staging model

Stripe staging model

Step 4: Write the Kanoniv Spec

Step 5: Reconcile with the Python SDK

Validate the spec

Load sources from dbt

Run reconciliation

Inspect clusters

Export to DataFrame

Step 6: Analytics with dbt Marts

Summary

Next Steps