Оптимизация иерархического сходства сходства адреса на большем данных о данных с DASK ⇐ Python

[Anonymous]

Я работаю с очень большим набором данных (десятки миллионов строк), который содержит комбинации адресов во многих странах. Каждая строка представляет собой пару адресов, и я уже проанализировал эти адреса, используя модель Deepparse, поэтому мой флаг Data теперь содержит структурированные поля: < /p>
'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'
< /code>
Теперь я хочу вычислить показатели сходства между каждой парой адресов, используя иерархическую логику (например, точное совпадение сначала, затем постепенно более слабое сочетание на основе проанализированных компонентов). В настоящее время я использую DASK для распределенных вычислений. Я ожидаю значительного растущего количества комбинаций адресов, поэтому я ищу идеи для оптимизации или перепроектирования подхода. < /P>
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


Подробнее здесь: https://stackoverflow.com/questions/796 ... ame-with-d