Tutorial: Snowflake + dbt + Kanoniv Cloud

Build a production identity resolution pipeline that extracts customer data from Snowflake, normalizes it with dbt, reconciles identities through Kanoniv Cloud, and writes the golden records back to Snowflake - all in a single Python script.

Time: 45 minutes Difficulty: Intermediate

What You Will Build

Snowflake (3 sources)  ->  dbt (stage + normalize)  ->  Kanoniv Cloud (reconcile)  ->  Snowflake (golden records)

Three raw Snowflake tables representing CRM contacts, billing customers, and support tickets
dbt staging models that normalize names, emails, and phone numbers
A Kanoniv spec that defines matching rules across all three sources
A Python script that reads staged data, sends it to Kanoniv Cloud for reconciliation, and writes golden records back to Snowflake

Prerequisites

Snowflake account with a warehouse and a database you can write to
dbt Core 1.3+ configured with a Snowflake profile (pip install dbt-snowflake)
Python 3.9+ with the cloud extras: pip install kanoniv[cloud] snowflake-connector-python pandas
Kanoniv Cloud API key - sign up at app.kanoniv.com and generate a key from Settings

Step 1: The Raw Data in Snowflake

Assume your data pipelines (Fivetran, Airbyte, or custom ETL) have loaded three raw tables into Snowflake:

`RAW.CRM_CONTACTS` - Salesforce CRM

ID	NAME	EMAIL	PHONE	COMPANY	CREATED_AT
sf_001	John Doe	[email protected]	555-0101	Acme Corp	2025-01-15
sf_002	Jane Smith	[email protected]	555-0102	Globex Inc	2025-02-20
sf_003	Bob Wilson	[email protected]	555-0103	Acme Corp	2025-03-10
sf_004	Alice Brown	[email protected]	(555) 010-4	Startup LLC	2025-04-05

`RAW.BILLING_CUSTOMERS` - Stripe

CUSTOMER_ID	FULL_NAME	EMAIL	PLAN	MRR	CREATED_AT
cus_001	Jonathan Doe	[email protected]	enterprise	499	2025-01-20
cus_002	Jane Smith	[email protected]	pro	99	2025-03-01
cus_003	R. Wilson	[email protected]	starter	29	2025-03-15
cus_005	Eve Martinez	[email protected]	pro	99	2025-05-01

`RAW.SUPPORT_TICKETS` - Zendesk

TICKET_ID	REQUESTER_NAME	REQUESTER_EMAIL	PRIORITY	STATUS	CREATED_AT
tkt_100	J. Doe	[email protected]	high	closed	2025-02-01
tkt_101	Jane S.	[email protected]	medium	open	2025-04-10
tkt_102	Alice B	[email protected]	low	closed	2025-05-15
tkt_103	Eve M	[email protected]	high	open	2025-06-01

Notice the overlaps: "John Doe" / "Jonathan Doe" / "J. Doe" all share the same email. "Bob Wilson" / "R. Wilson" share an email too. These are the same people represented differently across three systems.

Step 2: dbt Project Setup

Initialize a dbt project and install the dbt-kanoniv package for normalization macros.

bash

dbt init customer_identity
cd customer_identity

Add the package dependency:

packages.yml

yaml

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

bash

dbt deps

Configure the project:

dbt_project.yml

yaml

name: customer_identity
version: '1.0.0'
config-version: 2

profile: snowflake_identity

model-paths: ["models"]
macro-paths: ["macros"]

vars:
  kanoniv_default_country: 'US'
  kanoniv_email_normalize_gmail: true
  kanoniv_enable_models: false
  kanoniv_schema: 'identity'

Make sure your ~/.dbt/profiles.yml has the Snowflake connection:

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 3: Define Sources

Create a dbt sources file that references the three raw tables:

models/sources.yml

yaml

version: 2

