Building a Customer 360 Identity Graph with Snowflake, dbt, and Kanoniv Cloud
Published February 2026 · 15 min read
You have customer data everywhere. CRM, billing, support -- every team has its own system, its own IDs, and its own version of the truth. Marketing says you have 1.6 million customers. Finance says 600,000. Support says something else entirely.
They are all looking at the same people. They just do not know it yet.
This post walks through a complete, production-ready identity resolution pipeline. We will take raw data from three systems in Snowflake, clean it with dbt, resolve identities with Kanoniv Cloud, and write canonical customer IDs back to the warehouse. By the end, you will have a single identity graph that maps every record across every system to a unified customer.
No prior experience with identity resolution is required. We will explain every concept as it comes up.
The Problem: Same Person, Three Systems, Three Truths
You are an analytics engineer at Meridian Technologies, a B2B SaaS company. Customer data lives in three systems:
- Salesforce (CRM): 850,000 contacts
- Stripe (Billing): 320,000 customers
- Zendesk (Support): 410,000 tickets with requester data
Each system has its own identifier. Salesforce uses sf_id, Stripe uses cus_id, and Zendesk uses requester_id. There is no shared key between them.
Here is what the raw data looks like in Snowflake.
RAW.SALESFORCE_CONTACTS
| sf_id | email_address | full_name | phone | company |
|---|---|---|---|---|
| 003A00012xQ | [email protected] | Jonathan Reeves | +1-415-555-0192 | Meridian Technologies |
| 003A00014kR | [email protected] | Sarah Chen | (628) 555-0341 | DataBright Inc |
| 003A00018vW | [email protected] | Miriam Okafor | 312-555-0287 | Stackline |
RAW.STRIPE_CUSTOMERS
| cus_id | billing_email | customer_name | plan | created_at |
|---|---|---|---|---|
| cus_N4x8Kj | [email protected] | Jon Reeves | enterprise | 2023-06-14 |
| cus_R7m2Pq | [email protected] | Sarah Chen | growth | 2023-09-02 |
| cus_W3k9Lx | [email protected] | M. Okafor | starter | 2024-01-11 |
RAW.ZENDESK_TICKETS
| ticket_id | requester_email | requester_name | subject | priority |
|---|---|---|---|---|
| 482910 | [email protected] | J. Reeves | SSO configuration issue | high |
| 483221 | [email protected] | Sarah C. | API rate limit question | normal |
| 484017 | [email protected] | Miriam Okafor | Billing portal access | urgent |
Look carefully. Jonathan Reeves appears in all three systems -- but with three different names ("Jonathan Reeves", "Jon Reeves", "J. Reeves"), two different email addresses (jonathan.reeves@ and jon.reeves@), and no shared identifier. A simple JOIN ON email would miss the Stripe record entirely.
Sarah Chen is easier -- her emails differ only by username formatting. Miriam Okafor uses m.okafor@ in Salesforce and miriam@ in Stripe, but the same email in Zendesk as Stripe.
Multiply this by 1.58 million records across three systems, and you begin to see the problem.
Why This Is Hard
Identity resolution is the process of determining which records across multiple systems refer to the same real-world entity. It sounds simple. It is not.
Here is what a naive approach misses:
| Challenge | Example |
|---|---|
| Name variations | "Jonathan Reeves" vs "Jon Reeves" vs "J. Reeves" |
| Email aliases | jonathan.reeves@ vs jon.reeves@ (same person, different emails) |
| Missing data | Stripe has no phone number; Zendesk has no company |
| Formatting differences | +1-415-555-0192 vs (415) 555-0192 vs 4155550192 |
| Typos and abbreviations | "DataBright Inc" vs "DataBright" vs "Databright Inc." |
| Scale | 1.58M records means 1.25 trillion possible pairs to compare |
You cannot solve this with SQL JOINs. You need a system that understands fuzzy matching, weighted scoring, and transitive closure (if A=B and B=C, then A=B=C). That system also needs to produce explainable decisions -- not just "these records matched," but why they matched and how confident the system is.
This is what Kanoniv Cloud does. It takes a declarative Spec that encodes your identity logic, runs it against your data, and produces a persistent identity graph where every source record maps to a canonical customer.
The Solution Architecture
Here is the full pipeline we are building:
The pipeline has six stages:
- Raw data lands in Snowflake from Salesforce, Stripe, and Zendesk via your existing EL tool (Fivetran, Airbyte, etc.)
- dbt cleans and standardizes the raw tables into staging models with a common schema
- Kanoniv Cloud reads the staged data, runs identity resolution using your Spec, and builds the identity graph
- The identity graph persists canonical entities with links back to every source record
- Reverse ETL writes
canonical_idback to Snowflake so every downstream query can join on a unified customer ID - Ongoing resolution: new records resolve against the existing graph automatically
Let's build each stage.
Step 1: Model Your Data in dbt
Before identity resolution can work, the data from all three systems needs to share a common shape. Raw Salesforce columns (email_address, full_name) differ from Stripe columns (billing_email, customer_name). dbt staging models handle this normalization.
stg_salesforce_contacts.sql
-- models/staging/stg_salesforce_contacts.sql
with source as (
select * from {{ source('salesforce', 'contacts') }}
),
cleaned as (
select
sf_id as contact_id,
lower(trim(email_address)) as email,
trim(full_name) as name,
regexp_replace(phone, '[^0-9+]', '') as phone,
trim(company) as company,
'salesforce' as source_system,
current_timestamp() as _loaded_at
from source
where email_address is not null
and trim(email_address) != ''
)
select * from cleanedstg_stripe_customers.sql
-- models/staging/stg_stripe_customers.sql
with source as (
select * from {{ source('stripe', 'customers') }}
),
cleaned as (
select
cus_id as contact_id,
lower(trim(billing_email)) as email,
trim(customer_name) as name,
null as phone, -- Stripe has no phone
null as company,
'stripe' as source_system,
current_timestamp() as _loaded_at
from source
where billing_email is not null
and trim(billing_email) != ''
)
select * from cleanedstg_zendesk_requesters.sql
-- models/staging/stg_zendesk_requesters.sql
with source as (
select * from {{ source('zendesk', 'tickets') }}
),
-- Zendesk has one row per ticket, but we want unique requesters
deduplicated as (
select
requester_email,
requester_name,
row_number() over (
partition by lower(trim(requester_email))
order by created_at desc
) as rn
from source
where requester_email is not null
),
cleaned as (
select
md5(lower(trim(requester_email))) as contact_id,
lower(trim(requester_email)) as email,
trim(requester_name) as name,
null as phone,
null as company,
'zendesk' as source_system,
current_timestamp() as _loaded_at
from deduplicated
where rn = 1
)
select * from cleanedAfter running dbt run, you have three clean staging tables in Snowflake -- all with the same columns (contact_id, email, name, phone, company, source_system). The field names are normalized, emails are lowercased, phone numbers are stripped of formatting, and Zendesk tickets are deduplicated to unique requesters.
This is the data that Kanoniv will consume.
Step 2: Define Your Identity Spec
A Spec is a YAML file that declares your entire identity resolution configuration. It tells Kanoniv what entity you are resolving, where the data comes from, how records should be compared, when to merge, and which source to trust when values conflict.
The spec is the product. No imperative code, no pipeline wiring. You describe what identity means, and the engine figures out how to resolve it.
Here is the full spec for our Customer 360:
# specs/customer-360.yaml
api_version: kanoniv/v2
identity_version: customer_360_v1
entity:
name: customer
description: "Unified B2B customer across CRM, billing, and support"
sources:
salesforce:
adapter: snowflake
location: ANALYTICS.STAGING.STG_SALESFORCE_CONTACTS
primary_key: contact_id
schema:
email: { type: string, pii: true }
name: { type: string }
phone: { type: string, pii: true }
company: { type: string }
attributes:
email: email
name: name
phone: phone
company: company
stripe:
adapter: snowflake
location: ANALYTICS.STAGING.STG_STRIPE_CUSTOMERS
primary_key: contact_id
schema:
email: { type: string, pii: true }
name: { type: string }
attributes:
email: email
name: name
zendesk:
adapter: snowflake
location: ANALYTICS.STAGING.STG_ZENDESK_REQUESTERS
primary_key: contact_id
schema:
email: { type: string, pii: true }
name: { type: string }
attributes:
email: email
name: name
rules:
- name: email_exact
type: exact
field: email
weight: 1.0
- name: name_fuzzy
type: similarity
field: name
algorithm: jaro_winkler
threshold: 0.88
weight: 0.6
- name: phone_exact
type: exact
field: phone
weight: 0.8
- name: company_fuzzy
type: similarity
field: company
algorithm: jaro_winkler
threshold: 0.90
weight: 0.4
decision:
scoring: weighted_sum
thresholds:
match: 0.85
review: 0.65
survivorship:
default: source_priority
overrides:
- field: name
strategy: source_priority
priority: [salesforce, stripe, zendesk]
- field: email
strategy: source_priority
priority: [salesforce, stripe, zendesk]
- field: phone
strategy: source_priority
priority: [salesforce, zendesk, stripe]
- field: company
strategy: source_priority
priority: [salesforce, stripe, zendesk]Let's break this down section by section.
Entity
The Entity block declares what you are resolving. In our case, it is a customer -- a real-world person or company that might appear across multiple systems. The entity name appears in API responses, audit logs, and the SDK's result objects.
Sources
Sources define where data lives. Each source specifies an adapter type (snowflake), a location (the fully-qualified table name), and a primary_key that uniquely identifies rows. The attributes map creates the bridge between your source columns and the canonical field names used in rules. Since our dbt models already normalized column names, the attributes are simple one-to-one mappings. In real-world specs, this is where you would map billing_email to email or full_name to name. See the Source Adapters reference for all available adapter types.
Rules
Rules define how records are compared. Each rule specifies a type, the fields to compare, and a weight. We have four rules:
email_exact(weight 1.0): Exact email match after normalization. This is the strongest signal -- if two records share the same email, they are almost certainly the same person.name_fuzzy(weight 0.6): Fuzzy name match using Jaro-Winkler similarity with a 0.88 threshold. This catches "Jonathan Reeves" vs "Jon Reeves" (Jaro-Winkler score: ~0.91) while rejecting "Jonathan Reeves" vs "Jonathan Rivera" (score: ~0.82). See Matching Rules for algorithm details.phone_exact(weight 0.8): Exact phone match after stripping formatting. Phones are strong identifiers but sometimes get recycled, so the weight is slightly below email.company_fuzzy(weight 0.4): Fuzzy company name match. Useful as supporting evidence ("DataBright Inc" vs "DataBright") but not strong enough to trigger a merge on its own.
Decision
The Decision block controls the scoring method and the thresholds that turn scores into outcomes:
match: 0.85-- Pairs scoring 0.85 or above are auto-merged. This means an exact email match alone (normalized score: 1.0/1.0 when email is the only rule that fires on shared weights) is enough to merge.review: 0.65-- Pairs scoring between 0.65 and 0.85 enter a human review queue. A fuzzy name match plus a fuzzy company match might land here.- Anything below 0.65 is rejected automatically.
The scoring method is weighted_sum (the default). For each pair, every rule that evaluates produces a score contribution of weight * match_score. The final score is the sum divided by the sum of all applicable weights.
Survivorship
Survivorship determines which value wins when merged records disagree. We use source_priority for every field, with Salesforce as the highest-priority source. This means if Salesforce has a name, that name wins -- regardless of what Stripe or Zendesk say. If Salesforce's value is null, it falls through to Stripe, then Zendesk.
This reflects a real-world judgment: the CRM is the system of record for customer data at most companies.
Step 3: Connect Snowflake to Kanoniv
Now we wire the data into Kanoniv. This is where Source Adapters come in.
A Source Adapter is the bridge between your data and the Kanoniv engine. Instead of exporting CSVs or writing custom ETL, you tell Kanoniv where your data lives and it reads directly from the source. The SDK ships with adapters for the most common data environments:
| Adapter | Method | Best For |
|---|---|---|
| Warehouse | Source.from_warehouse() | Snowflake, BigQuery, Postgres, any SQLAlchemy-compatible DB |
| dbt | Source.from_dbt() | dbt models (resolves via manifest.json) |
| CSV | Source.from_csv() | Local files, quick prototyping |
| pandas | Source.from_pandas() | DataFrames in notebooks or scripts |
| JSON | Source.from_json() | JSON array files |
All adapters stream data in batches (the warehouse adapter uses 5,000-row batches by default), so they never load an entire table into memory. Connections are lazy — they are not opened until data is actually read.
Since our staging models are in dbt and materialized in Snowflake, we have two options:
Option A: Read directly from Snowflake
Use Source.from_warehouse() to read from the materialized staging tables:
from kanoniv import Source
SNOWFLAKE_CONN = "snowflake://analytics_user:password@meridian-tech/ANALYTICS/STAGING"
salesforce = Source.from_warehouse(
"salesforce",
"ANALYTICS.STAGING.STG_SALESFORCE_CONTACTS",
connection_string=SNOWFLAKE_CONN,
primary_key="contact_id",
)
stripe = Source.from_warehouse(
"stripe",
"ANALYTICS.STAGING.STG_STRIPE_CUSTOMERS",
connection_string=SNOWFLAKE_CONN,
primary_key="contact_id",
)
zendesk = Source.from_warehouse(
"zendesk",
"ANALYTICS.STAGING.STG_ZENDESK_REQUESTERS",
connection_string=SNOWFLAKE_CONN,
primary_key="contact_id",
)
sources = [salesforce, stripe, zendesk]The warehouse adapter streams rows in batches of 5,000 by default, so it never loads the entire table into memory. The connection is lazy -- it is not opened until data is actually read.
Option B: Read from dbt models
If you prefer to reference dbt models by name (and let the adapter resolve the compiled SQL via manifest.json), use Source.from_dbt():
from kanoniv import Source
SNOWFLAKE_CONN = "snowflake://analytics_user:password@meridian-tech/ANALYTICS/STAGING"
salesforce = Source.from_dbt(
"salesforce",
model="stg_salesforce_contacts",
manifest_path="target/manifest.json",
connection_string=SNOWFLAKE_CONN,
primary_key="contact_id",
)
stripe = Source.from_dbt(
"stripe",
model="stg_stripe_customers",
manifest_path="target/manifest.json",
connection_string=SNOWFLAKE_CONN,
primary_key="contact_id",
)
zendesk = Source.from_dbt(
"zendesk",
model="stg_zendesk_requesters",
manifest_path="target/manifest.json",
connection_string=SNOWFLAKE_CONN,
primary_key="contact_id",
)
sources = [salesforce, stripe, zendesk]The dbt adapter resolves the model name through manifest.json to database.schema.alias, then reads from the warehouse. It accepts both "stg_customers" and ref('stg_customers') syntax. See Source Adapters for the full adapter reference.
Both options produce the same result. Option A is simpler; Option B is better if your dbt project uses custom schemas or aliases.
Step 4: Run Cloud Reconciliation
With sources connected and the spec defined, we run Cloud Reconciliation. One function call handles everything: spec upload, data upload, job submission, polling, and result extraction.
import kanoniv
from kanoniv import Source, Spec
# Load the spec
spec = Spec.from_file("specs/customer-360.yaml")
# Validate locally first (catches errors before hitting the API)
from kanoniv import validate
result = validate(spec)
result.raise_on_error() # raises ValueError if invalid
# Connect sources (from Step 3)
SNOWFLAKE_CONN = "snowflake://analytics_user:password@meridian-tech/ANALYTICS/STAGING"
sources = [
Source.from_warehouse("salesforce", "ANALYTICS.STAGING.STG_SALESFORCE_CONTACTS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
Source.from_warehouse("stripe", "ANALYTICS.STAGING.STG_STRIPE_CUSTOMERS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
Source.from_warehouse("zendesk", "ANALYTICS.STAGING.STG_ZENDESK_REQUESTERS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
]
# Run cloud reconciliation
with kanoniv.cloud.reconcile(
sources,
spec,
api_key="kn_live_a1b2c3d4e5f6...",
timeout=600.0, # large dataset, allow 10 minutes
poll_interval=5.0, # check every 5 seconds
) as result:
print(result.summary())Under the hood, kanoniv.cloud.reconcile() does five things automatically:
- Creates a client from your API key
- Uploads and compiles the spec to the Kanoniv server
- Maps attributes and collects entities from each source
- Uploads entities in batches (500 records per request)
- Submits the reconciliation job and polls until completion
After a few minutes, the output looks like this:
Cloud Reconciliation — job d4e5f6a1-b2c3-4d5e-f6a1-b2c3d4e5f6a1
Status: completed
Duration: 187400ms
Health: healthy
Input entities: 1,580,000
Canonicals: 612,540
Links: 1,580,000
Merge rate: 61.3%
Match quality: {"accepted_matches": 489200, "rejected_matches": 12840}
Health flags: noneLet's interpret these numbers:
- 1,580,000 input entities -- every record from all three staging tables
- 612,540 canonical customers -- the deduplicated count of real-world people
- 61.3% merge rate -- 61.3% of input records were merged with at least one other record. The remaining 38.7% are singletons (appearing in only one system with no match elsewhere)
- Healthy -- no anomalies detected. A "degraded" or "unhealthy" status would indicate potential problems (unusually low match rates, oversized clusters, etc.)
That 61.3% merge rate means your company does not have 1.58 million customers. It has 612,540. Marketing, Finance, and Support can now agree on the same number, because they are looking at the same identity graph.
Step 5: Explore the Identity Graph
The identity graph is now live and queryable. Every source record maps to a canonical customer via a persistent link. You can resolve, search, and inspect entities at any time through the Python SDK.
Resolve a Source Record
Given a Salesforce contact ID, find its canonical identity:
from kanoniv import Client
with Client(api_key="kn_live_a1b2c3d4e5f6...") as client:
# Resolve Jonathan Reeves from Salesforce
result = client.resolve(system="salesforce", external_id="003A00012xQ")
print(result["canonical_id"])
# "c7e2f9a1-4b3d-4e5f-a6b7-c8d9e0f1a2b3"See All Linked Records
Once you have a canonical ID, you can see every source record linked to that identity:
linked = client.entities.get_linked(result["canonical_id"])
print(f"Canonical: {result['canonical_id']}")
for record in linked["linked"]:
print(f" {record['source_name']}: {record['external_id']}")
# salesforce: 003A00012xQ
# stripe: cus_N4x8Kj
# zendesk: a1b2c3d4e5f6 (md5 of email)Jonathan Reeves -- who appeared as "Jonathan Reeves" in Salesforce, "Jon Reeves" in Stripe, and "J. Reeves" in Zendesk -- is now one canonical customer linked to all three records.
Search the Graph
Find entities by email, name, or any indexed field:
results = client.entities.search(q="[email protected]")
for entity in results["data"]:
print(f"{entity['id']}: {entity['data']['name']} <{entity['data']['email']}>")
# "f3a4b5c6-...": Sarah Chen <[email protected]>Notice the golden record has [email protected] -- the Salesforce email, not the Stripe email -- because our survivorship rules prioritize Salesforce as the source of truth.
Before and After
Here is what identity resolution did to our data:
Three disconnected records with different names, different emails, and no shared key -- resolved into a single canonical customer. The golden record carries Salesforce's values (because of our survivorship priority), and every source record links back to the canonical ID.
Step 6: Reverse ETL -- Write Canonical IDs Back to Snowflake
The identity graph is powerful on its own, but the real value comes when canonical IDs flow back into your warehouse. Once every raw record has a canonical_id, you can join across systems in a single SQL query.
Write the Mapping Table
import kanoniv
from kanoniv import Source, Spec, Client
from sqlalchemy import create_engine, text
spec = Spec.from_file("specs/customer-360.yaml")
SNOWFLAKE_CONN = "snowflake://analytics_user:password@meridian-tech/ANALYTICS/STAGING"
sources = [
Source.from_warehouse("salesforce", "ANALYTICS.STAGING.STG_SALESFORCE_CONTACTS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
Source.from_warehouse("stripe", "ANALYTICS.STAGING.STG_STRIPE_CUSTOMERS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
Source.from_warehouse("zendesk", "ANALYTICS.STAGING.STG_ZENDESK_REQUESTERS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
]
# 1. Reconcile
with kanoniv.cloud.reconcile(sources, spec, api_key="kn_live_...") as result:
df = result.to_pandas()
# 2. Write canonical mappings back to Snowflake
engine = create_engine(SNOWFLAKE_CONN)
with engine.begin() as conn:
conn.execute(text("""
CREATE TABLE IF NOT EXISTS ANALYTICS.IDENTITY.CANONICAL_MAPPINGS (
source_name VARCHAR(100),
external_id VARCHAR(255),
canonical_id VARCHAR(36),
updated_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
"""))
conn.execute(text("TRUNCATE TABLE ANALYTICS.IDENTITY.CANONICAL_MAPPINGS"))
mapping_df = df[["source_name", "external_id", "id"]].rename(columns={"id": "canonical_id"})
mapping_df.to_sql(
"canonical_mappings",
engine,
schema="IDENTITY",
if_exists="append",
index=False,
)
print(f"Wrote {len(mapping_df):,} canonical mappings to Snowflake")
# Wrote 1,580,000 canonical mappings to SnowflakeQuery Across Systems
Now any analyst can join across all three systems using a single canonical_id:
-- How many unified customers are on the Enterprise plan in Stripe
-- AND have open high-priority tickets in Zendesk?
SELECT
cm_sf.canonical_id,
sf.full_name,
st.plan,
COUNT(zd.ticket_id) AS open_tickets
FROM analytics.identity.canonical_mappings cm_sf
JOIN raw.salesforce_contacts sf
ON cm_sf.external_id = sf.sf_id
AND cm_sf.source_name = 'salesforce'
JOIN analytics.identity.canonical_mappings cm_st
ON cm_sf.canonical_id = cm_st.canonical_id
AND cm_st.source_name = 'stripe'
JOIN raw.stripe_customers st
ON cm_st.external_id = st.cus_id
JOIN analytics.identity.canonical_mappings cm_zd
ON cm_sf.canonical_id = cm_zd.canonical_id
AND cm_zd.source_name = 'zendesk'
JOIN raw.zendesk_tickets zd
ON cm_zd.external_id = zd.requester_email
AND zd.priority = 'high'
AND zd.status = 'open'
WHERE st.plan = 'enterprise'
GROUP BY 1, 2, 3
ORDER BY open_tickets DESC;Before identity resolution, this query was impossible. Salesforce, Stripe, and Zendesk had no shared key. Now they do: canonical_id.
Create a dbt Model for It
Wrap the mapping in a dbt model so it stays in your transformation graph:
-- models/identity/int_canonical_mappings.sql
{{ config(materialized='table', schema='identity') }}
-- This table is populated by the Kanoniv reverse ETL script.
-- It maps every source record to its canonical customer ID.
select
source_name,
external_id,
canonical_id,
updated_at
from {{ source('identity', 'canonical_mappings') }}Step 7: Ongoing Resolution
Identity resolution is not a one-time job. New customers sign up in Stripe. Sales reps create contacts in Salesforce. Support tickets arrive in Zendesk. Each new record needs to resolve against the existing identity graph.
Scheduled Batch Resolution
Run reconciliation on a schedule (daily, hourly, or triggered by dbt job completion):
# resolve_daily.py — triggered by Airflow, Dagster, or cron
import kanoniv
from kanoniv import Source, Spec
spec = Spec.from_file("specs/customer-360.yaml")
SNOWFLAKE_CONN = "snowflake://analytics_user:password@meridian-tech/ANALYTICS/STAGING"
sources = [
Source.from_warehouse("salesforce", "ANALYTICS.STAGING.STG_SALESFORCE_CONTACTS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
Source.from_warehouse("stripe", "ANALYTICS.STAGING.STG_STRIPE_CUSTOMERS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
Source.from_warehouse("zendesk", "ANALYTICS.STAGING.STG_ZENDESK_REQUESTERS",
connection_string=SNOWFLAKE_CONN, primary_key="contact_id"),
]
with kanoniv.cloud.reconcile(sources, spec, api_key="kn_live_...") as result:
print(result.summary())
if result.health_status != "healthy":
# Alert on Slack, PagerDuty, etc.
print(f"WARNING: Health status is {result.health_status}")
print(f"Flags: {result.health_flags}")
# Write updated mappings back to Snowflake
df = result.to_pandas()
mapping_df = df[["source_name", "external_id", "id"]].rename(columns={"id": "canonical_id"})
# ... write to Snowflake (same as Step 6)Real-Time Resolution
For systems that need instant identity resolution (e.g., a customer logs in and you need their canonical ID immediately), use the client.resolve() API. See Real-World Scenarios for patterns including webhook endpoints, async batch resolution, and human-in-the-loop review queues.
from kanoniv import Client
with Client(api_key="kn_live_...") as client:
# New Stripe customer just signed up — resolve immediately
result = client.resolve(system="stripe", external_id="cus_NEW123")
if result.get("canonical_id"):
print(f"Existing customer: {result['canonical_id']}")
# They already exist in the graph — enrich the record
linked = client.entities.get_linked(result["canonical_id"])
sf_record = next(
(r for r in linked["linked"] if r["source_name"] == "salesforce"), None
)
if sf_record:
print(f"Salesforce contact: {sf_record['external_id']}")
else:
print("New customer — no prior records found")Every reconciliation run updates the graph incrementally. New records are matched against existing canonical entities, not just against other new records. The graph grows over time, and the merge rate typically improves as more signals accumulate.
What You Built
In this tutorial, you built a production-grade Customer 360 pipeline:
- dbt staging models that clean and normalize data from Salesforce, Stripe, and Zendesk
- An identity Spec with four matching rules, weighted scoring, and source-priority survivorship
- Cloud reconciliation that resolved 1.58M records into 612K canonical customers
- An identity graph queryable via the Python SDK -- resolve, search, and inspect any entity
- Reverse ETL that writes
canonical_idback to Snowflake for cross-system SQL joins - Ongoing resolution that keeps the graph current as new data arrives
What's Next
- Spec Reference -- Deep dive into every spec section: entity, sources, rules, decision, survivorship, and governance
- Real-World Scenarios -- Warehouse-to-Kanoniv, real-time webhooks, review queues, async batch processing
- API Reference -- Full reference for every client method: entities, sources, rules, jobs, reviews, overrides, audit
- CLI Reference -- Validate, plan, diff, and reconcile specs from the command line
- Matching Rules -- Algorithm reference for exact, Jaro-Winkler, Levenshtein, Soundex, and more
- Identity Graph -- How the graph works: canonical entities, links, transitive closure, and cluster management
Get started with pip install kanoniv[cloud]
