[Python] Optimizing Hierarchical Address Similarity Calculation on Large DataFrame with Dask

I’m working with a very large dataset (tens of millions of rows) that contains address combinations across many countries. Each row represents a pair of addresses, and I’ve already parsed these addresses using the deepparse model, so my DataFrame now contains structured fields :

'Address1_normalized',
'Address1_street_number',
'Address1_street_name',
'Address1_city',
'Address1_postal_code',
'Address1_unit'
'Address2_normalized',
'Address2_street_number',
'Address2_street_name',
'Address2_city',
'Address2_postal_code',
'Address2_unit'


Now, I want to compute similarity scores between each pair of addresses using a hierarchical logic (e.g., exact match first, then progressively looser matching based on parsed components). I’m currently using Dask for distributed computation.

However, the current solution is quite slow: for 25 million rows, it takes around 36 minutes. I expect the number of address combinations to grow significantly, so I’m looking for ideas to optimize or redesign the approach.

def calculate_address_similarities_dask(df: pd.DataFrame, threshold: float = None, npartitions: int = 100, blocking: bool = True):
"""
Dask-based address similarity with early stopping and optional blocking.

Args:
df (pd.DataFrame): Input DataFrame with parsed address components.
threshold (float): Optional similarity threshold to filter results.
npartitions (int): Number of Dask partitions.
blocking (bool): Whether to apply pre-filtering based on city/postal_code match.

Returns:
pd.DataFrame: Resulting DataFrame with similarity scores (filtered if threshold is set).
"""
import pandas as pd
import numpy as np
import dask.dataframe as dd
from rapidfuzz.fuzz import ratio
import logging
import time

start_time = time.time()
logging.info(f"\n=== Dask Address Similarity Calculation ===")
logging.info(f"Input size: {len(df):,} rows")

# Component weights
weights = {
'street_number': 0.15,
'street_name': 0.30,
'city': 0.25,
'postal_code': 0.20,
'unit': 0.10
}

max_remaining = np.cumsum(list(weights.values())[::-1])[::-1]

# Step 1: Pre-filter with blocking on city or postal_code
if blocking:
before_block = len(df)
city_match = df["Address1_city"] == df["Address2_city"]
postal_match = df["Address1_postal_code"] == df["Address2_postal_code"]
df = df[city_match | postal_match]
logging.info(f"Blocking by city/postal_code reduced from {before_block:,} to {len(df):,} rows")

#identical normalized addresses
df["similarity"] = np.nan
identical_mask = df["Address1_normalized"] == df["Address2_normalized"]
df.loc[identical_mask, "similarity"] = 100.0
df_to_process = df[~identical_mask].copy()
identical_count = identical_mask.sum()
logging.info(f"Identical addresses: {identical_count:,}")

#early stopping
def compute_similarity_row(row):
total = 0.0
i = 0
for comp, weight in weights.items():
val1 = str(row[f"Address1_{comp}"]) if pd.notna(row[f"Address1_{comp}"]) else ""
val2 = str(row[f"Address2_{comp}"]) if pd.notna(row[f"Address2_{comp}"]) else ""

if val1 == val2:
sim = 100.0 if val1 else 0.0
else:
sim = ratio(val1, val2)
total += sim * weight

#Early stopping
if threshold is not None:
max_possible = total + max_remaining * 100
if max_possible < threshold:
return 0.0 # can't reach threshold
i += 1
return total

# Use of Dask
ddf = dd.from_pandas(df_to_process, npartitions=npartitions)
ddf = ddf.assign(similarity=ddf.map_partitions(
lambda part: part.apply(compute_similarity_row, axis=1),
meta=('similarity', 'f8')
))

df_to_process = ddf.compute()
df.loc[~identical_mask, "similarity"] = df_to_process["similarity"].astype(np.float32)

#Apply threshold
if threshold is not None:
df = df[df["similarity"] >= threshold]
logging.info(f"After threshold filter (≥ {threshold}): {len(df):,} rows")

# Stats
total_time = time.time() - start_time
logging.info(f"Total time: {total_time:.2f}s")
logging.info(f"Similarity stats - Avg: {df['similarity'].mean():.2f}, Max: {df['similarity'].max():.2f}, Min: {df['similarity'].min():.2f}")
return df

Continue reading...