sources:
  - name: raw
    database: ANALYTICS
    schema: RAW
    tables:
      - name: crm_contacts
        description: Salesforce CRM contact records
        columns:
          - name: id
            description: Salesforce contact ID
          - name: name
          - name: email
          - name: phone
          - name: company
          - name: created_at

      - name: billing_customers
        description: Stripe billing customer records
        columns:
          - name: customer_id
            description: Stripe customer ID
          - name: full_name
          - name: email
          - name: plan
          - name: mrr
          - name: created_at

      - name: support_tickets
        description: Zendesk support ticket records
        columns:
          - name: ticket_id
            description: Zendesk ticket ID
          - name: requester_name
          - name: requester_email
          - name: priority
          - name: status
          - name: created_at

Verify the connection:

bash

dbt debug

Step 4: Staging Models

Create staging models that normalize each source into a consistent schema using dbt-kanoniv macros.

CRM staging model

models/staging/stg_crm_contacts.sql

sql

with source as (

    select * from {{ source('raw', 'crm_contacts') }}

)

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,
    company,
    created_at,
    'crm'                                           as source_system

from source
where email is not null

Billing staging model

models/staging/stg_billing_customers.sql

sql

with source as (

    select * from {{ source('raw', 'billing_customers') }}

)

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,
    null                                                 as company,
    created_at,
    'billing'                                            as source_system

from source
where email is not null

Support staging model

models/staging/stg_support_tickets.sql

sql

with source as (

    select * from {{ source('raw', 'support_tickets') }}

)

select
    ticket_id                                                 as contact_id,
    {{ dbt_kanoniv.normalize_name('requester_name') }}        as normalized_name,
    {{ dbt_kanoniv.normalize_email('requester_email') }}      as normalized_email,
    null                                                      as normalized_phone,
    null                                                      as company,
    created_at,
    'support'                                                 as source_system

from source
where requester_email is not null

All three models produce the same output schema: contact_id, normalized_name, normalized_email, normalized_phone, company, created_at, source_system. This consistency is what makes the Kanoniv spec work across all sources.

Build the staging models:

bash

dbt run --select staging

Step 5: Write the Kanoniv Spec

Create a spec that defines how to match records across the three staged sources.

specs/customer-cloud.yaml

yaml

api_version: kanoniv/v1
identity_version: "1.0"

entity:
  name: customer

sources:
  crm:
    adapter: snowflake
    primary_key: contact_id
    schema:
      name: { type: string }
      email: { type: string, pii: true }
      phone: { type: string, pii: true }
      company: { type: string }

  billing:
    adapter: snowflake
    primary_key: contact_id
    schema:
      name: { type: string }
      email: { type: string, pii: true }

  support:
    adapter: snowflake
    primary_key: contact_id
    schema:
      name: { type: string }
      email: { type: string, pii: true }

rules:
  - name: email_exact
    type: exact
    field: email
    weight: 1.0

  - name: name_fuzzy
    type: similarity
    field: name
    algorithm: jaro_winkler
    threshold: 0.85
    weight: 0.6

  - name: phone_exact
    type: exact
    field: phone
    weight: 0.8

decision:
  scoring: weighted_sum
  thresholds:
    match: 0.85
    review: 0.60

survivorship:
  strategy: source_priority
  source_order: [crm, billing, support]

Key design decisions:

Three rules - exact email is the strongest signal (weight 1.0); fuzzy name catches variations like "John Doe" / "J. Doe" (weight 0.6, Jaro-Winkler 0.85 threshold); exact phone provides an additional linkage path (weight 0.8)
Source priority - CRM is the system of record, billing fills in gaps, support is lowest priority
Match threshold at 0.85 - email match alone (1.0) exceeds the threshold; name similarity alone (0.6) does not, which prevents false positives from common names
Review band at 0.60-0.85 - edge cases land in a review queue instead of being auto-merged or discarded

Step 6: Reconcile with Kanoniv Cloud

This is the core script. It reads the staged data from Snowflake, sends it to Kanoniv Cloud for reconciliation, and writes the golden records back.

reconcile_cloud.py

python

import os

import pandas as pd
import snowflake.connector

