Sources

Sources define where your data lives. Each source specifies an adapter type, a location, a primary key, and optionally attributes, freshness constraints, and a schema declaration. The Python SDK loads data from these sources and feeds records into the matching engine.

A spec can contain up to 10 sources. Each source must have a unique name within the spec.

Adapter-First Format

The modern source format uses three required fields (adapter, location, and primary_key) to describe where data comes from and how to identify records within it.

Sources can be written in map-style (recommended) or array-style YAML:

# Map-style — the key becomes the source name
sources:
  crm:
    adapter: csv
    location: data/contacts.csv
    primary_key: id

# Array-style — name is an explicit field
sources:
  - name: crm
    adapter: csv
    location: data/contacts.csv
    primary_key: id

Both formats are equivalent. Map-style is more concise; array-style is the original format and remains fully supported.

Field	Type	Required	Description
name	string	Yes	Unique source identifier (the YAML map key, or explicit name field in array-style). Must be alphanumeric plus underscores.
adapter	string	Yes	Adapter type: csv, json, pandas, postgres, snowflake, bigquery, redshift, dbt
location	string	Yes	Adapter-specific data location (file path, connection string, reference)
primary_key	string	Yes	Column that uniquely identifies each record
attributes	map[string, string]	No	Mapping of canonical field names to source column names (see Attributes)
freshness	object	No	Freshness constraints (Cloud tier)
schema	object	No	Column type, PII, and nullability declarations
tags	array[string]	No	Arbitrary labels for filtering and governance
sampling	object	No	Row sampling configuration for testing
temporal	object	No	Temporal matching configuration (see Temporal)

Adapters

The adapter field declares what kind of data source this is. It serves as metadata — the adapter type is recorded in the compiled plan, audit logs, and API responses, and is used for validation (e.g., warehouse adapters require a location).

How Data Loading Works

The reconciliation engine does not connect to databases or warehouses directly. Data is loaded by the Python SDK using Source.from_csv(), Source.from_pandas(), Source.from_warehouse(), or Source.from_dbt(). The adapter and location fields in the spec describe where the data comes from — the SDK is responsible for actually reading it.

Source.from_warehouse() uses SQLAlchemy under the hood, so it works with any database that has a SQLAlchemy dialect: PostgreSQL, Snowflake, BigQuery, Redshift, MySQL, and more. You don't need a different adapter for each warehouse — one from_warehouse() call handles them all.

See Source Adapters for the full SDK adapter reference.

The location field documents where the data lives. Its format varies by adapter type:

CSV

Local CSV file.

sources:
  crm:
    adapter: csv
    location: data/contacts.csv       # relative or absolute file path
    primary_key: id
    attributes:
      name: full_name
      email: email_address
      phone: phone_number

• location: Path to the CSV file, relative to the working directory or absolute.

SDK: Source.from_csv("crm", "data/contacts.csv", primary_key="id")

Pandas (DataFrame)

In-memory Pandas DataFrame. Used when data is already loaded in a Python session.

sources:
  analytics:
    adapter: pandas
    location: dataframe://df_customers
    primary_key: user_id

• location: A dataframe:// URI where the path is a descriptive label for the DataFrame.

SDK: Source.from_pandas("analytics", df, primary_key="user_id")

PostgreSQL

PostgreSQL table or view.

sources:
  production_db:
    adapter: postgres
    location: public.contacts
    primary_key: id
    attributes:
      name: full_name
      email: email
      phone: phone

• location: Table reference (e.g., public.contacts).

SDK: Source.from_warehouse("production_db", "public.contacts", connection_string="postgresql://user:pass@host:5432/db")

Snowflake

Snowflake table or view.

sources:
  warehouse:
    adapter: snowflake
    location: ANALYTICS.CORE.CUSTOMERS
    primary_key: customer_id

• location: Fully qualified table reference (database.schema.table).

