From Zero to Normalized: Connecting Your First Multi-Vendor Feed

This walkthrough is for a data engineer who has two vendor CSVs with overlapping entity domains and wants to produce a clean, unified output table. We will go from a fresh HyperNorm account to canonical normalized output. Expected time: 12-15 minutes if you have your vendor files handy.

Prerequisites: Python 3.9+, a HyperNorm API key (free tier works for this walkthrough), and two CSV vendor files. We will use example files throughout this post; swap in your own paths.

Step 1: Install the SDK and Authenticate

pip install hypernorm

Once installed, set your API key as an environment variable. Do not hardcode it in your config file.

export HYPERNORM_API_KEY="hn_live_yourkey"

Verify the connection:

import hypernorm

client = hypernorm.Client()
print(client.ping())
# Expected: {"status": "ok", "account": "[email protected]"}

Step 2: Inspect Your Vendor Schemas

Before writing a normalization config, you need to understand what each vendor is actually sending. Run schema inference on both files:

schema_a = client.infer_schema("vendor_a_companies.csv")
schema_b = client.infer_schema("vendor_b_companies.csv")

print(schema_a.fields)
# [
#   {"name": "company_name",   "type": "string",  "completeness": 1.0},
#   {"name": "revenue_usd",    "type": "float",   "completeness": 0.94},
#   {"name": "hq_city",        "type": "string",  "completeness": 0.89},
#   {"name": "account_id",     "type": "string",  "completeness": 1.0}
# ]

print(schema_b.fields)
# [
#   {"name": "companyName",    "type": "string",  "completeness": 1.0},
#   {"name": "totalRevenue",   "type": "float",   "completeness": 0.97},
#   {"name": "city",           "type": "string",  "completeness": 0.92},
#   {"name": "vendorId",       "type": "string",  "completeness": 1.0}
# ]

You can see the pattern: company_name vs companyName, revenue_usd vs totalRevenue, hq_city vs city. The entity domain is companies. Both feeds contain company name, revenue, and city. The identifiers (account_id, vendorId) are vendor-specific and will not merge directly.

Step 3: Write the Normalization Config

Create a file called hypernorm-config.yaml. This is the canonical field mapping that HyperNorm will apply to both feeds:

version: "1.0"
canonical_schema:
  company_name:
    type: string
    required: true
  revenue_usd:
    type: float
    unit: USD
    required: false
  city:
    type: string
    required: false

sources:
  vendor_a:
    file: vendor_a_companies.csv
    field_map:
      company_name: company_name
      revenue_usd: revenue_usd
      city: hq_city
  vendor_b:
    file: vendor_b_companies.csv
    field_map:
      company_name: companyName
      revenue_usd: totalRevenue
      city: city

entity_resolution:
  entity_type: company
  match_field: company_name
  blocking_field: city
  similarity_metric: jaro_winkler
  threshold: 0.88
  output_confidence: true

A few things to notice here. The blocking_field: city setting tells the entity resolution step to first group candidate pairs by city before running fuzzy matching. This is important for performance at any non-trivial scale: without blocking you are comparing every company in feed A against every company in feed B, which is O(n x m). Blocking on city reduces the candidate space dramatically.

The threshold: 0.88 is a starting point. You will likely tune this based on your data. A threshold of 0.88 on Jaro-Winkler means matches must share at least 88% character-level similarity after the algorithm's transposition weighting. For company names, this tends to catch abbreviations (Acme Corp / Acme Corporation) while rejecting true non-matches. You should validate this against a sample of your data before running at full volume.

Step 4: Run the Pipeline

result = client.run(config="hypernorm-config.yaml")

print(result.summary)
# {
#   "records_processed": {"vendor_a": 4201, "vendor_b": 3887},
#   "normalized_records": 8088,
#   "entity_pairs_evaluated": 47300,
#   "entities_resolved": 1243,
#   "canonical_entities": 6845,
#   "mean_confidence": 0.936,
#   "low_confidence_count": 84,
#   "runtime_seconds": 8.4
# }