import kanoniv
from kanoniv import Source, Spec


# ── Configuration ─────────────────────────────────────────────
SNOWFLAKE_CONFIG = {
    "account": os.environ["SNOWFLAKE_ACCOUNT"],     # e.g. "xy12345.us-east-1"
    "user": os.environ["SNOWFLAKE_USER"],
    "password": os.environ["SNOWFLAKE_PASSWORD"],
    "database": "ANALYTICS",
    "schema": "IDENTITY",
    "warehouse": "TRANSFORM_WH",
}

KANONIV_API_KEY = os.environ["KANONIV_API_KEY"]      # e.g. "kn_live_abc123..."


# ── Step 1: Read staged data from Snowflake ──────────────────
def read_staged_table(conn, table_name: str) -> pd.DataFrame:
    """Read a staged dbt model from Snowflake into a DataFrame."""
    query = f"SELECT * FROM ANALYTICS.IDENTITY.{table_name}"
    cursor = conn.cursor()
    cursor.execute(query)
    columns = [desc[0].lower() for desc in cursor.description]
    rows = cursor.fetchall()
    return pd.DataFrame(rows, columns=columns)


conn = snowflake.connector.connect(**SNOWFLAKE_CONFIG)

print("Reading staged data from Snowflake...")
df_crm = read_staged_table(conn, "STG_CRM_CONTACTS")
df_billing = read_staged_table(conn, "STG_BILLING_CUSTOMERS")
df_support = read_staged_table(conn, "STG_SUPPORT_TICKETS")

print(f"  CRM:     {len(df_crm):,} records")
print(f"  Billing: {len(df_billing):,} records")
print(f"  Support: {len(df_support):,} records")


# ── Step 2: Build Kanoniv sources from DataFrames ────────────
# Rename normalized columns to match the spec's schema field names
crm_source = Source.from_pandas(
    "crm",
    df_crm.rename(columns={
        "normalized_name": "name",
        "normalized_email": "email",
        "normalized_phone": "phone",
    }),
    primary_key="contact_id",
)

billing_source = Source.from_pandas(
    "billing",
    df_billing.rename(columns={
        "normalized_name": "name",
        "normalized_email": "email",
    }),
    primary_key="contact_id",
)

support_source = Source.from_pandas(
    "support",
    df_support.rename(columns={
        "normalized_name": "name",
        "normalized_email": "email",
    }),
    primary_key="contact_id",
)


# ── Step 3: Load the spec and reconcile on Kanoniv Cloud ─────
spec = Spec.from_file("specs/customer-cloud.yaml")

print("\nSubmitting to Kanoniv Cloud...")
with kanoniv.cloud.reconcile(
    [crm_source, billing_source, support_source],
    spec,
    api_key=KANONIV_API_KEY,
) as result:
    print(result.summary())
    # Cloud Reconciliation - job a1b2c3d4-...
    #   Status:          completed
    #   Health:          healthy
    #   Input entities:  12
    #   Canonicals:      7
    #   Merge rate:      41.7%

    # ── Step 4: Fetch golden records ─────────────────────────
    df_golden = result.to_pandas()
    print(f"\nGolden records: {len(df_golden)}")
    print(df_golden[["canonical_id", "name", "email", "company"]].to_string(index=False))


# ── Step 5: Write golden records back to Snowflake ───────────
print("\nWriting golden records to Snowflake...")

cursor = conn.cursor()

# Create the destination table if it does not exist
cursor.execute("""
    CREATE TABLE IF NOT EXISTS ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS (
        canonical_id    VARCHAR(64)   PRIMARY KEY,
        name            VARCHAR(256),
        email           VARCHAR(256),
        phone           VARCHAR(32),
        company         VARCHAR(256),
        source_count    INTEGER,
        reconciled_at   TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
    )
""")

# Truncate and reload (full-refresh pattern)
cursor.execute("TRUNCATE TABLE ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS")