SDK: Source.from_warehouse("warehouse", "ANALYTICS.CORE.CUSTOMERS", connection_string="snowflake://user:pass@account/ANALYTICS")

dbt

A dbt model or source, resolved via the dbt manifest.

sources:
  dbt_customers:
    adapter: dbt
    location: ref('stg_customers')
    primary_key: customer_id

• location: A dbt ref('model_name') or source('source_name', 'table_name') expression.

SDK: Source.from_dbt("dbt_customers", model="stg_customers", connection_string="snowflake://...")

BigQuery

Google BigQuery table or view.

sources:
  warehouse:
    adapter: bigquery
    location: project.dataset.customers
    primary_key: customer_id

• location: GCP project.dataset.table reference.

SDK: Source.from_warehouse("warehouse", "project.dataset.customers", connection_string="bigquery://project-id/dataset")

Redshift

Amazon Redshift table or view.

sources:
  analytics:
    adapter: redshift
    location: public.customers
    primary_key: customer_id

• location: Table reference within the Redshift cluster.

SDK: Source.from_warehouse("analytics", "public.customers", connection_string="redshift://user:[email protected]:5439/db")

Primary Key

The primary_key field identifies the column that uniquely identifies each record within a source. It is the anchor for all downstream operations: matching, merge tracking, audit trails, and golden record lineage.

primary_key: id

Requirements:

• Must reference an actual column name in the source data.
• Values must be non-null. Records with a null primary key are dropped and logged as warnings.
• Values must be unique within the source. Duplicate primary keys cause a validation error at reconciliation time.
• The primary key is a single field name (compound keys are not currently supported).

The primary key is not used as a matching field unless you explicitly include it in a matching rule. Its purpose is record identification, not record comparison.

Attributes

The attributes field maps canonical field names to source column names. This lets you unify differently-named columns across sources into a common namespace for matching and survivorship.

attributes:
  email: email_address      # canonical_name: source_column
  phone: phone_number
  name: full_name

The keys are canonical names (used in rules and survivorship), and the values are the actual column names in the source data. When the canonical name and source column are the same, you still write both:

attributes:
  email: email              # source column is also called "email"
  name: display_name        # source column differs

When omitted, all columns from the source are included with their original names as canonical names. This is convenient for small datasets or early prototyping, but can introduce noise from irrelevant columns and makes cross-source field alignment implicit.

When specified, only the mapped columns are loaded from the source. This has several benefits:

• Cross-source alignment: Different sources can map different column names to the same canonical field, so matching rules work uniformly across all sources.
• Performance: Fewer columns means less data to read, normalize, and compare.
• Clarity: The spec explicitly documents which fields matter for identity resolution.
• Safety: Sensitive columns that should not participate in matching (e.g., ssn, salary) are excluded by default.

TIP

Start without attributes to explore your data, then add explicit mappings once you know which fields matter and how columns differ across sources.

Align Name Fields Across Sources

If one source has separate first_name / last_name columns and another has a combined name column, map them to the same canonical field that your matching rules reference. Otherwise the rule compares a populated field on one side against an empty field on the other — scoring 0.0 every time.

# Bad — rules on "first_name" will never match Stripe records
sources:
  salesforce:
    attributes:
      first_name: first_name   # canonical: first_name
      last_name: last_name
  stripe:
    attributes:
      name: name               # canonical: name (different!)

# Good — both sources feed into the same canonical field
sources:
  salesforce:
    attributes:
      first_name: first_name
      last_name: last_name
  stripe:
    attributes:
      first_name: name         # maps combined "name" → canonical "first_name"

Jaro-Winkler handles partial matches well — comparing "Kevin" against "Kevin Patel" scores ~0.87, which clears most thresholds. The important thing is that both sides have a value to compare.

Freshness

Cloud Feature

Freshness constraints require the Cloud tier.

Freshness constraints ensure your sources contain recent data. Stale data degrades match quality. A CRM that stopped syncing last week will produce outdated golden records.

