Real-Time Matching Tutorial¶

Version 1.0, August 3, 2026

This tutorial describes how to match records in real time with madmatcher-pro: instead of running a batch job over two whole tables, you match one incoming record (or a live stream of them) against a fixed set of your own records the moment it arrives. We first discuss how real-time matching works and how it relates to the batch pipeline you may already know from Sparkly, Delex, Semantic blocking, and MatchFlow, and then provide a step-by-step guide that walks you through publishing a serving bundle, loading it, and every way you can drive requests through it. You should read the “How Real-Time Matching Works” section first, because it defines the terms used later.

Real-time matching is a separately-licensed add-on. RealtimeMatcher.load checks that your license includes the realtime entitlement and raises a LicenseError without it; a base (batch) license does not grant it. Contact support@madmatcher.ai to enable it.

How Real-Time Matching Works¶

The batch pipeline answers “which pairs across these two tables are the same entity?” by blocking, featurizing, and predicting over both tables at once on Spark. Real-time matching answers a narrower, faster question: “which of my known records does this one incoming record match?”, and answers it immediately. It runs the same three steps (block, featurize, predict) for a single record, from state prepared once and loaded into memory.

Three ideas carry the whole design:

Build once, offline. Everything that depends only on your corpus, the search index, the preprocessed corpus features, and the trained model, is produced by an ordinary batch job and frozen into a bundle on disk. This is the expensive part, and it happens away from the request path.
Load resident. A serving process loads the bundle into memory once. From then on the index, the corpus feature state, and the model live in the process.
Answer per record. For each incoming record the matcher blocks it against the corpus (retrieving a few candidates), featurizes each (record, candidate) pair with the exact batch featurizer, and predicts. Because the kernels are the batch kernels, a served prediction equals the batch prediction for the same record and corpus.

We reuse the terminology of the other tutorials, adapted to one-record-at-a-time serving:

The corpus is the fixed set of records you match against. It plays the role of the batch indexing table (table A): the bundle is built over it once, and every incoming record is compared to it.
An incoming record is the record you want to match, arriving live or from a stream. It plays the role of one row of the batch search table (table B).
A candidate is a corpus record that blocking retrieves as a possible match.
A match is a candidate the model predicts is the same entity.
A bundle is the on-disk package a serving process loads: the index, the corpus feature state, the model, and a manifest that pins the configuration so serving cannot drift from training.

For this tutorial we assume a corpus of catalog products you already matched and cleaned in a batch job, and incoming product records (new listings, supplier feeds, marketplace items) that you match against it as they arrive.

Your corpus (matched against)

_id	name	brand	price
0	Sony WH-1000XM5 headphones	Sony	399
1	Logitech MX Master 3S	Logitech	99
…	…	…	…

An incoming record (matched live)

name	brand	price
Sony wireless over-ear noise-cancelling headset	Sony	349

Matching that incoming record returns the corpus records the model predicts it is the same as:

Column	Type	Meaning
`id1`	long	the matched corpus record’s id
`id2`	long or string	the incoming record’s key: its own id (from `id_col`), or a correlation key you pass
`prediction`	float	`1.0` on every row (the frame holds only predicted matches)
`confidence`	float	the model’s confidence for that pair, highest first

The result is a pandas.DataFrame, already filtered to predicted matches. An incoming record with no match returns an empty frame with those same columns, so “no match” is distinct from “not processed”.

One typing difference from batch is deliberate. In batch, both tables are indexed, so both id columns must be 32- or 64-bit integers; id1 keeps that rule here, because the corpus is still an indexed table. The incoming record, though, is never indexed: id2 only identifies the request in its results, so it is not restricted to an integer. An integer id and a string key (a request id, a queue offset, a UUID) both work; pick one kind and use it consistently within a run, because a sink writing to typed storage such as parquet cannot store a column that mixes the two.

id1 and id2 are the result frame’s column names only (the same pair naming batch featurize uses); no input needs a column named either. The corpus carries its ids in a column named _id (Step 1.1 covers this requirement), and an incoming record carries its own id, when it has one, in the column named by id_col (default _id as well). Serving maps them into id1 and id2 on the way out.

The choices you make¶

Standing up serving comes down to a few independent choices, each covered in the steps below:

Which blocker the bundle carries (Step 1): Sparkly (lexical), Delex (lexical), Semantic (dense), a fusion of one lexical blocker plus Semantic, or none at all (you supply candidates per request).
Whether the model uses the semantic cosine feature (Step 1): if you trained with it, serving reproduces the per-candidate cosine from the semantic blocker’s scores, from a stored corpus embedding table, or both.
How an incoming record is embedded (Steps 1 and 2), whenever the bundle blocks semantically or reproduces the cosine feature: in-process by a resident embedder (you publish the same provider you used in batch), or attached to the request as a precomputed probe_embedding (which is required when the bundle carries no embedder).
Whether the matcher is sklearn or SparkML (Step 1): sklearn/xgboost predicts in-process with no Spark on the request path; SparkML runs one Spark job per request.
How you drive requests (Steps 2 to 8): a single call, a batch, a concurrent pool, or an orchestrated stream, plus how you hand off from a batch build and refresh the corpus over time.

Durability and Progress Tracking¶

Real-time matching is one of madmatcher-pro’s premium features. Two capabilities apply to the operations that run long or must not drop records, both opt-in:

Durability. The orchestrated streaming runner has a durable form (run_matching_durable) that records per-record completion, so a crashed run resumes where it stopped, and several workers pointed at the same body of incoming records can divide it between them with no broker or queue. Publishing a bundle can likewise be made crash-resumable, and the batch-to-serving handoff spools arriving records durably so none is lost while the build runs.
Progress tracking. Every serving surface can report live throughput to the same web dashboard the batch engines use (track=True), and a publish shows up as a job while it runs.

The Reference: Durability and Progress Tracking section at the end covers the claim protocol, the delivery guarantee, the dashboard, and every argument in full.

Step-by-Step Walkthrough¶

This section walks through writing serving into your application, in the following steps:

Publishing a Serving Bundle
Loading the Matcher and Matching One Record
Matching a Batch
Serving Concurrent Requests
Streaming Records Through a Matcher
Durable and Multi-Worker Streaming
Handing Off From a Batch Build
Refreshing the Bundle Without Downtime