# Insert golden records
insert_sql = """
    INSERT INTO ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS
        (canonical_id, name, email, phone, company, source_count)
    VALUES (%(canonical_id)s, %(name)s, %(email)s, %(phone)s,
            %(company)s, %(source_count)s)
"""

records = df_golden.to_dict("records")
for record in records:
    cursor.execute(insert_sql, {
        "canonical_id": record.get("canonical_id", ""),
        "name": record.get("name", ""),
        "email": record.get("email", ""),
        "phone": record.get("phone", ""),
        "company": record.get("company", ""),
        "source_count": record.get("source_count", 1),
    })

conn.commit()
print(f"Wrote {len(records)} golden records to ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS")


# ── Step 6: Query the live identity graph ────────────────────
from kanoniv import Client

with Client(api_key=KANONIV_API_KEY) as client:
    # Resolve a CRM contact to its canonical identity
    result = client.resolve(system="crm", external_id="sf_001")
    print(f"\nCRM contact sf_001 resolved to canonical: {result['canonical_id']}")

    # See every linked record across all three systems
    linked = client.entities.get_linked(result["canonical_id"])
    print("Linked records:")
    for record in linked["linked"]:
        print(f"  [{record['source_name']}] {record['external_id']}")
    # [crm] sf_001
    # [billing] cus_001
    # [support] tkt_100

conn.close()
print("\nDone.")

Optional: Arrow Fast Path

If you install the dataplane extra, the same pipeline runs ~6x faster using columnar Arrow reads and Parquet bulk ingest instead of row-by-row JSON uploads.

bash

pip install kanoniv[cloud,dataplane]

With the dataplane extra installed, you can replace the manual read_staged_table() + Source.from_pandas() approach with Source.from_warehouse(). The Arrow path is auto-detected:

python

import kanoniv
from kanoniv import Source, Spec

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

# Source.from_warehouse() replaces manual Snowflake reads + Source.from_pandas()
crm = Source.from_warehouse(
    "crm",
    "ANALYTICS.IDENTITY.STG_CRM_CONTACTS",
    connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)

billing = Source.from_warehouse(
    "billing",
    "ANALYTICS.IDENTITY.STG_BILLING_CUSTOMERS",
    connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)

support = Source.from_warehouse(
    "support",
    "ANALYTICS.IDENTITY.STG_SUPPORT_TICKETS",
    connection_string="snowflake://user:pass@acct/ANALYTICS/IDENTITY",
)

# Arrow path is used automatically - same API, faster ingest
with kanoniv.cloud.reconcile(
    [crm, billing, support],
    spec,
    api_key=os.environ["KANONIV_API_KEY"],
) as result:
    print(result.summary())
    df = result.to_pandas()

What changes under the hood:

Step	Without dataplane	With dataplane
Read from Snowflake	Row-by-row cursor + pandas	Columnar Arrow fetch
Column mapping	Manual `.rename(columns={...})`	Automatic via spec attribute mappings
Upload to API	500-record JSON batches	Parquet file per source
Server ingest	Row-by-row INSERT	COPY FROM Parquet

The column renaming from normalized_name to name is handled automatically by the staging layer using your spec's attribute mappings - no manual .rename() needed.

See the Arrow Data Plane guide for the full architecture and manual pipeline options.

Run it:

bash

export SNOWFLAKE_ACCOUNT="xy12345.us-east-1"
export SNOWFLAKE_USER="KANONIV_SVC"
export SNOWFLAKE_PASSWORD="your-password"
export KANONIV_API_KEY="kn_live_abc123..."

python reconcile_cloud.py

Expected output:

Reading staged data from Snowflake...
  CRM:     4 records
  Billing: 4 records
  Support: 4 records

Submitting to Kanoniv Cloud...
Cloud Reconciliation - job a1b2c3d4-e5f6-7890-abcd-ef1234567890
  Status:          completed
  Duration:        1,240ms
  Health:          healthy
  Input entities:  12
  Canonicals:      5
  Links:           12
  Merge rate:      58.3%