freshness:
  max_age: "24h"           # reject data older than 24 hours
  warn_after: "12h"        # log a warning after 12 hours

You can also use the _hours aliases for convenience:

freshness:
  max_age_hours: 24        # equivalent to max_age: "24h"
  warn_after_hours: 12     # equivalent to warn_after: "12h"

If both max_age and max_age_hours are present, max_age takes precedence. The duration string format (max_age) supports values like "24h", "30m", "7d".

Field	Type	Required	Description
max_age	string	Yes*	Maximum acceptable age as a duration string (e.g., "24h").
max_age_hours	integer	Yes*	Maximum acceptable age in hours. Alias for max_age.
warn_after	string	No	Warning threshold as a duration string (e.g., "12h").
warn_after_hours	integer	No	Warning threshold in hours. Alias for warn_after.

* One of max_age or max_age_hours is required.

Freshness is evaluated at the start of each reconciliation run by checking the most recent data timestamp for the source.

Freshness integrates with Governance (require_freshness: true makes all freshness checks mandatory) and with SIEM Integration for automated alerting. See Source Freshness Monitoring for the full reference.

Schema Declaration

The schema block lets you declare expected field names, data types, PII sensitivity, and nullability for each source. At runtime, this configuration enables field-presence drift detection and PII masking. It does not perform per-record type checking or record exclusion.

schema:
  email: { type: string, pii: true }
  phone: { type: string, pii: true, nullable: true }
  amount: { type: number }
  active: { type: boolean }
  name: { type: string, nullable: true }

Each field in the schema map supports three properties:

Property	Type	Default	Description
type	string	N/A	Expected data type: string, number, integer, boolean. Stored in the compiled plan for downstream tooling but not enforced at runtime.
pii	boolean	false	Whether this field contains personally identifiable information. PII fields are masked in the canonical API for non-admin users and are tracked for compliance. This property IS enforced at runtime.
nullable	boolean	false	Whether null/empty values are allowed. Stored in the compiled plan for downstream tooling but not enforced at runtime.

How Drift Detection Works

During reconciliation, the worker checks each source that has a schema defined:

1. It takes the first entity from that source as a sample.
2. It compares the actual data field names against the expected field names declared in the schema.
3. If there are missing fields (expected but not present) or extra fields (present but not expected), schema drift is detected and a warning is logged.

Drift detection is a field-presence check only. It does not validate data types, nullability, or individual record values. No records are excluded based on schema drift.

Governance Integration

When the governance policy require_schema: true is set and schema drift is detected, the entire pipeline aborts with a governance violation error. Without governance enforcement, drift produces a warning but reconciliation continues normally. See Governance for details.

Example: PII-Aware Schema

sources:
  crm:
    adapter: csv
    location: data/contacts.csv
    primary_key: id
    schema:
      email:
        type: string
        pii: true
      phone:
        type: string
        pii: true
        nullable: true
      status:
        type: string
      lifetime_value:
        type: number

PII annotations integrate with the compliance system. Fields marked pii: true are:

• Automatically listed in the source's PII field inventory.
• Masked in API responses for non-admin users when PII masking is enabled.
• Included in compliance audit reports.

WARNING

The field_type and nullable properties are compiled into the identity plan and are available for downstream tooling (e.g., data quality dashboards, lineage systems) but are not checked by the reconciliation engine at runtime. Only pii has a runtime effect (PII masking).

Tags

Tags are arbitrary string labels attached to a source. They have no effect on matching or survivorship; they exist for organizational, filtering, and governance purposes.

tags: [production, finance, priority-1]

Use tags to:

• Filter sources in dashboards and reports.
• Scope governance policies to sources with specific tags (e.g., require freshness only for production sources).
• Document ownership and data classification (e.g., pii, external, team-analytics).

Tags are returned in API responses and included in audit log events.

Sampling

Sampling lets you run reconciliation against a subset of rows from a source. This is useful for testing specs against large datasets without waiting for a full run.