Steps 1 and 2 are the minimal path: publish a bundle, then match against it. Steps 3 to 8 add the modes you reach for as your workload grows: batches, concurrency, streaming, durability, and live corpus refresh. Read Steps 1 and 2 first; the rest are independent and you can jump to the one you need.

Step 1: Publishing a Serving Bundle¶

Everything the real-time matcher serves from is prepared in this step, in two parts. In Step 1.1 you produce the batch artifacts serving reuses: a blocker built over your corpus, and a trained matcher. These are made with the engines the other tutorials teach, so this part tells you what to build, which tutorial shows how, and what to keep when you are done. In Step 1.2 you assemble those artifacts into a serving bundle with one call, publish_serving_artifacts. The result is a self-contained folder on disk; every later step starts by loading it.

Step 1.1: Building the Batch Artifacts¶

Entity matching runs in two steps: blocking, which finds, for a record, a small set of likely matches among your corpus records (its candidates), and matching, which predicts which of those candidates truly match. The real-time matcher performs both steps for each incoming record, so the bundle carries an artifact for each, and you have options on both sides.

The blocking side. Choose what will find candidates for each incoming record:

Sparkly finds candidates by keyword similarity (BM25) over an inverted index built on your corpus. To review how Sparkly works and build the index, see the Sparkly tutorial; its Step 3 builds the index. Keep the index directory when you are done (this guide uses build/lucene); publishing copies it into the bundle.
Delex finds candidates with blocking rules you write. To review how Delex works and construct a program, see the Delex tutorial; its Step 3 constructs the BlockingProgram. Keep the program object and the corpus DataFrame it applies to; there is no index directory to keep, because serving stores both in the bundle and rebuilds Delex’s state when the bundle is loaded.
Semantic finds candidates by meaning, comparing embedding vectors. To review how semantic blocking works, embed your corpus, and build the index, see the semantic tutorial; its Steps 3 and 4 embed the records and build the index. Keep three things: the index directory, the embedding provider you used (for example the SentenceTransformerProvider and its model id), and the list of fields you embedded. Serving reads the default (flat) index; an index built with a compression codec is refused at publish.
Bring your own candidates. Build no blocker at all. Each request then passes the record’s candidate ids itself (match(record, candidates=[...]), shown in Step 2), for when another system already produces candidates. Nothing to keep.

A bundle can carry Sparkly or Delex (not both), can add semantic beside either one (their candidates are then fused per record, as in batch fusion), can carry semantic alone, or can carry no blocker at all.

One id column ties everything together. Your corpus must carry a unique 32- or 64-bit integer id column named _id: the feature state serving loads is keyed by that exact name, and a result’s id1 is one of its values. Build every blocker over that same column (IndexConfig(id_col="_id") for Sparkly, delex_id_col for Delex, the semantic index’s id_col), so the ids a blocker retrieves are the ids the feature state and your corpus know. id1 and id2 exist only in the result frame; no input table needs a column with those names.

If you bring your own embeddings. Semantic blocking and the cosine feature (Step 1.2) work with embeddings in two shapes:

The corpus embedding table: one row per corpus record, holding the record’s id and its vector, as an id column plus an array-of-float column with unique ids. This is exactly the table create_embeddings writes in the semantic tutorial’s Step 3, so if you followed that tutorial you already have it. A table you embedded outside MadMatcher works too, in that same shape.
A per-request probe embedding: one plain float vector (for example a numpy array) for one incoming record, produced by the same model as the corpus vectors. Step 2 shows where a request passes it.

The matching side. The matcher is a model trained with MatchFlow; the MatchFlow technical guide covers that whole flow: creating features, featurizing candidate pairs, getting labels (from labeled data you already have, from seeds, or through active learning, with or without down-sampling), and training. Whichever route you take, serving needs three things out of that run:

The trained model: a SKLearnModel (wrapping an sklearn or xgboost estimator) or a SparkMLModel.
The feature list that create_features produced and featurize used. Serving uses the same list to build each incoming pair’s feature vector exactly as training did.
The fill_na value you passed to featurize. Serving fills missing feature values with the same number, and it cannot be recovered from the model afterwards, so note it down.

Carrying the artifacts to the publish script. If you publish in the same script that finishes the build, hold the corpus, the features, and the model in memory and continue to Step 1.2. If you build and publish in separate scripts, MatchFlow has a save/load helper for each artifact. In the build script:

from madmatcher_pro.matchflow import save_features, save_dataframe

model.save("build/model")
save_features(features, "build/features.pkl")
save_dataframe(corpus, "build/corpus")

And in the publish script:

from madmatcher_pro.matchflow import MLModel, load_features, load_dataframe

model = MLModel.load("build/model")
features = load_features("build/features.pkl")
corpus = load_dataframe("build/corpus", "sparkdf")

The reloaded model predicts identically to the one you trained. The index directories are already on disk where you built them (publishing reads them by path), and a Delex BlockingProgram is a plain Python object you can rebuild in the publish script or pickle alongside the rest.

Step 1.2: Publishing the Bundle¶

With the artifacts in hand, one call assembles the bundle. publish_serving_artifacts takes the output folder plus four required arguments:

out_dir (the first, positional argument) is the folder the bundle is written into. Use a fresh folder per publish (more on this below).
indexed_table is the corpus as a Spark DataFrame: the records every incoming record is matched against.
features is the feature list from Step 1.1.
model is the trained matcher from Step 1.1.
fill_na is the value you gave featurize in training. It is required and must match exactly: a different value silently changes missing-value features at serve time, and it cannot be validated against the model, so publishing refuses to guess it.

The blocker is passed alongside, per option (each shown below). A minimal Sparkly-blocked publish:

from madmatcher_pro.serving import publish_serving_artifacts

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus,                 # the corpus (a Spark DataFrame)
    features=features,                    # the feature list from training
    model=model,                          # the trained model
    sparkly_index_path="build/lucene",    # your built Sparkly index directory
    fill_na=0.0,                          # the same value you gave featurize()
)

This writes a self-contained bundle at bundles/v1 and returns its path. The _PUBLISHED.json marker is written last, so a crash mid-publish leaves an incomplete bundle the loader refuses rather than a half-written one it would serve.

Publish a new directory per version. publish_serving_artifacts rewrites out_dir in place. That is safe for a fresh directory, but never publish over a bundle a RealtimeMatcher is actively serving (a live reader holds the corpus memory-map open). To refresh a live process, publish the new version to its own directory and swap it in (Step 8).

Two optional arguments matter to every bundle:

