Scaling up to large datasets¶
The previous tutorials showed how to build, estimate and use a linkage model. On small to medium sized data, the default settings 'just work' and you don't need to do anything special.
As your input data grows to tens or hundreds of millions of records, some steps can become slow to run, or you may run out of memory. This tutorial introduces the tools Splink provides to ensure you can still run linkages even on very large datasets.
- Model training — using sampling to speed up blocking analysis and parameter estimation.
- Inference — using chunking to control the memory used by
predict(), and to distribute the work across multiple processes or machines.
Only reach for these techniques when you need them
The techniques on this page trade something away — either accuracy (sampling) or simplicity (chunking). There is no need to use any of them if your job is already running quickly. Start with the defaults, and only introduce sampling or chunking at the point where a step becomes too slow or runs out of memory.
To keep this tutorial fast to run, we use the small fake_1000 dataset. In reality you would never need these techniques on data this small — the examples are here purely to demonstrate the API. As you read, imagine the same code running on a dataset that has hundreds of millions of rows.
import splink.comparison_library as cl
from splink import (
ColumnExpression,
DuckDBAPI,
Linker,
SettingsCreator,
block_on,
splink_datasets,
)
df = splink_datasets.fake_1000
db_api = DuckDBAPI()
df_sdf = db_api.register(df)
settings = SettingsCreator(
link_type="dedupe_only",
comparisons=[
cl.NameComparison("first_name"),
cl.NameComparison("surname"),
cl.LevenshteinAtThresholds(ColumnExpression("dob").cast_to_string(), 1),
cl.ExactMatch("city").configure(term_frequency_adjustments=True),
cl.EmailComparison("email"),
],
blocking_rules_to_generate_predictions=[
block_on("first_name", "city"),
block_on("surname"),
block_on("dob"),
],
retain_intermediate_calculation_columns=True,
)
linker = Linker(df_sdf, settings)
Section 1: Blocking analysis and model training¶
Sampling in blocking rule analysis¶
In tutorial 3 we used count_comparisons_from_blocking_rules and chart_comparisons_from_blocking_rules to count how many comparisons our blocking rules generate.
Computing these counts exactly means executing the blocking join over the whole dataset, which can be expensive on large data. To avoid this, both functions accept a record_sample_proportion argument. Splink takes a deterministic sample of the input records on each side of the blocking join, counts the comparisons generated by the sample, and scales the result back up to estimate the full count.
record_sample_proportion defaults to 0.05 (a 5% sample), so these functions already estimate from a sample by default. Lowering it makes the estimate faster but less precise; setting record_sample_proportion=1.0 computes the exact count.
Sampling is quadratically cheaper
Blocking is a join, so the number of comparisons grows quadratically with the number of records. Because record_sample_proportion samples both sides of the join, taking a fraction p of records causes a reduction in comparisons of p². Splink then scales the count back up by 1 / p²). So a 10% sample (record_sample_proportion=0.1) generates roughly 1% of the comparisons.
from splink.blocking_analysis import count_comparisons_from_blocking_rules
# Estimate the count from a 30% sample of records (faster, approximate)
counts_estimated = count_comparisons_from_blocking_rules(
df_sdf,
blocking_rules=block_on("first_name", "city"),
link_type="dedupe_only",
record_sample_proportion=0.3,
)
estimated_count = counts_estimated[0]["marginal_comparison_count"]
print(f"Estimated from 30% sample: {estimated_count:,} comparisons")
Estimated from 30% sample: 356 comparisons
/home/runner/work/splink/splink/splink/internals/blocking_analysis.py:668: UserWarning: The sampled blocking analysis estimate for blocking rule '(l."first_name" = r."first_name") AND (l."city" = r."city")' is based on 32 sampled pairwise comparisons. This is below the recommended minimum of 1,000, so the estimate may be unstable. Increase record_sample_proportion for a more stable estimate.
return _cumulative_comparisons_to_be_scored_from_blocking_rules(
# Compute the exact count over all records (slower, exact)
counts_exact = count_comparisons_from_blocking_rules(
df_sdf,
blocking_rules=block_on("first_name", "city"),
link_type="dedupe_only",
record_sample_proportion=1.0,
)
exact_count = counts_exact[0]["marginal_comparison_count"]
print(f"Exact over all records: {exact_count:,} comparisons")
Exact over all records: 315 comparisons
On a small dataset like this the sampled estimate may differ noticeably from the exact count — there simply aren't many records to sample from. On large datasets a small sample gives a good estimate, which is exactly where the speed-up matters.
The same record_sample_proportion argument is available on chart_comparisons_from_blocking_rules.
Sampling when estimating probability_two_random_records_match¶
estimate_probability_two_random_records_match works by counting how many pairs are matched by a set of deterministic rules. On large data, counting these matches exactly can be slow.
The method accepts the same record_sample_proportion argument, which samples each side of the deterministic blocking join and scales the observed match count back up. It defaults to 1.0 (exact). Lower it on large data to trade a little accuracy for speed.
deterministic_rules = [
block_on("first_name", "dob"),
"l.first_name = r.first_name and levenshtein(r.surname, l.surname) <= 2",
block_on("email"),
]
linker.training.estimate_probability_two_random_records_match(
deterministic_rules,
recall=0.7,
record_sample_proportion=0.5,
)
/home/runner/work/splink/splink/splink/internals/linker_components/training.py:102: UserWarning: The sampled blocking analysis estimate for blocking rule '(l."first_name" = r."first_name") AND (l."dob" = r."dob")' is based on 57 sampled pairwise comparisons. This is below the recommended minimum of 1,000, so the estimate may be unstable. Increase record_sample_proportion for a more stable estimate.
records = _cumulative_comparisons_to_be_scored_from_blocking_rules(
/home/runner/work/splink/splink/splink/internals/linker_components/training.py:102: UserWarning: The sampled blocking analysis estimate for blocking rule 'l.first_name = r.first_name and levenshtein(r.surname, l.surname) <= 2' is based on 66 sampled pairwise comparisons. This is below the recommended minimum of 1,000, so the estimate may be unstable. Increase record_sample_proportion for a more stable estimate.
records = _cumulative_comparisons_to_be_scored_from_blocking_rules(
/home/runner/work/splink/splink/splink/internals/linker_components/training.py:102: UserWarning: The sampled blocking analysis estimate for blocking rule 'l."email" = r."email"' is based on 129 sampled pairwise comparisons. This is below the recommended minimum of 1,000, so the estimate may be unstable. Increase record_sample_proportion for a more stable estimate.
records = _cumulative_comparisons_to_be_scored_from_blocking_rules(
Probability two random records match is estimated to be 0.00288.
This means that amongst all possible pairwise record comparisons, one in 346.88 are expected to match. With 499,500 total possible comparisons, we expect a total of around 1,440.00 matching pairs
Controlling the cost of estimate_u_using_random_sampling with max_pairs¶
estimate_u_using_random_sampling estimates the u probabilities by sampling random pairs of records. The max_pairs argument controls how many pairs are sampled and is the main lever on both runtime and accuracy:
- Larger
max_pairs→ more accurateuestimates, but longer runtime. - Smaller
max_pairs→ faster, but noisier estimates.
While iterating on a model specification, a smaller value such as 1e6 or 1e7 keeps the feedback loop fast. For a final model we recommend 1e8 or even 1e9.
Importantly, note that for a constant max_pairs estimate_u_using_random_sampling will run at a similar speed irrespective of the size of the dataset, since it's the number of comparison pairs which is the main determinant of the runtime for this function.
linker.training.estimate_u_using_random_sampling(max_pairs=2e6)
----- Estimating u probabilities using random sampling -----
Estimating u with: max_pairs = 2,000,000, min_count_per_level = 100, num_chunks = 10
Estimating u for: first_name (Comparison 1 of 5)
Running probe chunk (~1.00% of max_pairs)
Min u_count: 7 for comparison level Jaro-Winkler distance of first_name >= 0.88 (cvv=2)
Probe did not converge; restarting with normal chunking
Running chunk 1/10
Count of 31 for level Jaro-Winkler distance of first_name >= 0.88 (cvv=2). Chunk took 0.0 seconds.
Min u_count not hit, continuing.
Running chunk 2/10
Count of 92 for level Jaro-Winkler distance of first_name >= 0.88 (cvv=2). Chunk took 0.0 seconds.
Min u_count not hit, continuing.
Running chunk 3/10
Count of 156 for level Jaro-Winkler distance of first_name >= 0.88 (cvv=2). Chunk took 0.0 seconds.
Exiting early since min count of 156 exceeds min_count_per_level = 100
Estimating u for: surname (Comparison 2 of 5)
Running probe chunk (~1.00% of max_pairs)
Min u_count: 0 for comparison level Jaro-Winkler distance of surname >= 0.88 (cvv=2)
Probe did not converge; restarting with normal chunking
Running chunk 1/10
Count of 54 for level Jaro-Winkler distance of surname >= 0.88 (cvv=2). Chunk took 0.0 seconds.
Min u_count not hit, continuing.
Running chunk 2/10
Count of 89 for level Jaro-Winkler distance of surname >= 0.88 (cvv=2). Chunk took 0.0 seconds.
Min u_count not hit, continuing.
Running chunk 3/10
Count of 150 for level Jaro-Winkler distance of surname >= 0.88 (cvv=2). Chunk took 0.0 seconds.
Exiting early since min count of 150 exceeds min_count_per_level = 100
Estimating u for: dob (Comparison 3 of 5)
Running probe chunk (~1.00% of max_pairs)
Min u_count: 19 for comparison level Exact match on transformed dob (cvv=2)
Probe did not converge; restarting with normal chunking
Running chunk 1/10
Count of 92 for level Levenshtein distance of transformed dob <= 1 (cvv=1). Chunk took 0.0 seconds.
Min u_count not hit, continuing.
Running chunk 2/10
Count of 154 for level Levenshtein distance of transformed dob <= 1 (cvv=1). Chunk took 0.0 seconds.
Exiting early since min count of 154 exceeds min_count_per_level = 100
Estimating u for: city (Comparison 4 of 5)
Running probe chunk (~1.00% of max_pairs)
Min u_count: 414 for comparison level Exact match on city (cvv=1)
Exiting early since min count of 414 exceeds min_count_per_level = 100
Estimating u for: email (Comparison 5 of 5)
Running probe chunk (~1.00% of max_pairs)
Min u_count: 1 for comparison level Jaro-Winkler >0.88 on username (cvv=1)
Probe did not converge; restarting with normal chunking
Running chunk 1/10
Count of 13 for level Jaro-Winkler >0.88 on username (cvv=1). Chunk took 0.1 seconds.
Min u_count not hit, continuing.
Running chunk 2/10
Count of 22 for level Jaro-Winkler >0.88 on username (cvv=1). Chunk took 0.1 seconds.
Min u_count not hit, continuing.
Running chunk 3/10
Count of 35 for level Jaro-Winkler >0.88 on username (cvv=1). Chunk took 0.1 seconds.
Min u_count not hit, continuing.
Running chunk 4/10
Count of 61 for level Jaro-Winkler >0.88 on username (cvv=1). Chunk took 0.1 seconds.
Min u_count not hit, continuing.
Running chunk 5/10
Count of 99 for level Jaro-Winkler >0.88 on username (cvv=1). Chunk took 0.1 seconds.
Min u_count not hit, continuing.
Running chunk 6/10
Count of 113 for level Jaro-Winkler >0.88 on username (cvv=1). Chunk took 0.1 seconds.
Exiting early since min count of 113 exceeds min_count_per_level = 100
Estimated u probabilities using random sampling
Your model is not yet fully trained. Missing estimates for:
- first_name (no m values are trained).
- surname (no m values are trained).
- dob (no m values are trained).
- city (no m values are trained).
- email (no m values are trained).
Sampling in Expectation Maximisation with max_pairs¶
estimate_parameters_using_expectation_maximisation generates pairwise comparisons using the training blocking rule and iterates over them. On large data, a training rule can still generate a very large number of pairs, making each EM iteration slow.
Two arguments let you cap this cost:
max_pairs(defaultNone): if set, Splink limits the approximate number of blocked pairs used for EM training to this value. It runs a quick preliminary blocking pass to estimate the full blocked-pair count, then applies a deterministic filter to the input records so that the resulting number of blocked pairs is approximatelymax_pairs. With the default ofNone, all blocked pairs are used (no sampling).record_sample_proportion(default0.01): the fraction of input records sampled on each side of that preliminary pass used to estimate the full blocked-pair count. It does not directly set the final number of training pairs —max_pairsremains the primary control.
In other words, set max_pairs to cap how much data each EM training session uses. Here we pass a deliberately high max_pairs so the small fake_1000 model is not actually downsampled.
An important caveat is that for EM training to work, your pairs must contain a reasonable number of matches. If you choose a very loose blocking rule like block_on("gender") on a large dataset, even a large max_pairs value may result in very few true matches, and the training is unlikely to succeed.
training_blocking_rule = block_on("dob")
linker.training.estimate_parameters_using_expectation_maximisation(
training_blocking_rule,
max_pairs=1e7
)
----- Starting EM training session -----
[EM sampling] Probe at proportion 0.010000 (actual fraction 0.010000, sample_threshold=100 / sample_modulus=10,000) -> 0 blocked pairs
[EM sampling] Probe returned zero blocked pairs; skipping sampling
Estimating the m probabilities of the model by blocking on:
l."dob" = r."dob"
Parameter estimates will be made for the following comparison(s):
- first_name
- surname
- city
- email
Parameter estimates cannot be made for the following comparison(s) since they are used in the blocking rules:
- dob
WARNING:
Level Jaro-Winkler >0.88 on username on comparison email not observed in dataset, unable to train m value
Iteration 1: Largest change in params was 0.454 in probability_two_random_records_match
Iteration 2: Largest change in params was 0.331 in probability_two_random_records_match
Iteration 3: Largest change in params was 0.0824 in the m_probability of first_name, level `All other comparisons`
Iteration 4: Largest change in params was 0.0617 in the m_probability of first_name, level `All other comparisons`
Iteration 5: Largest change in params was 0.0204 in probability_two_random_records_match
Iteration 6: Largest change in params was 0.00825 in probability_two_random_records_match
Iteration 7: Largest change in params was 0.00369 in probability_two_random_records_match
Iteration 8: Largest change in params was 0.00174 in probability_two_random_records_match
Iteration 9: Largest change in params was 0.000845 in probability_two_random_records_match
Iteration 10: Largest change in params was 0.000415 in probability_two_random_records_match
Iteration 11: Largest change in params was 0.000205 in probability_two_random_records_match
Iteration 12: Largest change in params was 0.000102 in probability_two_random_records_match
Iteration 13: Largest change in params was 5.06e-05 in probability_two_random_records_match
EM converged after 13 iterations
m probability not trained for email - Jaro-Winkler >0.88 on username (comparison vector value: 1). This usually means the comparison level was never observed in the training data.
Your model is not yet fully trained. Missing estimates for:
- dob (no m values are trained).
- email (some m values are not trained).
<EMTrainingSession, blocking on l."dob" = r."dob", deactivating comparisons dob>
We now have a trained model, which we will use to demonstrate the inference techniques below.
Section 2: Scaling inference with chunking¶
predict() generates every blocked pair and scores it in a single pass. On very large datasets, this may be time consuming, or you may even run into out of memory or out of disk errors. Splink provides a family of chunking tools that split this work into smaller pieces.
By 'chunking' we mean that Splink allows you to compute a fixed proportion of the result. This allows you to complete your linkage in n smaller pieces, which, when combined represent the full result.
The pieces are defined by a grid: the left side of each comparison is split into num_chunks_left slices and the right side into num_chunks_right slices, giving num_chunks_left × num_chunks_right chunks in total. Every blocked pair falls into exactly one chunk, so processing all chunks produces exactly the same result as a single predict() call.
Chunked predict() with num_chunks_left and num_chunks_right¶
The simplest way to use chunking is to pass num_chunks_left and num_chunks_right to predict(). Splink then processes the chunks one after another (in series) and unions the results for you, so the return value is identical to an unchunked predict().
This gives two benefits on large jobs:
- Lower peak memory — each chunk materialises only a fraction of the blocked pairs at a time.
- Progress reporting — because the chunks run in series, Splink logs progress after each one. With the default logging level (
INFO) you will see messages such as:
Processing chunk (1, 4) x (1, 4) [1/16]
Completed chunk 1/16 (6%) | Elapsed: 60.0s | Remaining: ~900.0s | Total: ~960.0s
Processing chunk (1, 4) x (2, 4) [2/16]
...
This makes it possible to estimate how long a long-running job will take, which you don't get from a single opaque predict() call.
predictions_chunked = linker.inference.predict(
threshold_match_probability=0.9,
num_chunks_left=4,
num_chunks_right=4,
)
# The result is identical to an unchunked predict()
predictions_single = linker.inference.predict(threshold_match_probability=0.9)
chunked_count = predictions_chunked.as_duckdbpyrelation().count("*").fetchone()[0]
single_count = predictions_single.as_duckdbpyrelation().count("*").fetchone()[0]
print(f"Chunked predict produced {chunked_count} rows")
print(f"Single predict produced {single_count} rows")
Processing chunk (1, 4) x (1, 4) [1/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 1/16 (6%) | Elapsed: 0.1s | Remaining: ~1.3s | Total: ~1.4s
Processing chunk (1, 4) x (2, 4) [2/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 2/16 (12%) | Elapsed: 0.2s | Remaining: ~1.2s | Total: ~1.4s
Processing chunk (1, 4) x (3, 4) [3/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 3/16 (19%) | Elapsed: 0.3s | Remaining: ~1.1s | Total: ~1.4s
Processing chunk (1, 4) x (4, 4) [4/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 4/16 (25%) | Elapsed: 0.3s | Remaining: ~1.0s | Total: ~1.3s
Processing chunk (2, 4) x (1, 4) [5/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.08 seconds
Completed chunk 5/16 (31%) | Elapsed: 0.4s | Remaining: ~0.9s | Total: ~1.4s
Processing chunk (2, 4) x (2, 4) [6/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 6/16 (38%) | Elapsed: 0.5s | Remaining: ~0.9s | Total: ~1.4s
Processing chunk (2, 4) x (3, 4) [7/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 7/16 (44%) | Elapsed: 0.6s | Remaining: ~0.8s | Total: ~1.4s
Processing chunk (2, 4) x (4, 4) [8/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.06 seconds
Completed chunk 8/16 (50%) | Elapsed: 0.7s | Remaining: ~0.7s | Total: ~1.3s
Processing chunk (3, 4) x (1, 4) [9/16]
Blocking time: 0.00 seconds
Predict time (post-blocking): 0.06 seconds
Completed chunk 9/16 (56%) | Elapsed: 0.7s | Remaining: ~0.6s | Total: ~1.3s
Processing chunk (3, 4) x (2, 4) [10/16]
Blocking time: 0.00 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 10/16 (62%) | Elapsed: 0.8s | Remaining: ~0.5s | Total: ~1.3s
Processing chunk (3, 4) x (3, 4) [11/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.06 seconds
Completed chunk 11/16 (69%) | Elapsed: 0.9s | Remaining: ~0.4s | Total: ~1.3s
Processing chunk (3, 4) x (4, 4) [12/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 12/16 (75%) | Elapsed: 1.0s | Remaining: ~0.3s | Total: ~1.3s
Processing chunk (4, 4) x (1, 4) [13/16]
Blocking time: 0.00 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 13/16 (81%) | Elapsed: 1.1s | Remaining: ~0.2s | Total: ~1.3s
Processing chunk (4, 4) x (2, 4) [14/16]
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 14/16 (88%) | Elapsed: 1.1s | Remaining: ~0.2s | Total: ~1.3s
Processing chunk (4, 4) x (3, 4) [15/16]
Blocking time: 0.00 seconds
Predict time (post-blocking): 0.07 seconds
Completed chunk 15/16 (94%) | Elapsed: 1.2s | Remaining: ~0.1s | Total: ~1.3s
Processing chunk (4, 4) x (4, 4) [16/16]
Blocking time: 0.00 seconds
Predict time (post-blocking): 0.08 seconds
Completed chunk 16/16 (100%) | Elapsed: 1.3s | Remaining: ~0.0s | Total: ~1.3s
Total chunked prediction time: 1.32 seconds
-- WARNING --
You have called predict(), but there are some parameter estimates which have neither been estimated or specified in your settings dictionary. To produce predictions the following untrained parameters will use default values.
Comparison: 'dob':
m values not fully trained
Comparison: 'email':
m values not fully trained
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.05 seconds
-- WARNING --
You have called predict(), but there are some parameter estimates which have neither been estimated or specified in your settings dictionary. To produce predictions the following untrained parameters will use default values.
Comparison: 'dob':
m values not fully trained
Comparison: 'email':
m values not fully trained
Chunked predict produced 989 rows
Single predict produced 989 rows
Verifying your pipeline with predict_chunk()¶
Before committing to a long full run, it's useful to confirm that a single chunk works end to end. predict_chunk() computes and scores just one slice of the grid.
In the below (1,4) means 'compute chunk 1 of of 4' on the left side, and 'compute chunk 1 of 4' on the right side.
single_chunk = linker.inference.predict_chunk(
left_chunk=(1, 4),
right_chunk=(1, 4),
threshold_match_probability=0.9,
)
single_chunk.as_duckdbpyrelation().limit(5).show(max_width=10000)
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.09 seconds
-- WARNING --
You have called predict(), but there are some parameter estimates which have neither been estimated or specified in your settings dictionary. To produce predictions the following untrained parameters will use default values.
Comparison: 'dob':
m values not fully trained
Comparison: 'email':
m values not fully trained
┌────────────────────┬────────────────────┬─────────────┬─────────────┬──────────────┬──────────────┬──────────────────┬───────────────────────┬───────────────────────┬───────────────────┬──────────────────────┬───────────┬───────────┬───────────────┬──────────────────────┬──────────────────────┬───────────────────┬─────────────────────┬────────────┬────────────┬───────────┬───────────────────┬───────────┬───────────┬────────────┬───────────────────────┬─────────────────────┬─────────────────────┬────────────────────┬──────────────────────────────┬──────────────────────────────┬─────────────┬───────────────────────┬───────────────────────┬───────────────────┬─────────────────────┬───────────┐
│ match_weight │ match_probability │ unique_id_l │ unique_id_r │ first_name_l │ first_name_r │ gamma_first_name │ tf_first_name_l │ tf_first_name_r │ mw_first_name │ mw_tf_adj_first_name │ surname_l │ surname_r │ gamma_surname │ tf_surname_l │ tf_surname_r │ mw_surname │ mw_tf_adj_surname │ dob_l │ dob_r │ gamma_dob │ mw_dob │ city_l │ city_r │ gamma_city │ tf_city_l │ tf_city_r │ mw_city │ mw_tf_adj_city │ email_l │ email_r │ gamma_email │ tf_email_l │ tf_email_r │ mw_email │ mw_tf_adj_email │ match_key │
│ double │ double │ int64 │ int64 │ varchar │ varchar │ int32 │ double │ double │ double │ double │ varchar │ varchar │ int32 │ double │ double │ double │ double │ date │ date │ int32 │ double │ varchar │ varchar │ int32 │ double │ double │ double │ double │ varchar │ varchar │ int32 │ double │ double │ double │ double │ varchar │
├────────────────────┼────────────────────┼─────────────┼─────────────┼──────────────┼──────────────┼──────────────────┼───────────────────────┼───────────────────────┼───────────────────┼──────────────────────┼───────────┼───────────┼───────────────┼──────────────────────┼──────────────────────┼───────────────────┼─────────────────────┼────────────┼────────────┼───────────┼───────────────────┼───────────┼───────────┼────────────┼───────────────────────┼─────────────────────┼─────────────────────┼────────────────────┼──────────────────────────────┼──────────────────────────────┼─────────────┼───────────────────────┼───────────────────────┼───────────────────┼─────────────────────┼───────────┤
│ 5.720304184869038 │ 0.9813852847260484 │ 79 │ 84 │ Ryan │ Ryan │ 4 │ 0.006016847172081829 │ 0.006016847172081829 │ 6.309869029835824 │ 0.040393298600093885 │ Cole │ Cole │ 4 │ 0.006105006105006105 │ 0.006105006105006105 │ 6.580052440474188 │ -0.4280902099027699 │ 1987-05-25 │ 1988-05-27 │ 0 │ -5.31710654504937 │ Brritsol │ Bristol │ 0 │ 0.0012300123001230013 │ 0.01968019680196802 │ -1.0009194349559887 │ 0.0 │ r.cole1@ramirez-anthony.com │ r.cole1@ramtrez-anihony.com │ 3 │ 0.0063371356147021544 │ 0.0012674271229404308 │ 7.970212534477016 │ 0.0 │ 1 │
│ 16.76010466025434 │ 0.9999909904758449 │ 176 │ 178 │ Elelie │ Ellie │ 2 │ 0.0012033694344163659 │ 0.0024067388688327317 │ 5.776225974372055 │ 0.0 │ Ellis │ Eislls │ 1 │ 0.003663003663003663 │ 0.001221001221001221 │ 2.451279207282434 │ 0.0 │ 2005-12-08 │ 2005-12-08 │ 2 │ 8.996493872732788 │ London │ NULL │ -1 │ 0.21279212792127922 │ NULL │ 0.0 │ 0.0 │ elliee@guzman.com │ elliee@oguzmanoc.m │ 3 │ 0.0038022813688212928 │ 0.0012674271229404308 │ 7.970212534477016 │ 0.0 │ 2 │
│ 21.90509080470398 │ 0.9999997453694551 │ 528 │ 529 │ Oliver │ Oliver │ 4 │ 0.03369434416365824 │ 0.03369434416365824 │ 6.309869029835824 │ -2.445033528570148 │ Dawson │ Dawon │ 3 │ 0.002442002442002442 │ 0.001221001221001221 │ 6.543833593077295 │ 0.0 │ 1985-11-03 │ 1985-11-03 │ 2 │ 8.996493872732788 │ Liverpool │ Liverpool │ 1 │ 0.04059040590405904 │ 0.04059040590405904 │ 2.3251008108518043 │ 1.4500845869128298 │ odawson45@malone-cabrera.com │ odawson45@malone-cabrera.com │ 4 │ 0.0038022813688212928 │ 0.0038022813688212928 │ 8.059112052696065 │ -0.9002626842225219 │ 0 │
│ 19.878366639943387 │ 0.9999989624359255 │ 552 │ 553 │ Riley │ Riley │ 4 │ 0.006016847172081829 │ 0.006016847172081829 │ 6.309869029835824 │ 0.040393298600093885 │ Kiing │ King │ 3 │ 0.001221001221001221 │ 0.01221001221001221 │ 6.543833593077295 │ 0.0 │ 1984-04-19 │ 1984-04-19 │ 2 │ 8.996493872732788 │ Aberdeen │ NULL │ -1 │ 0.01968019680196802 │ NULL │ 0.0 │ 0.0 │ riley.k@zavala.com │ riley.k@zavala.com │ 4 │ 0.0063371356147021544 │ 0.0063371356147021544 │ 8.059112052696065 │ -1.6372282783887275 │ 2 │
│ 6.871025192534534 │ 0.9915292696875624 │ 266 │ 267 │ Henry │ Hnry │ 3 │ 0.010830324909747292 │ 0.0024067388688327317 │ 6.3086382484117 │ 0.0 │ Simmons │ NULL │ -1 │ 0.001221001221001221 │ NULL │ 0.0 │ 0.0 │ 2019-01-31 │ 2019-01-31 │ 2 │ 8.996493872732788 │ Leeds │ NULL │ -1 │ 0.020910209102091022 │ NULL │ 0.0 │ 0.0 │ hsimmons@robebrts.com │ NULL │ -1 │ 0.0012674271229404308 │ NULL │ 0.0 │ 0.0 │ 2 │
└────────────────────┴────────────────────┴─────────────┴─────────────┴──────────────┴──────────────┴──────────────────┴───────────────────────┴───────────────────────┴───────────────────┴──────────────────────┴───────────┴───────────┴───────────────┴──────────────────────┴──────────────────────┴───────────────────┴─────────────────────┴────────────┴────────────┴───────────┴───────────────────┴───────────┴───────────┴────────────┴───────────────────────┴─────────────────────┴─────────────────────┴────────────────────┴──────────────────────────────┴──────────────────────────────┴─────────────┴───────────────────────┴───────────────────────┴───────────────────┴─────────────────────┴───────────┘
This is a good way to de-risk a large job: if a single chunk runs successfully, you know the whole pipeline works, and you can simply iterate over every (left_chunk, right_chunk) combination to produce the complete result set.
Running that loop yourself and concatenating the outputs is exactly what predict(num_chunks_left=..., num_chunks_right=...) does for you — so for a job that runs on a single machine, prefer the chunked predict() shown above. The value of predict_chunk() is that each chunk is fully independent, which is what makes the distributed workflow in the next section possible.
Distributing across multiple machines with predict_chunk()¶
Because every predict_chunk() call is completely self-contained, you can score different chunks on different machines — even with DuckDB, which is a single-node engine (i.e. runs on a single computer, not a cluster). This is how you scale a single linkage job horizontally: each machine loads the model and input data, scores its assigned chunk(s) with predict_chunk(), and writes the result to shared storage. A coordinator then unions the per-chunk outputs into the final result.
Note there is no automatic scheduler — you have to trigger the per-chunk jobs and combine their outputs yourself.
A typical distributed workflow looks like this:
- Coordinator: save the trained model to shared storage so that every worker loads an identical model.
- Workers: each machine loads the model and input data, calls
predict_chunk()for its assigned(left_chunk, right_chunk)slice(s), and writes the scored output to shared storage (e.g. parquet on cloud storage). - Coordinator: collect and concatenate the per-chunk prediction outputs.
Below we simulate this on a single machine. First, the coordinator saves the model so that workers can load an identical copy:
import os
import tempfile
work_dir = tempfile.mkdtemp()
model_path = os.path.join(work_dir, "model.json")
# The coordinator saves the trained model so every worker loads an identical model
linker.misc.save_model_to_json(model_path, overwrite=True)
{'link_type': 'dedupe_only',
'probability_two_random_records_match': 0.002882882882882883,
'retain_matching_columns': True,
'retain_intermediate_calculation_columns': True,
'additional_columns_to_retain': [],
'sql_dialect': 'duckdb',
'linker_uid': 'qeu2gc5b',
'em_convergence': 0.0001,
'max_iterations': 25,
'match_weight_column_prefix': 'mw_',
'term_frequency_adjustment_column_prefix': 'tf_',
'comparison_vector_value_column_prefix': 'gamma_',
'unique_id_column_name': 'unique_id',
'source_dataset_column_name': 'source_dataset',
'blocking_rules_to_generate_predictions': [{'blocking_rule': '(l."first_name" = r."first_name") AND (l."city" = r."city")',
'sql_dialect': 'duckdb'},
{'blocking_rule': 'l."surname" = r."surname"', 'sql_dialect': 'duckdb'},
{'blocking_rule': 'l."dob" = r."dob"', 'sql_dialect': 'duckdb'}],
'comparisons': [{'output_column_name': 'first_name',
'comparison_levels': [{'sql_condition': '"first_name_l" IS NULL OR "first_name_r" IS NULL',
'label_for_charts': 'first_name is NULL',
'fix_m_probability': False,
'fix_u_probability': False,
'is_null_level': True},
{'sql_condition': '"first_name_l" = "first_name_r"',
'label_for_charts': 'Exact match on first_name',
'm_probability': 0.4908947887649066,
'u_probability': 0.006187690390473553,
'fix_m_probability': False,
'fix_u_probability': False,
'tf_adjustment_column': 'first_name',
'tf_adjustment_weight': 1.0},
{'sql_condition': 'jaro_winkler_similarity("first_name_l", "first_name_r") >= 0.92',
'label_for_charts': 'Jaro-Winkler distance of first_name >= 0.92',
'm_probability': 0.15160172798813126,
'u_probability': 0.0019125588479645527,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'jaro_winkler_similarity("first_name_l", "first_name_r") >= 0.88',
'label_for_charts': 'Jaro-Winkler distance of first_name >= 0.88',
'm_probability': 0.07398853217357397,
'u_probability': 0.0013500415397396843,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'jaro_winkler_similarity("first_name_l", "first_name_r") >= 0.7',
'label_for_charts': 'Jaro-Winkler distance of first_name >= 0.7',
'm_probability': 0.07905043858131171,
'u_probability': 0.017983245638327334,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'ELSE',
'label_for_charts': 'All other comparisons',
'm_probability': 0.2044645124920764,
'u_probability': 0.9725664635834949,
'fix_m_probability': False,
'fix_u_probability': False}],
'comparison_description': 'NameComparison'},
{'output_column_name': 'surname',
'comparison_levels': [{'sql_condition': '"surname_l" IS NULL OR "surname_r" IS NULL',
'label_for_charts': 'surname is NULL',
'fix_m_probability': False,
'fix_u_probability': False,
'is_null_level': True},
{'sql_condition': '"surname_l" = "surname_r"',
'label_for_charts': 'Exact match on surname',
'm_probability': 0.43412147124264217,
'u_probability': 0.004537515363786746,
'fix_m_probability': False,
'fix_u_probability': False,
'tf_adjustment_column': 'surname',
'tf_adjustment_weight': 1.0},
{'sql_condition': 'jaro_winkler_similarity("surname_l", "surname_r") >= 0.92',
'label_for_charts': 'Jaro-Winkler distance of surname >= 0.92',
'm_probability': 0.22504393958141566,
'u_probability': 0.0024119990019314474,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'jaro_winkler_similarity("surname_l", "surname_r") >= 0.88',
'label_for_charts': 'Jaro-Winkler distance of surname >= 0.88',
'm_probability': 0.04427190114778548,
'u_probability': 0.0013862063229491078,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'jaro_winkler_similarity("surname_l", "surname_r") >= 0.7',
'label_for_charts': 'Jaro-Winkler distance of surname >= 0.7',
'm_probability': 0.07778284167837138,
'u_probability': 0.014222476873457846,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'ELSE',
'label_for_charts': 'All other comparisons',
'm_probability': 0.21877984634978528,
'u_probability': 0.9774418024378748,
'fix_m_probability': False,
'fix_u_probability': False}],
'comparison_description': 'NameComparison'},
{'output_column_name': 'dob',
'comparison_levels': [{'sql_condition': 'CAST("dob_l" AS TEXT) IS NULL OR CAST("dob_r" AS TEXT) IS NULL',
'label_for_charts': 'transformed dob is NULL',
'fix_m_probability': False,
'fix_u_probability': False,
'is_null_level': True},
{'sql_condition': 'CAST("dob_l" AS TEXT) = CAST("dob_r" AS TEXT)',
'label_for_charts': 'Exact match on transformed dob',
'u_probability': 0.0018599835094245557,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'levenshtein(CAST("dob_l" AS TEXT), CAST("dob_r" AS TEXT)) <= 1',
'label_for_charts': 'Levenshtein distance of transformed dob <= 1',
'u_probability': 0.0014764817549040286,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'ELSE',
'label_for_charts': 'All other comparisons',
'u_probability': 0.9966635347356714,
'fix_m_probability': False,
'fix_u_probability': False}],
'comparison_description': 'LevenshteinAtThresholds'},
{'output_column_name': 'city',
'comparison_levels': [{'sql_condition': '"city_l" IS NULL OR "city_r" IS NULL',
'label_for_charts': 'city is NULL',
'fix_m_probability': False,
'fix_u_probability': False,
'is_null_level': True},
{'sql_condition': '"city_l" = "city_r"',
'label_for_charts': 'Exact match on city',
'm_probability': 0.5557346018302922,
'u_probability': 0.11090275917492633,
'fix_m_probability': False,
'fix_u_probability': False,
'tf_adjustment_column': 'city',
'tf_adjustment_weight': 1.0},
{'sql_condition': 'ELSE',
'label_for_charts': 'All other comparisons',
'm_probability': 0.4442653981697078,
'u_probability': 0.8890972408250737,
'fix_m_probability': False,
'fix_u_probability': False}],
'comparison_description': 'ExactMatch'},
{'output_column_name': 'email',
'comparison_levels': [{'sql_condition': '"email_l" IS NULL OR "email_r" IS NULL',
'label_for_charts': 'email is NULL',
'fix_m_probability': False,
'fix_u_probability': False,
'is_null_level': True},
{'sql_condition': '"email_l" = "email_r"',
'label_for_charts': 'Exact match on email',
'm_probability': 0.5433412103558991,
'u_probability': 0.0020372211654850424,
'fix_m_probability': False,
'fix_u_probability': False,
'tf_adjustment_column': 'email',
'tf_adjustment_weight': 1.0},
{'sql_condition': 'NULLIF(regexp_extract("email_l", \'^[^@]+\', 0), \'\') = NULLIF(regexp_extract("email_r", \'^[^@]+\', 0), \'\')',
'label_for_charts': 'Exact match on username',
'm_probability': 0.22359074256583447,
'u_probability': 0.0008916219955040636,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'jaro_winkler_similarity("email_l", "email_r") >= 0.88',
'label_for_charts': 'Jaro-Winkler distance of email >= 0.88',
'm_probability': 0.23306804623841282,
'u_probability': 0.0008808144561646205,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'jaro_winkler_similarity(NULLIF(regexp_extract("email_l", \'^[^@]+\', 0), \'\'), NULLIF(regexp_extract("email_r", \'^[^@]+\', 0), \'\')) >= 0.88',
'label_for_charts': 'Jaro-Winkler >0.88 on username',
'u_probability': 0.0006106259726785405,
'fix_m_probability': False,
'fix_u_probability': False},
{'sql_condition': 'ELSE',
'label_for_charts': 'All other comparisons',
'm_probability': 8.398535634403338e-10,
'u_probability': 0.9955797164101677,
'fix_m_probability': False,
'fix_u_probability': False}],
'comparison_description': 'EmailComparison'}]}
Now a worker — which could be a completely separate machine — loads the model and input data, scores its assigned chunk with predict_chunk(), and writes the result to shared storage:
# On a worker machine: rebuild the linker from the shared model and input data
db_api_worker = DuckDBAPI()
df_sdf_worker = db_api_worker.register(df)
linker_worker = Linker(df_sdf_worker, model_path)
# Score just this worker's assigned chunk and write the result to shared storage
chunk_path = os.path.join(work_dir, "predictions_chunk_1_1.parquet")
chunk_predictions = linker_worker.inference.predict_chunk(
left_chunk=(1, 4),
right_chunk=(1, 4),
threshold_match_probability=0.9,
)
chunk_predictions.as_duckdbpyrelation().to_parquet(chunk_path)
chunk_count = chunk_predictions.as_duckdbpyrelation().count("*").fetchone()[0]
print(f"Worker scored {chunk_count} pairs and wrote them to {chunk_path}")
Blocking time: 0.01 seconds
Predict time (post-blocking): 0.12 seconds
-- WARNING --
You have called predict(), but there are some parameter estimates which have neither been estimated or specified in your settings dictionary. To produce predictions the following untrained parameters will use default values.
Comparison: 'dob':
m values not fully trained
Comparison: 'email':
m values not fully trained
Worker scored 66 pairs and wrote them to /tmp/tmproclzfem/predictions_chunk_1_1.parquet
In a real deployment you would run that worker step once per chunk, each on its own machine, and then union the per-chunk parquet outputs to obtain the full set of predictions.
Niche: separating blocking from scoring with compute_blocked_pairs_for_predict_chunk()
The workflow above scores a whole chunk in a single predict_chunk() call, which is all that most distributed jobs ever need. For the very largest jobs, you may wish to split blocking into separate jobs in your DAG, and run predictions separately. To do so you do the following:
compute_blocked_pairs_for_predict_chunk(left_chunk=..., right_chunk=...)materialises only the blocked pairs (the candidate record-id pairs) for a chunk, without scoring them, which you persist to shared storage.- A worker later reads those pairs, registers them with
register_blocked_pairs_for_predict(), and callspredict()to score exactly that table — no re-blocking. (Once blocked pairs are registered this way, the chunking argumentsnum_chunks_left/num_chunks_rightandpredict_chunk()are unavailable, because the pairs are already materialised.)
This is a niche technique — reach for it only when scoring a single chunk with predict_chunk() is itself too large.
Summary: when to use each technique¶
| Technique | Where | Use when |
|---|---|---|
record_sample_proportion |
Blocking analysis, estimate_probability_two_random_records_match |
Exact counts over the full dataset are too slow |
max_pairs |
estimate_u_using_random_sampling, estimate_parameters_using_expectation_maximisation |
Training passes generate too many pairs to process quickly |
num_chunks_left / num_chunks_right on predict() |
Inference | predict() is running out of memory, or you want progress reporting |
predict_chunk() |
Inference | Verifying the pipeline on one slice, or distributing a job across machines by scoring one chunk per worker |
compute_blocked_pairs_for_predict_chunk() + register_blocked_pairs_for_predict() |
Inference | Niche: separating blocking from scoring when even a single chunk is too large to score in one pass |
Remember the guiding principle: start with the defaults and only adopt these techniques when a specific step becomes too slow or runs out of memory. Sampling costs you a little accuracy, and chunking adds operational complexity, so neither is worth it while everything is running quickly.
Further Reading
For more on performance and scaling, see the Performance Topic Guides.
For the full API, see the Blocking analysis, Training and Inference API documentation.