First Reconciliation

Run your first identity reconciliation entirely offline — no server, no database, just Python and two CSV files.

Prerequisites

• Python 3.9+
• pip install kanoniv

Step 1: Create Sample Data

Create two CSV files that represent the same customers in different systems.

crm.csv — your CRM contacts:

id,name,email,phone,company
1,John Doe,[email protected],555-0101,Acme Corp
2,Jane Smith,[email protected],555-0102,Globex Inc
3,Bob Wilson,[email protected],555-0103,Acme Corp
4,Alice Brown,[email protected],555-0104,Startup LLC
5,Charlie Davis,[email protected],,BigCo

billing.csv — your Stripe billing records:

customer_id,full_name,email,card_last4,plan
cus_001,Jonathan Doe,[email protected],4242,enterprise
cus_002,Jane Smith,[email protected],1234,pro
cus_003,Robert Wilson,[email protected],5678,starter
cus_005,Charlie D.,[email protected],9999,enterprise
cus_006,Eve Martinez,[email protected],3333,pro

Notice the overlaps: "John Doe" / "Jonathan Doe", "Bob Wilson" / "Robert Wilson", and "Charlie Davis" / "Charlie D." all share email addresses.

Step 2: Write an Identity Schema

Create my-first-schema.yaml:

api_version: kanoniv/v1
identity_version: "1.0"

entity:
  name: customer

sources:
  crm:
    adapter: csv
    location: crm.csv
    primary_key: id
    schema:
      name: { type: string }
      email: { type: string, pii: true }
      phone: { type: string, pii: true }
      company: { type: string }

  billing:
    adapter: csv
    location: billing.csv
    primary_key: customer_id
    schema:
      full_name: { type: string }
      email: { type: string, pii: true }
      card_last4: { type: string }
      plan: { type: string }

blocking:
  keys:
    - [email]

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

decision:
  thresholds:
    match: 0.9
    review: 0.5

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

This schema says:

• Two sources: CRM and billing, both CSVs with declared schemas
• Blocking: group candidate pairs by email (records must share an email to be compared)
• One rule: exact email match with full weight
• Decision: auto-match above 0.9, review between 0.5-0.9
• Survivorship: prefer CRM data over billing data

Step 3: Validate

from kanoniv import Spec, validate

spec = Spec.from_file("my-first-schema.yaml")
result = validate(spec)

if result:
    print("Schema is valid!")
else:
    for error in result.errors:
        print(f"  {error}")
    raise SystemExit(1)

Expected output:

Schema is valid!

Step 4: Preview the Plan

from kanoniv import plan

execution_plan = plan(spec)
print(execution_plan.summary())

Expected output:

Execution Plan for: customer
  Hash: sha256:a1b2c3...
  Stages:
    1. normalize   — Standardize fields across 2 sources
    2. block       — Group candidates by [email]
    3. match       — Apply 1 rule (email_exact)
    4. decide      — Thresholds: match=0.9, review=0.5
    5. merge       — Create canonical entities
    6. survive     — Strategy: source_priority [crm, billing]

Step 5: Reconcile

from kanoniv import Source, reconcile

sources = [
    Source.from_csv("crm", "crm.csv"),
    Source.from_csv("billing", "billing.csv"),
]

result = reconcile(sources, spec)

Step 6: Inspect Results

Clusters

Clusters show which record IDs were grouped together:

for i, cluster in enumerate(result.clusters):
    print(f"Cluster {i + 1}: {cluster}")

Expected output:

Cluster 1: ['crm:1', 'billing:cus_001']
Cluster 2: ['crm:2', 'billing:cus_002']
Cluster 3: ['crm:3', 'billing:cus_003']
Cluster 4: ['crm:5', 'billing:cus_005']
Cluster 5: ['crm:4']
Cluster 6: ['billing:cus_006']

Records with matching emails were grouped into clusters. Alice Brown (crm:4) and Eve Martinez (billing:cus_006) had no matches, so they form their own single-record clusters.

Golden Records

Golden records are the merged canonical entities with survivorship applied:

for record in result.golden_records:
    print(record)

Because we set source_priority: [crm, billing], the CRM's name field wins over billing's full_name:

{
    'name': 'John Doe',
    'email': '[email protected]',
    'phone': '555-0101',
    'company': 'Acme Corp',
    'card_last4': '4242',
    'plan': 'enterprise'
}

Fields that only exist in one source (like card_last4 from billing, or phone from CRM) are preserved regardless of priority.

Match Decisions

Decisions show why records were matched:

for decision in result.decisions:
    print(decision)

{
    'left': 'crm:1',
    'right': 'billing:cus_001',
    'score': 1.0,
    'outcome': 'match',
    'rules': [{'name': 'email_exact', 'score': 1.0, 'field': 'email'}]
}

Every merge has a reason code. This is what makes Kanoniv deterministic and auditable.

Summary Statistics

print(f"Clusters: {result.cluster_count}")
print(f"Golden records: {len(result.golden_records)}")
print(f"Merge rate: {result.merge_rate:.1%}")

Clusters: 6
Golden records: 6
Merge rate: 40.0%

4 out of 10 input records were merged into pairs, giving a 40% merge rate. The remaining 2 unmatched records (Alice Brown and Eve Martinez) become their own golden records.

Export Results

Golden records are plain Python dicts — write them wherever your data lives.

Snowflake:

import snowflake.connector

conn = snowflake.connector.connect(
    account="xy12345.us-east-1",
    user="KANONIV_SVC",
    password="...",
    database="ANALYTICS",
    schema="IDENTITY",
)

cursor = conn.cursor()
for record in result.golden_records:
    cursor.execute(
        "INSERT INTO golden_customers (name, email, phone, company, plan) "
        "VALUES (%(name)s, %(email)s, %(phone)s, %(company)s, %(plan)s)",
        record,
    )

BigQuery:

from google.cloud import bigquery

client = bigquery.Client()
client.insert_rows_json("analytics.identity.golden_customers", result.golden_records)

PostgreSQL:

import psycopg2

conn = psycopg2.connect("postgresql://localhost/analytics")
cursor = conn.cursor()
for record in result.golden_records:
    cursor.execute(
        "INSERT INTO golden_customers (name, email, phone) VALUES (%(name)s, %(email)s, %(phone)s)",
        record,
    )
conn.commit()

pandas (for notebooks or further analysis):

df = result.to_pandas()
df.to_csv("golden_customers.csv", index=False)

Golden records are your data — write them to any warehouse, database, file, or reverse ETL pipeline.

What Happened

1. Normalize — Fields were standardized across both sources
2. Block — Records were grouped by email address to reduce comparisons
3. Match — Records with identical emails scored 1.0 (above the 0.9 match threshold)
4. Merge — Matched records were combined into canonical entities
5. Survive — CRM fields took priority; billing-only fields (card_last4, plan) were kept

Next Steps

• Add fuzzy matching - Catch "John Doe" / "Jon Doe" variations
• Configure survivorship - Use most_recent or most_complete strategies
• Core Concepts - Understand entities, identity graphs, and matching
• Identity Schema Configuration - Every configuration option explained
• Customer 360 Tutorial - Build a complete 3-source reconciliation