What this is telling you: 1,243 entity pairs were resolved (meaning HyperNorm determined they represent the same company across both feeds). 84 pairs had confidence below the alert threshold (0.70 by default). Those are worth reviewing manually or routing to a data quality queue.

Step 5: Inspect and Export the Canonical Output

df = result.to_dataframe()

print(df.head(3))
#    canonical_id    canonical_name   revenue_usd  city        confidence  sources
# 0  ent_0a1b2c3d  Acme Corporation  1850000.0    Mumbai      0.97        [vendor_a, vendor_b]
# 1  ent_1d2e3f4g  Veltrix Analytics     NaN      Bengaluru   1.00        [vendor_b]
# 2  ent_2g3h4i5j  Karbon Data         320000.0   Pune        0.92        [vendor_a, vendor_b]

result.export("canonical_companies.csv")
result.export("canonical_companies.parquet")  # preferred for downstream warehouse load

The sources column is the provenance trail. For any downstream anomaly, you can trace back to which vendor contributed which record. For the revenue_usd field, where both vendors have a value, HyperNorm uses the weighted confidence score to prefer the higher-completeness source by default. You can override this with an explicit merge strategy in the config if your domain logic says otherwise (e.g., always use vendor_a for revenue because their data quality agreement is stricter).

Step 6: Handle Low-Confidence Matches

The 84 low-confidence pairs deserve attention. They are not necessarily wrong matches, but they are cases where the algorithm is less certain. In most pipelines, you want a review queue rather than auto-merge for these:

review_queue = result.low_confidence_pairs(threshold=0.70)

for pair in review_queue[:3]:
    print(pair)
# {
#   "candidate_a": {"company_name": "Sailwhip Logistics Pvt Ltd", "city": "Chennai"},
#   "candidate_b": {"company_name": "Sailwhip Log.", "city": "Chennai"},
#   "confidence": 0.68,
#   "pair_id": "pair_09f2a1c3"
# }

# Accept or reject manually, or route to a data team workflow tool
result.accept_pair("pair_09f2a1c3")
result.reject_pair("pair_other_id")

In this case, "Sailwhip Logistics Pvt Ltd" and "Sailwhip Log." are almost certainly the same entity, but the truncation dropped the confidence below threshold. After manual review, accepting this pair adds the accepted resolution to your entity registry so future runs recognize the match automatically.

Step 7: Version Your Config and Schedule the Run

The normalization config is infrastructure. Version it in the same repository as your dbt models or your Airflow DAGs. When a vendor changes their schema, the config change is a pull request with a clear diff, not a one-off script edit.

For scheduled runs, HyperNorm's Python SDK works cleanly in an Airflow operator or Prefect task. The config file stays in version control; only the file paths and API key change per environment:

from airflow.decorators import task
import hypernorm

@task
def run_normalization():
    client = hypernorm.Client()
    result = client.run(
        config="configs/hypernorm-config.yaml",
        sources={
            "vendor_a": "s3://your-bucket/vendor_a/latest.csv",
            "vendor_b": "s3://your-bucket/vendor_b/latest.csv"
        }
    )
    result.export("s3://your-bucket/canonical/companies-latest.parquet")
    return result.summary

What Comes Next

This walkthrough covered the single most common case: two overlapping feeds, company entity type, Jaro-Winkler matching on name with city blocking. The same config structure extends to more vendors, different entity types (products, addresses, people), and custom blocking strategies.

A few things to investigate as you get comfortable with your pipeline: the schema_version field in the config for handling vendor format changes without losing historical records; the alert_rules section for setting completeness drop thresholds that trigger notifications; and the entity registry API for querying canonical IDs directly from downstream models without re-running the full pipeline.

The full config reference is in the API docs. The connector docs cover loading directly from Snowflake or BigQuery tables rather than flat files, which is the more common production setup once you have the pipeline working.