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:

1. Raw data lands in Snowflake from Salesforce, Stripe, and Zendesk via your existing EL tool (Fivetran, Airbyte, etc.)
2. dbt cleans and standardizes the raw tables into staging models with a common schema
3. Kanoniv Cloud reads the staged data, runs identity resolution using your Spec, and builds the identity graph
4. The identity graph persists canonical entities with links back to every source record
5. Reverse ETL writes canonical_id back to Snowflake so every downstream query can join on a unified customer ID
6. 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 cleaned

stg_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 cleaned

stg_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 cleaned

After 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:

1. Creates a client from your API key
2. Uploads and compiles the spec to the Kanoniv server
3. Maps attributes and collects entities from each source
4. Uploads entities in batches (500 records per request)
5. 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:    none

Let'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 Snowflake

Query 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:

1. dbt staging models that clean and normalize data from Salesforce, Stripe, and Zendesk
2. An identity Spec with four matching rules, weighted scoring, and source-priority survivorship
3. Cloud reconciliation that resolved 1.58M records into 612K canonical customers
4. An identity graph queryable via the Python SDK -- resolve, search, and inspect any entity
5. Reverse ETL that writes canonical_id back to Snowflake for cross-system SQL joins
6. 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]