id_col (default "_id") names the column an incoming record carries its own id in. Serving stamps that id onto each result as id2 when a request passes no explicit key, so a served row always says which record produced it.
block_limit (default 50) caps how many candidates blocking retrieves per incoming record, which is how many pairs the model scores. It is fixed in the bundle at publish.

If you are deduplicating, add one. When the corpus contains the records you serve, every blocker returns each record’s own row at rank 1, and that self-match occupies one of the block_limit slots. The batch equivalent asked for K+1 candidates and dropped the self pair, so publish block_limit=K+1 to get the same K real candidates, and drop id1 == id2 from the results yourself (serving has no self-match exclusion). Getting this wrong is not an error, just one candidate fewer per record, which surfaces later as a small recall gap against the batch run. block_limit is read from MANIFEST.json at load and used nowhere else, so an already-published bundle can be corrected by editing that one field instead of re-publishing.

The rest of this step shows the publish call for each blocking option from Step 1.1, then the cosine feature, then the two model kinds.

Publishing when the driver is not the only machine¶

The bundle is assembled as ordinary files on the driver. Two things follow from that.

Where the bundle sits decides whether a DataFrame argument works. embedding_store and delex_corpus are the two arguments that can be Spark DataFrames, and Spark executes a DataFrame write on the executors, against the destination path as each of them resolves it:

The bundle is on storage every node shares — a single machine, a network mount (NFS and friends), a lakehouse volume. The executors write exactly where the driver reads, so passing a DataFrame is fine and needs nothing extra.
The bundle is on the driver’s own disk while the executors are elsewhere — the usual cluster shape. Each executor writes to its local disk under that same path, and the bundle gets nothing. Publishing verifies the driver can read each Spark-written artifact back, so this fails naming the cause rather than surfacing steps later as a parquet schema error.

The question is where the bundle is, not whether the cluster has a distributed filesystem: a file:// write ignores fs.defaultFS entirely, so having HDFS or GCS available does not by itself make a DataFrame argument work.

For the second shape, stage the table once to a filesystem both sides can see and pass that URI. Both arguments accept a path or URI on any filesystem Spark can address (hdfs://, s3a://, gs://, abfss://, or a local/shared-mount path), and publishing pulls it into the bundle itself with no Spark write:

# once: written by the executors, to storage the driver can also read
embeddings.write.parquet("gs://bucket/staging/embeddings")
corpus.write.parquet("gs://bucket/staging/corpus")

publish_serving_artifacts(
    "bundles/v1", …,
    embedding_store="gs://bucket/staging/embeddings",   # copied into the bundle, no Spark write
    delex_corpus="gs://bucket/staging/corpus",
)

Publish in its own Spark application. A batch featurize run registers its preprocessed corpus with the SparkContext (addFile), and Spark ships every registered file to all executors of every later job in that context. On a large corpus that is tens of gigabytes per executor for jobs that never read it, enough to fill the executors’ disks and get the nodes marked unhealthy. Publishing itself registers nothing, but it cannot undo a registration a preceding job made, so run it as its own application rather than as a tail stage of a matching job.

Option 1: Sparkly (lexical)¶

The example above. Pass sparkly_index_path= pointing at a built LuceneIndex directory. This is the default choice and needs no Spark at serve time.

Option 2: Delex (lexical)¶

Delex has no cross-process resident form, so the bundle stores the raw corpus and the (unbuilt) BlockingProgram, and the loader rebuilds the resident probe once at load. Fusing Delex therefore makes the loader require a SparkSession, and the rebuild is a one-time Spark job at load, not per request.

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    delex_program=program,          # a delex BlockingProgram (see the Delex tutorial)
    delex_corpus=corpus,            # a DataFrame or a parquet path/URI (defaults to indexed_table)
    delex_id_col="_id",
)

Option 3: Semantic, and fusion¶

Semantic blocking retrieves candidates by meaning. Publish it by pointing at a built flat SemanticIndex (a lossy-codec index is refused; serving decodes full vectors only) and giving serving the embedder and the fields it embedded, so it can turn each incoming record into a query vector identically to the corpus:

from madmatcher_pro.semantic import SentenceTransformerProvider

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    sparkly_index_path="build/lucene",              # lexical half of the fusion
    semantic_index_path="build/semantic",           # a built flat SemanticIndex
    embed_provider=SentenceTransformerProvider(model_id="BAAI/bge-small-en-v1.5"),
    embed_fields=["name", "brand"],                 # the fields the corpus was embedded from
    nprobe=32,
    fusion_weights={"sparkly": 0.5, "semantic": 0.5},
)

With both a lexical index and a semantic index, serving fuses them per record with the same weighted reciprocal-rank fusion the batch pipeline uses (fusion_weights and fusion_kappa). Publish the semantic index alone (omit sparkly_index_path) for semantic-only blocking. The embedder is stored as config, not weights; the loader rebuilds one resident embedder that the semantic probe and the cosine store share, and it loads its model lazily on the first request.

The embed_provider can be any provider you used for the batch embeddings (a SentenceTransformerProvider, FastEmbedProvider, OpenAIEmbeddingProvider, or a model-id string). Pass embed_provider=None to require a precomputed probe_embedding on every match call instead of embedding in-process.

Option 4: Matching-only (bring your own candidates)¶

Omit every blocker to publish a bundle that only featurizes and predicts. You then supply the candidate ids on each request (match(record, candidates=[...]), Step 2). Use this when another system already produces candidates.

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    # no sparkly_index_path / semantic_index_path / delex_program
)

The cosine feature¶

If you trained the model with the semantic cosine appended as its last feature (see “optional cosine feature for matching” in the Semantic tutorial), serving must reproduce that feature per candidate. Set cosine_feature=True and give it a cosine source:

semantic_index_path lets serving reuse the semantic blocker’s own cosine for each candidate it ranked.
embedding_store (a corpus embedding table with an id column and an embedding column: a parquet path or URI, or a Spark DataFrame — see “publishing when the driver is not the only machine” above) lets serving recompute cosine, fetching a candidate’s stored vector and dotting it with the incoming record’s embedding. This covers candidates a lexical-only blocker surfaced that the semantic blocker never ranked.
Both is the most faithful: serving reuses the semantic score where it has one and recomputes only the gaps.

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    sparkly_index_path="build/lucene",
    semantic_index_path="build/semantic",
    embed_provider=SentenceTransformerProvider(model_id="BAAI/bge-small-en-v1.5"),
    embed_fields=["name", "brand"],
    cosine_feature=True,                    # the model was trained with the cosine feature
    embedding_store="build/embeddings",     # recompute cosine for candidates the scores miss
)