Golden records: 5
     canonical_id            name              email       company
 c7e2f9a1-...        John Doe      [email protected]     Acme Corp
 b3d4e5f6-...      Jane Smith  [email protected]  Globex Inc
 a1b2c3d4-...      Bob Wilson       [email protected]     Acme Corp
 d5e6f7g8-...     Alice Brown    [email protected]   Startup LLC
 e7f8g9h0-...    Eve Martinez       [email protected]          None

Writing golden records to Snowflake...
Wrote 5 golden records to ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS

CRM contact sf_001 resolved to canonical: c7e2f9a1-...
Linked records:
  [crm] sf_001
  [billing] cus_001
  [support] tkt_100

Done.

Step 7: Real-Time Resolution with Probabilistic Scoring

After the batch reconciliation in Step 6, your identity graph is live. But the real power is what happens next: when a new record arrives between batch runs, you can resolve it against the existing graph in real time using the same Fellegi-Sunter probabilistic model that powers your batch pipeline.

The POST /v1/resolve/realtime endpoint uses your compiled spec to score incoming records against existing canonical entities. It generates blocking keys, retrieves candidates, and applies the full FS scoring pipeline - not just exact field matching.

reconcile_realtime.py

python

import os
from kanoniv import Client

KANONIV_API_KEY = os.environ["KANONIV_API_KEY"]

with Client(api_key=KANONIV_API_KEY) as client:
    # A new support ticket arrives between batch runs.
    # Resolve it against the existing identity graph.
    result = client.post("/v1/resolve/realtime", json={
        "source": "support",
        "external_id": "tkt_200",
        "fields": {
            "name": "Jonathan Doe",
            "email": "[email protected]",
            "company": "Acme Corp",
        },
    })

    if result.get("matched"):
        print(f"Matched to canonical entity: {result['canonical_id']}")
        print(f"  Score: {result['score']:.2f}")
        print(f"  Matched on: {result['matched_on']}")
        # Matched to canonical entity: c7e2f9a1-...
        #   Score: 11.4
        #   Matched on: ['email_exact', 'name_jaro_winkler', 'company_last_name']
    else:
        print(f"New entity created: {result['canonical_id']}")

    # When a record bridges two existing entities, the endpoint
    # can merge them inline if the scores exceed merge_threshold.
    bridging = client.post("/v1/resolve/realtime", json={
        "source": "billing",
        "external_id": "cus_new_001",
        "fields": {
            "name": "Alice Brown",
            "email": "[email protected]",
            "phone": "555-0104",
        },
    })

    if bridging.get("merged_entities"):
        print(f"Merged {len(bridging['merged_entities'])} entities into {bridging['canonical_id']}")

How it works under the hood:

Blocking key generation. The endpoint generates composite blocking keys from the input fields (email, phone, last_name+first_name, company+last_name) and queries canonical_entities via SQL to find candidates.
Fellegi-Sunter scoring. Each candidate is scored using the same compiled FS plan from your spec - with the same normalizers (email, phone, name, nickname, domain) and comparators (exact, jaro_winkler, levenshtein, soundex, metaphone, cosine).
Two-threshold decision. The endpoint uses two thresholds:
- match_threshold (from your spec's scoring.thresholds.match) - for linking the new record to an existing entity
- merge_threshold (from scoring.thresholds.merge, defaults to match * 1.5) - for merging two existing entities when the new record bridges them
Inline merge. If the new record matches N existing entities above merge_threshold, the losers are merged into the highest-scoring winner with a full audit trail (entity_merges, entity_events).
Fallback. If no compiled FS plan exists (e.g., you have not run a batch reconciliation yet), the endpoint falls back to exact email/phone matching. This keeps the API backwards compatible.

You can configure the merge threshold in your spec:

yaml

scoring:
  method: fellegi_sunter
  thresholds:
    match: 8.0
    possible: 4.0
    non_match: -4.0
    merge: 12.0  # optional, defaults to match * 1.5

The higher merge threshold prevents accidental entity merges from borderline scores. Only high-confidence bridging records trigger inline merges.

What Just Happened?

Here is the full flow, step by step:

1. Extract (Snowflake -> dbt)

Your raw data lives in three Snowflake tables loaded by separate pipelines. The dbt staging models applied dbt-kanoniv normalization macros to standardize names (uppercase, strip titles), emails (lowercase, Gmail dot-trick), and phone numbers (E.164 format). This preprocessing is critical - without it, "John Doe" and "J. Doe" would never be compared.

2. Load (Snowflake -> Python)

The Python script read the staged tables into pandas DataFrames and wrapped them as Kanoniv Source objects. Column names were remapped from normalized_name to name to match the spec's schema declarations.

3. Upload (Python -> Kanoniv Cloud)

kanoniv.cloud.reconcile() handled the upload automatically:

Uploaded the YAML spec to the API
Created source registrations on the server
Batched the records (500 per request) and ingested them via /v1/ingest/batch

4. Reconcile (Kanoniv Cloud)

The cloud engine ran the full reconciliation pipeline:

Block - grouped records by email domain to reduce the comparison space
Compare - applied all three rules (exact email, fuzzy name, exact phone) to every candidate pair within each block
Score - computed weighted scores: email match = 1.0, name similarity = 0.6 * Jaro-Winkler, phone match = 0.8
Decide - pairs scoring above 0.85 were auto-matched; pairs between 0.60 and 0.85 went to review
Cluster - connected components formed identity clusters (e.g., sf_001 + cus_001 + tkt_100 all linked by email)
Survive - CRM fields won via source priority; billing and support fields filled gaps

5. Write Back (Python -> Snowflake)

The script fetched the golden records from the cloud API via result.to_pandas() and wrote them to ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS using a truncate-and-reload pattern.

6. Live Resolution

After reconciliation, the identity graph stays live on Kanoniv Cloud. Any downstream service can call client.resolve(system="crm", external_id="sf_001") to get the canonical identity in real time - no re-running the pipeline.

7. Real-Time Probabilistic Resolution

The POST /v1/resolve/realtime endpoint goes further. When a new record arrives between batch runs, it scores the record against existing canonical entities using the same Fellegi-Sunter model from your spec. If the record matches an entity above match_threshold, it is linked. If it bridges multiple entities above merge_threshold, they are merged inline with a full audit trail. No batch re-run needed.

Scheduling in Production

For production, wrap the reconciliation in an orchestrator so it runs after dbt completes.

Airflow example

python

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG("identity_resolution", schedule="0 6 * * *", start_date=datetime(2026, 1, 1)):

    dbt_run = BashOperator(
        task_id="dbt_staging",
        bash_command="cd /opt/dbt/customer_identity && dbt run --select staging",
    )

    reconcile = BashOperator(
        task_id="reconcile_cloud",
        bash_command="python /opt/scripts/reconcile_cloud.py",
    )

    dbt_run >> reconcile

dbt Cloud + webhook

If you use dbt Cloud, trigger the Python script via a dbt Cloud webhook that fires on successful run completion.

Querying the Results in Snowflake

Once golden records are written back, your analysts and BI tools can query them directly.

Customer 360 view

sql

-- Join golden records with source data for a full Customer 360
SELECT
    g.canonical_id,
    g.name,
    g.email,
    g.company,
    g.source_count,
    b.plan,
    b.mrr,
    s.priority AS last_ticket_priority,
    s.status   AS last_ticket_status
FROM ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS g
LEFT JOIN ANALYTICS.RAW.BILLING_CUSTOMERS b
    ON g.email = LOWER(b.email)
LEFT JOIN (
    SELECT DISTINCT ON (requester_email) *
    FROM ANALYTICS.RAW.SUPPORT_TICKETS
    ORDER BY requester_email, created_at DESC
) s
    ON g.email = LOWER(s.requester_email)
ORDER BY g.source_count DESC;

Duplicate report

sql

-- Find customers matched across multiple systems
SELECT
    canonical_id,
    name,
    email,
    source_count
FROM ANALYTICS.IDENTITY.GOLDEN_CUSTOMERS
WHERE source_count > 1
ORDER BY source_count DESC;

Troubleshooting

"No records returned from staging table" - Make sure dbt run --select staging completed successfully. Check that the tables exist in ANALYTICS.IDENTITY with SHOW TABLES IN SCHEMA ANALYTICS.IDENTITY.

"Cloud reconciliation job timed out" - The default timeout is 300 seconds. For larger datasets, increase it: kanoniv.cloud.reconcile(..., timeout=600.0). You can also lower the poll interval: poll_interval=1.0.

"Health status: degraded" - This usually means the merge rate is unusually high or low, or too many records landed in the review queue. Check result.health_flags for specific signals and adjust your thresholds or rules accordingly.

"Column name mismatch" - The Source.from_pandas() column names must match the field names in your spec's schema block. Use .rename(columns={...}) to map the dbt normalized column names to the spec's expected names.

Next Steps

Arrow Data Plane - 6x faster warehouse-to-cloud ingest with Arrow + DuckDB + Parquet
Active Learning - Iteratively improve match quality by labeling uncertain pairs
Tuning Match Quality - Diagnose false positives and false negatives, iterate on precision and recall
Spec Reference - Full reference for every spec option including Fellegi-Sunter probabilistic matching
Kanoniv Cloud - Complete cloud SDK reference with entity lifecycle, audit, and review queue APIs
dbt + Snowflake Tutorial - The local-only version of this pipeline using Source.from_dbt()

Tutorial: Snowflake + dbt + Kanoniv Cloud ​

What You Will Build ​

Prerequisites ​

Step 1: The Raw Data in Snowflake ​

RAW.CRM_CONTACTS - Salesforce CRM ​

RAW.BILLING_CUSTOMERS - Stripe ​

RAW.SUPPORT_TICKETS - Zendesk ​

Step 2: dbt Project Setup ​

Step 3: Define Sources ​

Step 4: Staging Models ​

CRM staging model ​

Billing staging model ​

Support staging model ​

Step 5: Write the Kanoniv Spec ​

Step 6: Reconcile with Kanoniv Cloud ​

Optional: Arrow Fast Path ​

Step 7: Real-Time Resolution with Probabilistic Scoring ​

What Just Happened? ​

1. Extract (Snowflake -> dbt) ​

2. Load (Snowflake -> Python) ​

3. Upload (Python -> Kanoniv Cloud) ​

4. Reconcile (Kanoniv Cloud) ​

5. Write Back (Python -> Snowflake) ​

6. Live Resolution ​

7. Real-Time Probabilistic Resolution ​

Scheduling in Production ​

Airflow example ​

dbt Cloud + webhook ​

Querying the Results in Snowflake ​

Customer 360 view ​

Duplicate report ​

Troubleshooting ​

Next Steps ​

Tutorial: Snowflake + dbt + Kanoniv Cloud

What You Will Build

Prerequisites

Step 1: The Raw Data in Snowflake

`RAW.CRM_CONTACTS` - Salesforce CRM

`RAW.BILLING_CUSTOMERS` - Stripe

`RAW.SUPPORT_TICKETS` - Zendesk

Step 2: dbt Project Setup

Step 3: Define Sources

Step 4: Staging Models

CRM staging model

Billing staging model

Support staging model

Step 5: Write the Kanoniv Spec

Step 6: Reconcile with Kanoniv Cloud

Optional: Arrow Fast Path

Step 7: Real-Time Resolution with Probabilistic Scoring

What Just Happened?

1. Extract (Snowflake -> dbt)

2. Load (Snowflake -> Python)

3. Upload (Python -> Kanoniv Cloud)

4. Reconcile (Kanoniv Cloud)

5. Write Back (Python -> Snowflake)

6. Live Resolution

7. Real-Time Probabilistic Resolution

Scheduling in Production

Airflow example

dbt Cloud + webhook

Querying the Results in Snowflake

Customer 360 view

Duplicate report

Troubleshooting

Next Steps