sampling:
  strategy: random    # random, stratified, or time_based
  rate: 0.1           # 10% of rows

Field	Type	Required	Default	Description
strategy	string	Yes	N/A	Sampling strategy: random, stratified, or time_based.
rate	float	No	1.0	Fraction of rows to sample (0.0 to 1.0). 1.0 means all rows.

Sampling Strategies

Strategy	Description
random	Uniform random sampling. Each row has an equal probability of being selected.
stratified	Preserves the distribution of key fields. Ensures proportional representation across groups.
time_based	Samples based on temporal ordering. Useful when you want the most recent or a time-windowed slice.

Sampling is applied at read time, before any normalization or matching.

WARNING

Sampling is intended for development and testing. Remove or set rate: 1.0 before running production reconciliations, as sampling can cause legitimate matches to be missed.

Temporal

Temporal configuration enables time-aware matching. When records have validity windows (e.g., a customer address that was valid from January to March), the temporal strategy controls how overlapping and non-overlapping time ranges are handled.

temporal:
  valid_from: effective_date        # field containing the start timestamp
  valid_to: expiration_date         # field containing the end timestamp (null = still active)
  strategy: latest_only             # latest_only, point_in_time, or bi_temporal

Field	Type	Required	Default	Description
valid_from	string	Yes	N/A	Field name containing the valid-from timestamp.
valid_to	string	No	N/A	Field name containing the valid-to timestamp. Null or omitted means the record is still active.
strategy	string	No	latest_only	Temporal strategy: latest_only, point_in_time, or bi_temporal.

Temporal Strategies

Strategy	Description
latest_only	Only the most recent version of each record participates in matching. Historical records are ignored.
point_in_time	Records are matched based on overlapping validity windows. Two records can only match if their time ranges overlap.
bi_temporal	Full bi-temporal matching considering both transaction time and valid time.

Complete Example

A full spec with two richly configured sources demonstrating freshness, schema validation, tags, attributes, temporal, and sampling.

api_version: kanoniv/v2
identity_version: "2.1"

entity:
  name: customer
  description: "Unified customer identity across CRM and billing systems"

sources:
  # -- CRM (loaded via Source.from_warehouse with PostgreSQL) -----------
  crm:
    adapter: postgres
    location: public.contacts
    primary_key: contact_id
    attributes:
      name: full_name
      email: email_address
      phone: phone_number
      company: company_name
    freshness:
      max_age: "24h"
      warn_after: "12h"
    schema:
      email:
        type: string
        pii: true
      phone:
        type: string
        pii: true
        nullable: true
      name:
        type: string
    tags: [production, crm, priority-1]
    temporal:
      valid_from: created_at
      strategy: latest_only

  # -- Monthly billing export -------------------------------------------
  billing:
    adapter: csv
    location: data/exports/stripe_customers_2026_01.csv
    primary_key: stripe_customer_id
    attributes:
      name: name
      email: email
      phone: phone
      plan: plan
      mrr: mrr
    freshness:
      max_age_hours: 48
      warn_after_hours: 36
    schema:
      email:
        type: string
        pii: true
      mrr:
        type: number
      plan:
        type: string
    tags: [production, billing, finance]
    sampling:
      strategy: random
      rate: 0.1

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

survivorship:
  default: source_priority
  overrides:
    - field: email
      strategy: most_recent
    - field: mrr
      strategy: aggregate
      function: max

decision:
  scoring: weighted_sum
  thresholds:
    match: 0.9
    review: 0.7

metadata:
  owner: [email protected]
  tags: [production, customer, v2]
  description: "Production customer identity resolution spec"

Next Steps

• Schema Reference: Full reference for every top-level key in a spec
• Rules: Configure matching algorithms, weights, and composites
• Survivorship: Control how golden record fields are chosen
• Source Freshness Monitoring: Deep dive into freshness checks and alerting
• Python SDK: Load sources programmatically with Source.from_csv, Source.from_pandas, and more