The embedding store is validated at publish time (each corpus id must map to exactly one vector), so a bad table fails now rather than at serve time. You can validate one up front with check_embedding_store(table, id_col="_id", embedding_col="embedding").

Both cosine sources also need the incoming record’s own embedding. Serving produces it in-process with the resident embedder you published (embed_provider, the same provider class you used in batch), so a plain match(record) needs nothing extra, and that one embedding serves both semantic blocking and the cosine feature. To skip in-process embedding, attach a precomputed vector with match(record, probe_embedding=...); publishing with embed_provider=None makes that mandatory, so every call must carry its own probe_embedding.

sklearn vs SparkML¶

The model kind decides the request path, and publish records it in the manifest:

An sklearn/xgboost model (SKLearnModel) predicts in-process with no Spark on the request path. This is the fast path and what real-time serving is built for.
A SparkML model (SparkMLModel) has no in-process predict, so serving runs one Spark job per match call. It is supported for batch interoperability (a bundle that must match your SparkML training exactly), but it is slower and the loader needs a SparkSession.

Publish validates the model against the pinned feature set at load: the trained model must expect exactly as many features as the bundle produces (the feature list, plus one when cosine_feature=True), so a model trained on a different feature list is caught cleanly before it ever serves.

Step 2: Loading the Matcher and Matching One Record¶

RealtimeMatcher.load is the only supported way to create a matcher. It checks the realtime entitlement, then loads the bundle into resident state:

from madmatcher_pro.serving import RealtimeMatcher

matcher = RealtimeMatcher.load("bundles/v1")

result = matcher.match({"name": "Sony wireless over-ear noise-cancelling headset",
                        "brand": "Sony", "price": 349},
                       key="req-7")
print(result)
#    id1     id2  prediction  confidence
# 0    0   req-7         1.0        0.98

load takes:

artifacts_dir is the bundle from Step 1 (a local path, or a Spark URI such as s3://... for the artifacts Spark reads).
spark (default None) defaults to the active or a new session, and is needed only for a Delex or SparkML bundle. A resident sklearn bundle loads without it.
track (default False) shows this matcher’s throughput on the dashboard.
id_col overrides which column supplies a record’s own id; it wins over what the bundle recorded.

Loading is where the gate and the heavy lifting live: load checks the license and the realtime entitlement (a LicenseError here means the base license is valid but the add-on is not enabled), then opens the index, maps the corpus feature state into memory, and loads the model, so match itself does none of that. One exception is deferred: a bundle with a resident embedder loads the embedding model’s weights on the first request, so the first match is slower than the rest. If first-request latency matters, serve one throwaway record right after load.

match takes the incoming record (a dict, a pandas.Series, a one-row pandas.DataFrame, or a pyspark Row) and returns the [id1, id2, prediction, confidence] frame. Its keyword arguments:

key is the value stamped as id2 on every result row, and it can be an integer or a string. Resolution is first-hit-wins: an explicit key, else the record’s own id under id_col, else None. Use an explicit key for a record with no id of its own, for example a string request id for a record pulled off a queue, and it is echoed back on every match so the result stays attributable once it leaves the call.
candidates is the bring-your-own list for a matching-only bundle (Option 4): when given, blocking is skipped and the record is matched only against those corpus ids. An id absent from the corpus is dropped; an empty or all-unknown list gives the empty result.
probe_embedding is a precomputed embedding for the semantic source or cosine store, skipping the resident embedder for this call. It is required when the bundle was published with embed_provider=None.

What fields should the incoming record carry? The columns your features compare, which are the corpus columns you trained over (the record in the example carries name, brand, and price because the corpus has them). It does not need all of them: a missing field is treated as a null value, exactly as a stored null in the corpus is, so the features that compare it fall back to fill_na instead of raising. Fields the features never read are ignored, and each value is cast to the corpus column’s type before comparing, so an incoming 1999 where the corpus stores text (or "1999" where it stores a number) still compares consistently.

Reading the result. The trained model decides what is a match: prediction is its call and confidence its probability for that call, computed by the same model call the batch apply_matcher uses. The frame holds the candidates the model predicts as matches, sorted by confidence descending (ties broken by id1, so results are identical run to run). A result carries the matched record’s id, not its fields: id1 is a key into your own corpus, so join it back to your corpus table when a consumer needs the record itself. On the streaming paths (Step 5), the incoming record’s fields can be stored beside each match with passthrough.

Release the matcher when done. A matcher holds Lucene readers and memory-maps open. Call matcher.close() when you are finished, or before loading a new one:

matcher.close()

Step 3: Matching a Batch¶

When you have several records in hand, match_batch is faster than calling match on each: a batch shares one embedder forward pass and one model predict (one Spark job, for a SparkML model). The results are exactly what per-record match would give, one frame per input record, in order:

results = matcher.match_batch([
    {"name": "Sony wireless noise-cancelling headset", "brand": "Sony"},
    {"name": "wireless ergonomic mouse, USB-C", "brand": "Logitech"},
], keys=["req-1", "req-2"])

for res in results:            # each is a [id1, id2, prediction, confidence] frame
    print(res)

match_batch takes:

records is a list of record mappings (each any shape match accepts) or a pandas.DataFrame whose rows are the records. An empty input returns an empty list.
keys (optional) is a sequence aligned one-to-one with records, each entry as match’s key. Where an entry is None (or the whole argument is omitted), that record falls back to its own id under id_col.
candidates (optional) is an aligned sequence: each entry a bring-your-own candidate list for that record, or None to block that record as usual.
probe_embeddings (optional) is an aligned sequence of precomputed embeddings, each entry as match’s probe_embedding.

It returns a list of [id1, id2, prediction, confidence] frames, one per input record, in input order; each frame is exactly what match would have returned for that record on its own.

Step 4: Serving Concurrent Requests¶

Under concurrent load, wrap the matcher in a pool. match is thread-safe, and the pools add parallelism and load-shedding on top. Each pool lets you submit(record) from any number of request threads and get a concurrent.futures.Future, or call the blocking match(record). MatcherPool and ProcessMatcherPool also stream an iterable through map (results in submission order) or imap_unordered (results as they finish); the two thread pools (MatcherPool and MicroBatchMatcher) can scale(n) the worker count up or down at any time. They are in-process helpers: no service and no network hop, and the caller owns transport (an HTTP handler, a Kafka consumer).

Option 1: MatcherPool (threads)¶

A pool of worker threads, each serving one record at a time through the loaded matcher. Simplest to reason about, and right for a handful of concurrent requests or when spawning processes is undesirable. Because the featurize string comparisons run in Python and hold the GIL, a thread pool tops out near 2x on CPU-bound matching:

from madmatcher_pro.serving import MatcherPool

with MatcherPool(matcher, n_workers=8) as pool:
    for matches in pool.imap_unordered(incoming_records):   # served in parallel, streamed
        handle(matches)

MatcherPool takes the loaded matcher and n_workers (default: the machine’s CPU count, capped at 16), plus the queue policy covered under Backpressure below (max_queue, on_full). track (default True) shows a live request counter and worker count on the dashboard whenever a session is active.

Option 2: MicroBatchMatcher (coalesce into batches)¶

Coalesces concurrent single-record requests into match_batch windows, for throughput on a request stream without changing the per-request contract. Each worker groups requests arriving close together (up to max_batch, or until max_delay_ms) and serves the window with one match_batch, so the window shares one embedder pass and one predict. Reach for it when embedding or prediction dominates and requests arrive faster than one at a time:

from madmatcher_pro.serving import MicroBatchMatcher

batcher = MicroBatchMatcher(matcher, n_workers=2, max_batch=32, max_delay_ms=10.0)
future = batcher.submit({"name": "apple watch series 9", "brand": "Apple"})
matches = future.result()
batcher.close()

max_batch (default 32) caps the window size and max_delay_ms (default 10) caps how long a window waits to fill, so a lone request is never stuck: it is served after at most max_delay_ms. n_workers (default 1) sets how many windows are served in parallel. Requests stay isolated even when coalesced: one failing request fails only its own Future, never the other requests in its window.

Option 3: ProcessMatcherPool (near-linear across cores)¶

For CPU-bound matching where you want near-linear throughput across cores, worker processes bypass the GIL. Because each worker loads its own matcher, this pool takes a bundle path, not a loaded matcher, and each process pays its own load cost (its own PyLucene JVM, model, and embedder). Memory-mapped bundle artifacts (the corpus rows, vectorizer arrays, Lucene segments, cosine embeddings) are shared across the processes, but the per-process state is not, so size the pool against measured per-process memory times n_procs, not the bundle size:

from madmatcher_pro.serving import ProcessMatcherPool

with ProcessMatcherPool("bundles/v1", n_procs=8) as pool:
    for matches in pool.imap_unordered(incoming_records):   # served across processes
        handle(matches)

ProcessMatcherPool takes the bundle path, n_procs (default: the CPU count), and load_kwargs, a dict forwarded to each worker’s RealtimeMatcher.load (for example id_col). Do not put spark in it: a session cannot be sent to another process, and a resident sklearn bundle loads without one (a Delex or SparkML bundle needs Spark in every worker, so prefer MatcherPool for those). semantic_cache_mb is a pool-wide semantic cache budget split evenly across the workers, and max_in_flight (default 4 per process) bounds outstanding requests the way the thread pools’ queue does.

A bundle published with the shared semantic cell store (every bundle published from this version on) does not use that cache at all: all workers memory-map one read-only copy of the semantic index’s cells, so the semantic memory cost of the pool is one copy per machine rather than one per worker, and you can size the pool by CPU instead of by RAM. For a bundle published before the cell store existed (or one whose semantic index you rebuilt in place, which invalidates the store and drops that bundle back to the cached path with a warning at load), you have three options: re-publish the bundle, run madmatcher_pro.semantic.cell_store.stage_cells_durably(<bundle>/blocking/semantic) once against the bundle’s semantic dir to (re)build the shared store in place (much cheaper than a re-publish; the next load picks it up), or keep budgeting semantic_cache_mb.

A bundle carrying an embedder or a Delex program is more expensive per process (each worker loads its own copy of the model weights, or rebuilds every Delex plan node), and the pool warns when it detects that so you can choose a smaller pool or fall back to MatcherPool.

How results are keyed in a pool. The thread pools (MatcherPool, MicroBatchMatcher) serve each record without a per-request key, so a result’s id2 comes from the record’s own id (the id_col from load), or is None for a record carrying none. Feed them records that carry their ids. ProcessMatcherPool.submit additionally accepts an explicit key=, and the streaming runners in Steps 5 and 6 assign keys for you.

Backpressure. Every pool bounds its queue by default, so a producer faster than the pool gets backpressure rather than unbounded memory growth. on_full="block" (the default) waits for space; on_full="reject" raises QueueFullError so a caller can shed load.

Shutting down. close() stops accepting new requests (further submit raises), serves what is already queued, and joins the workers; a request still unserved when the workers exit resolves its Future with an error rather than hanging its caller. Using a pool as a with block closes it for you.

Step 5: Streaming Records Through a Matcher¶

When you have a file or a stream to run start to finish, run_matching owns the loop and the buffering: it pulls from a RecordSource, matches through the matcher, and writes each result to a ResultSink, holding only a bounded window in memory so an unbounded source works.

from madmatcher_pro.serving import run_matching, ParquetSource, ParquetSink

with ParquetSource("incoming/", id_col="_id") as src, \
     ParquetSink("out/matches") as sink:
    stats = run_matching(src, sink, matcher)
print(stats.records, stats.matches)

Sources yield (key, record) pairs and are iterated lazily (never materialized whole): IterableSource (any iterable or generator), ParquetSource, CsvSource, and QueueSource (drain a queue.Queue a producer thread fills). Sinks take each result frame: MemorySink and DataFrameSink (accumulate in memory), CallbackSink (hand each result to your function), ParquetSink (append parquet parts through local/HDFS/S3/GCS storage), and FanOutResultSink (several at once). See the Appendix for the full tables.

run_matching takes source, sink, and matcher positionally; matcher can be a RealtimeMatcher or any pool from Step 4, so streaming through parallel workers is just passing the pool. Its keyword arguments:

batch_size (default None) matches in groups of this size through match_batch, so each group shares one embedder pass and one predict. None matches one record at a time.
passthrough (default False) also hands the sink each incoming record, so it can store the original fields beside the match (ParquetSink selects which with passthrough_cols).
progress_every (default 0, silent) logs a progress line every N records.

It returns a MatchRunStats whose fields are records (records matched), matches (match rows written), and elapsed (wall-clock seconds). A record whose key resolves to None is assigned its position in the stream as its key, so every written row stays attributable to an input. The sink is flushed when the run ends, even when it ends on an error.

Keeping the incoming record’s fields with its matches. A match row alone carries only ids and scores. To write the incoming record’s own fields beside each of its matches, so a consumer reads results without a second lookup, turn on passthrough and name the fields on the sink:

sink = ParquetSink("out/matches", passthrough_cols=["name", "brand"])
stats = run_matching(ParquetSource("incoming/", id_col="_id"), sink, matcher,
                     passthrough=True)

Each written row then carries [id1, id2, prediction, confidence] plus that record’s name and brand. The same flag serves the other sinks: MemorySink keeps the records beside the result frames, and CallbackSink receives each (result, record) pair.

Step 6: Durable and Multi-Worker Streaming¶

run_matching is not durable: a crash starts the run over from the beginning of the source. run_matching_durable adds a per-record claim protocol on shared storage, which buys three things from one mechanism. “The input” here is one body of incoming records, the same source (a parquet directory, a queue) each worker reads:

Crash resumption. Re-running the same command after a crash skips what already completed.
Multiple workers. Several processes given the same source and the same checkpoint_dir split that work: each reads the records and claims what it can, and together they cover the input once, with no broker or coordinator. Workers can join or leave mid-run.
Unconnected machines. The same command on several machines that can all see checkpoint_dir cooperates the same way.

from madmatcher_pro.serving import run_matching_durable, ParquetSource, ParquetSink

stats = run_matching_durable(
    ParquetSource("incoming/", id_col="_id"),
    ParquetSink("out/matches"),
    matcher,
    checkpoint_dir="ckpt/run-7",
)

Delivery is at-least-once: a worker that dies after writing results but before recording completion leaves the record claimable, and a later worker redoes it. Pass exactly_once=True to skip a record already marked done; to close the last window (a crash between the sink write and the completion marker) use an idempotent sink that de-duplicates on (id1, id2).

The key is a record’s durable identity. Claims and completion markers are named by each record’s key (integer or string, either works), so “this record is already done” means “a record with this key is done”. Keys must therefore be stable across re-runs and unique within the run: a source whose records carry ids (id_col) or explicit keys resumes exactly, while unkeyed records fall back to their position in the source, which resumes correctly only if the source replays in the same order.

Every worker reads the whole source. Claims decide who processes a record, not who reads it: each worker iterates the source in order and skips past records it does not claim. That is cheap for a parquet or CSV source; for a source that is expensive to read, give each machine its own slice of the input instead of the identical source.

Why it is opt-in. Durability is not free, so run_matching is the default. At record granularity every record costs a claim marker, a completion marker, and a sink flush (several small writes to checkpoint_dir), delivery becomes at-least-once (so you often want an idempotent sink), and you must provision shared checkpoint storage every worker can reach. For a bounded run on one machine that you can simply re-run on failure, plain run_matching is simpler and faster. Reach for the durable form when repeating a crashed run is expensive, or when you want several workers on one input.

Granularity. The default granularity="record" claims one record at a time, so a record is processed the moment it arrives, which is what a live stream needs. granularity="shard" batches claims into fixed-size groups for cheaper markers, but a shard is processed only once it fills, so use it only for bulk backfill of records that are all already present. One consequence of per-record durability: the sink is flushed after every record (results must be durable before the completion marker), so a ParquetSink in record mode writes one part file per matched record. For compact bulk output prefer shard mode or a coalescing sink. The Reference section documents lease_seconds, max_units, worker_id, and the rest.

Step 7: Handing Off From a Batch Build¶

A batch build takes a while, and records arriving during it have nowhere to go: the bundle they would match against does not exist yet. BatchToServing closes that gap without dropping records, by moving through four phases: it accepts arriving records into a durable spool while the build runs, publishes the bundle when the build finishes, drains the spooled backlog against the new bundle, then serves live records from there on.

from madmatcher_pro.serving import BatchToServing, RealtimeMatcher, ParquetSink, publish_serving_artifacts

h = BatchToServing(spool_dir="spool/", bundle_dir="bundles/v1",
                   id_col="_id", checkpoint_dir="ckpt/drain")

# While the batch build runs, records keep arriving. Spool them durably.
h.accept({"_id": 1001, "name": "apple pie", "brand": "acme"})

# The build finishes: publish from its artifacts, load, and go live.
h.publish(lambda out: publish_serving_artifacts(
    out, indexed_table=corpus, features=features, model=model,
    sparkly_index_path="build/lucene", fill_na=0.0))
h.start_serving(RealtimeMatcher.load("bundles/v1"))

# Match the backlog, then serve live.
h.drain(ParquetSink("out/backlog"))
h.match({"_id": 1002, "name": "banana bread", "brand": "acme"})   # live from here on

BatchToServing takes:

spool_dir is where accepted records are buffered durably, as append-only segment files. Construction also recovers any segment a crashed writer left open, so records accepted before a restart are not stranded.
bundle_dir is where publish writes the bundle.
id_col (optional) supplies each accepted record’s key, exactly as in Step 2.
checkpoint_dir (optional) makes the drain durable with Step 6’s per-record claims, so a crash mid-drain resumes instead of restarting the backlog. Without it the drain is a plain run_matching.
segment_rows (default 1000) is how many records go into one spool segment before it is sealed. Smaller segments become drainable sooner; larger ones make fewer files.

Its methods, in the order the phases use them:

accept(record, key=None) buffers one record durably and returns its key. It is safe to call from many threads, and it keeps working after serving starts (it just spools), so a producer never needs to track which phase the handoff is in.
publish(publish_fn) seals what has been spooled so far, then calls publish_fn(bundle_dir); you supply the publish_serving_artifacts call from Step 1.2, as in the example. Records still arriving keep spooling meanwhile.
start_serving(matcher) goes live with a loaded matcher (or pool). match raises until this is called; before that, accept is the way in.
drain(sink, batch_size=None, passthrough=False, clear=True) matches the backlog into a sink and returns the same MatchRunStats as Step 5. It snapshots the sealed segments at its start and matches exactly that snapshot, so a record arriving mid-drain is neither double-matched nor lost (it waits for the next drain, or for live serving). clear=True (the default) deletes the drained segments, and only after their results are in the sink; pass clear=False to keep them, for example to re-drain into a second sink. batch_size groups a plain (non-durable) drain through match_batch; on a durable drain, pass runner options such as lease_seconds as extra keyword arguments instead.
match(record, key=None) serves one record live, as in Step 2.
close() stops accepting and seals the spool. It does not close the matcher; you own that lifecycle, which usually outlives the handoff.

The underlying RecordSpool is itself a RecordSource, so you can also point run_matching at it directly.

Step 8: Refreshing the Bundle Without Downtime¶

To fold new corpus records in, rebuild a bundle offline (this is not a per-record live upsert) and swap it into the running process. HotSwappableMatcher serves from a loaded bundle and can atomically switch to a newer one: the new bundle loads and warms while the old one keeps serving, then requests switch over in one step, and the old bundle is retired once its in-flight requests finish (each request finishes on the bundle it started on).

from madmatcher_pro.serving import HotSwappableMatcher

live = HotSwappableMatcher("bundles/v1")
live.match({"name": "apple pie", "brand": "acme"})

# Later, after publishing bundles/v2 offline:
live.swap("bundles/v2")     # v2 now serves; v1 retired once its requests drain

swap re-checks the license and realtime entitlement, so a lapsed license stops future serving, and a failed load (an invalid or half-published bundle) leaves the current bundle serving unchanged.

HotSwappableMatcher takes the first bundle to load, spark (for a Delex or SparkML bundle), and any RealtimeMatcher.load keyword arguments, which it forwards to every load it performs (for example id_col). drain_timeout (default 30 seconds) is how long a swap waits for the old bundle’s in-flight requests before giving up on closing it: a request that outlives the timeout leaves its old bundle open rather than being torn down mid-request, and that bundle is retired later, once the request finishes (the undrained property counts bundles in that state; normally 0). version names the bundle currently serving, and track (default True) shows each swap’s load, drain, and go-live on the dashboard.

Many replicas. To tell N serving replicas that a new bundle exists, write a pointer file to the storage every replica already reads bundles from, so no message bus is needed. A publisher writes the pointer after publishing, and each replica runs a BundleWatcher that polls it and hot-swaps when the version changes:

from madmatcher_pro.serving import publish_bundle_version, BundleWatcher

# On the publisher, after publish_serving_artifacts("s3://.../bundles/v2", ...):
publish_bundle_version("s3://.../bundles", "s3://.../bundles/v2", version="v2")

# On each replica:
live = HotSwappableMatcher("s3://.../bundles/v1")
watcher = BundleWatcher(live, "s3://.../bundles", poll_seconds=30.0).start()
# ... serve; new versions are picked up automatically ...
watcher.stop()

Write the pointer only after publish_serving_artifacts has returned: the pointer tells replicas the bundle is safe to load, so writing it early would send them at a half-written bundle. The pointer itself is one small JSON file (CURRENT.json) written atomically, so a replica polling mid-write sees the old version or the new one, never a half-write; read_bundle_version(pointer_dir) reads it back.

BundleWatcher takes the live matcher and the pointer directory, plus poll_seconds (default 30; the check is one small file read, so anywhere from seconds to minutes is normal) and an optional on_swap callback called after each successful swap. start() polls from a background thread and stop() ends the polling without closing the matcher; to drive the check from your own loop instead, call check_once(), which polls once and returns whether it swapped. A replica that was down picks up the current version when it starts, a failed swap is logged and the replica keeps serving the version it has (so a broken bundle never takes a replica down), and you roll back by writing the pointer again.

Closing Notes¶

You built the corpus state once with a batch job, published it as a bundle, and matched incoming records against it in memory. From there you scaled up along independent axes: batches, thread and process pools, orchestrated streaming, durable multi-worker runs, a lossless handoff from the batch build, and live bundle refresh across replicas. Every path reuses the exact batch kernels, so a served match equals the batch match for the same record and corpus. When you need the durability and throughput-tracking details, the Reference section documents them in full, and the Appendix collects the sources, the sinks, and a table for choosing a serving surface.

Reference: Durability and Progress Tracking¶

This is the complete reference for the durability and progress features used in the steps above.

How durable streaming works¶

run_matching_durable coordinates workers through files under checkpoint_dir, with no broker:

Claim. Before processing a unit (a record, or a shard in shard mode), a worker writes a claim file naming that unit. A unit already claimed or completed is skipped, so two workers never process the same unit at once.
Process and write. The worker matches the unit and writes its results to the sink, then flushes, so results are durable before the next step.
Complete. The worker writes a completion marker for the unit. The marker, not the claim, is what records the unit as done.
Resume. Re-running reads the markers, skips completed units, and processes the rest. Several workers sharing the directory each take what they can, so together they finish the input once.

The manifest under checkpoint_dir pins the run’s configuration (granularity, shard_size) and refuses a mismatch: claims are named per unit, so mixing settings would double-process some records and skip others. Use a fresh checkpoint_dir for a different input.

The delivery guarantee¶

Delivery is at-least-once by default. A worker that dies after writing results but before its completion marker leaves the record claimable, so a later worker (or a re-run) redoes it. Two options tighten this:

exactly_once=True skips any record already marked done, suppressing that redelivery.
An idempotent sink that de-duplicates on (id1, id2) closes the remaining window, the gap between the sink write and the completion marker, where a crash could produce a duplicate.

A stalled or dead worker’s claim is reclaimed once it is deemed abandoned: a same-host worker whose process is gone, or any claim older than a lease. lease_seconds sets that lease (raise it for a granularity="shard" run whose shards legitimately take longer than the default, so a slow shard is not reclaimed from its worker). A reclaim is only ever a redo, which at-least-once already permits.

Resumable publish¶

Publishing writes the bundle in independent artifact stages (the corpus feature state, the index copies, the embedder, the model), each committed after its data is durable. enable_crash_recovery=True with a checkpoint_dir makes a crash mid-publish resume from the last committed stage instead of rebuilding the whole bundle:

publish_serving_artifacts(
    "bundles/v1", indexed_table=corpus, features=features, model=model,
    sparkly_index_path="build/lucene", fill_na=0.0,
    enable_crash_recovery=True, checkpoint_dir="ckpt/publish-v1",
)

Keep the checkpoint outside the bundle and use a fresh directory per publish: resuming an existing checkpoint continues that exact publish, so a re-publish (a refreshed corpus or a changed config) needs its own directory. With validate_input=True (the default) the corpus row count is pinned and re-checked, so a resume after the corpus grew or shrank is refused rather than serving stale artifacts. Reaching a non-local checkpoint_dir needs a live SparkSession.

What happens when a job crashes¶

Nothing is left corrupted. A committed unit (a completed record, or a published bundle) is safe, because its marker is written only after its output is durable. A unit interrupted mid-write has no marker, so it is treated as not done. To recover, run the same command again with the same checkpoint_dir: there is no separate resume command. A completed record is never reprocessed (except as an at-least-once redo), an interrupted one is redone, and a finished run does no work.

Viewing throughput on the dashboard¶

Every serving surface can report to the same web dashboard the batch engines use. Pass track=True when loading a matcher or constructing a pool, or create a MadMatcherSession near the top of your script so everything that follows is tracked:

from madmatcher_pro import MadMatcherSession

with MadMatcherSession.builder.open_browser(True).getOrCreate() as session:
    print(f"Dashboard: {session.url}")
    matcher = RealtimeMatcher.load("bundles/v1", track=True)
    # each match / match_batch advances a live "Serving requests" counter

Each matcher and pool shows a live throughput counter with per-surface KPIs (a pool’s worker count, a micro-batcher’s window sizes). A HotSwappableMatcher shows a swap-lifecycle job counting versions swapped in, and a publish shows up as a job while it runs. Tracking is best-effort: if the dashboard cannot start (a busy port, a headless host) the work still runs. The dashboard has no built-in authentication, so on a shared host reach it by SSH port-forward or secure it at the network layer, exactly as the engine tutorials describe. The read-only viewer (python -m madmatcher_pro.reliability.dashboard <checkpoint_dir>) works against a durable run’s checkpoint directory from any machine that can read it.

Durability arguments¶

run_matching_durable(..., checkpoint_dir=...) is required and names where claims and completion markers live. It must be reachable by every worker sharing the run, and not reused across different inputs.

granularity ("record" default, or "shard") claims per record (safe for live streams) or per fixed-size group (cheaper markers, bulk backfill only). It does not: change the result, only how work is claimed and flushed.

shard_size (default 1000) is records per shard when granularity="shard".

exactly_once (default False) skips records already marked done instead of re-processing them. It does not: close the crash-between-write-and-marker window on its own; pair it with an idempotent sink for that.

lease_seconds (default None, the store default) is how long before an incomplete claim is assumed abandoned and reclaimable. It does not: affect a single-worker crash-resume, where the same host reclaims its own dead process’s work regardless.

batch_size (default None, shard mode only) matches within a shard in groups through match_batch.

max_units (default None) stops this worker after completing that many units.

worker_id (default host+pid) identifies this worker in claim files.

spark (default None) is needed only for a non-local checkpoint_dir.

progress_every (default 0, silent) logs a line every N records.

Appendix¶

Record sources¶

A RecordSource yields (key, record) pairs lazily, so an unbounded stream works.

Source	What it reads
`IterableSource(records, id_col=None)`	any Python iterable or generator of records (or of `(key, record)` pairs). The in-memory case and the adapter for a reader you already have.
`ParquetSource(path, id_col=None, columns=None, batch_rows=10_000)`	a parquet file or directory, streamed one batch at a time through pyarrow.
`CsvSource(path, id_col=None, chunk_rows=10_000, **read_csv_kwargs)`	a CSV, in chunks. Extra kwargs pass to `pandas.read_csv`.
`QueueSource(queue_=None, maxsize=1000, id_col=None)`	a `queue.Queue` a producer thread fills. Iteration ends when the producer calls `close()`.

id_col names the column each record carries its own id in, used as its key. batched(source, size) groups a source into lists of at most size for match_batch or shard runs.

Result sinks¶

A ResultSink takes each result frame. A sink is called from several worker threads at once, so a custom write should lock around shared state.

Sink	Where results go
`MemorySink()`	accumulated in memory (`.frames`, `.to_frame()`). Grows with the stream; for bounded runs and tests.
`DataFrameSink()`	accumulated into one `pandas.DataFrame` (`.result`), to hand straight to the next pipeline step.
`CallbackSink(fn)`	each result handed to your `fn(result, record)` (or `fn(result)`). Holds nothing, so an unbounded stream is fine.
`ParquetSink(out_dir, batch_rows=10_000, passthrough_cols=None, spark=None)`	appended as parquet parts through local/HDFS/S3/GCS storage, each written atomically.
`FanOutResultSink([...])`	every result sent to several sinks at once (for example keep it in a DataFrame and also write it durably).

Pass passthrough=True on the runner (or passthrough_cols= on ParquetSink) to store the incoming record’s fields beside each match.

Choosing a serving surface¶

You have	Use	Why
one record, want the answer now	`RealtimeMatcher.match`	the base call
several records in hand	`RealtimeMatcher.match_batch`	one embed pass and one predict for the batch
concurrent requests, a few cores	`MatcherPool`	worker threads over one shared matcher
concurrent requests, coalesce for throughput	`MicroBatchMatcher`	groups requests into `match_batch` windows
CPU-bound, want near-linear scaling	`ProcessMatcherPool`	worker processes bypass the GIL
a file or stream to run start to finish	`run_matching`	owns the loop and bounded buffering
the above, crash-safe or multi-worker	`run_matching_durable`	per-record claims and resume
records arriving during the batch build	`BatchToServing`	spool, then drain, then serve
refresh the corpus without downtime	`HotSwappableMatcher` (+ `BundleWatcher`)	atomic bundle swap, per replica

What a bundle contains¶

publish_serving_artifacts writes a self-contained directory:

MANIFEST.json: the pinned configuration (feature fingerprint, model kind, fill_na, block_limit, id_col, the blocking query spec, and each enabled blocker’s settings).
_PUBLISHED.json: the terminal marker, written last. Its absence means an incomplete bundle, which the loader refuses.
feature_state/: the preprocessed corpus feature state.
model/: the trained model in its native (no-pickle) format.
blocking/: the published blockers, sparkly/ (the Lucene index), semantic/ (the flat index), delex/ (the corpus and unbuilt program), embedding_store/ (the corpus vectors as a shared memmap), and embed_provider.pkl (the embedder config), whichever were published.

For the full API, see the reference at https://docs.madmatcher.ai/madmatcher-pro/{version}/madmatcher_pro.serving.html ({version} is your installed release). For help, email support@madmatcher.ai or visit madmatcher.ai.