Real-time serving (add-on)¶

Resident, in-process record matching (madmatcher_pro.serving): publish a batch build’s artifacts as a bundle once, load it into memory, and answer each incoming record from the loaded state. This is a separately-licensed add-on; RealtimeMatcher.load requires the realtime license entitlement. Start with the real-time matching tutorial; this page documents every public entry point.

Publishing a bundle¶

madmatcher_pro.serving.publish_serving_artifacts(out_dir, *, indexed_table, features, model, sparkly_index_path=None, fill_na, block_limit=50, id_col='_id', prebuilt_feature_table=None, query_spec=None, semantic_index_path=None, embed_provider=None, embed_fields=None, serializer=None, nprobe=32, fusion_weights=None, fusion_kappa=60, cosine_feature=False, embedding_store=None, embedding_store_id_col='_id', embedding_store_col='embedding', delex_program=None, delex_corpus=None, delex_id_col='_id', delex_optimize=False, delex_estimate_cost=False, spark=None, enable_crash_recovery=False, checkpoint_dir=None, validate_input=True, dashboard=None, dashboard_port=None, open_browser=None)¶

Write a serving bundle at out_dir and return its path.

Publish a new directory per version. out_dir is rewritten in place, which is safe for a fresh directory but must not target a bundle a RealtimeMatcher is actively serving (a live reader holds the corpus memory-map open). To switch a live process to a refreshed bundle with no downtime, publish the new version to its own directory and swap it in with HotSwappableMatcher.

The bundle is assembled as ordinary files on the driver, which has two consequences once other machines are involved:

A Spark DataFrame is written by the executors, to the destination path as each of THEM resolves it. embedding_store and delex_corpus are the two arguments that can be DataFrames. If the bundle directory is on storage every node shares – one machine, a network mount, a lakehouse volume – the executors write exactly where the driver reads and a DataFrame is fine. If the bundle is on the driver’s own disk while the executors are elsewhere, each of them writes to its own local disk under that path and the bundle gets nothing. Publishing verifies the driver can read each Spark-written artifact back, so that case fails naming the cause instead of surfacing steps later as a schema error. The fix is to stage the table once to a filesystem both sides can see and pass that URI: embedding_store and delex_corpus accept a path or URI on any filesystem Spark can address (hdfs://, s3a://, gs://, abfss://, a local or shared-mount path), and publishing copies it into the bundle itself with no Spark write. Note that the relevant question is where the bundle is, not whether the cluster has a distributed filesystem: a file:// write ignores fs.defaultFS entirely.
Give publishing its own Spark application. Publishing itself registers nothing with the SparkContext, but a batch featurize run in the same context has already registered its own copy of the preprocessed corpus (addFile), and Spark ships every registered file to all executors of every later job. For a large corpus that is tens of gigabytes per executor, enough to fill their disks and get the nodes marked unhealthy. Run publishing as its own application rather than as a tail stage of a matching job.

Example

Publish a bundle from a finished build, then serve from it:

from madmatcher_pro.serving import publish_serving_artifacts, RealtimeMatcher

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus_df,          # your records (a Spark DataFrame)
    features=features,                # the feature list from training
    model=trained_model,              # the trained model
    sparkly_index_path="lucene_dir",  # the search index you built
    fill_na=0.0,                      # same value you gave featurize() in training
)
matcher = RealtimeMatcher.load("bundles/v1")

Parameters:

indexed_table (Spark DataFrame) – the corpus to serve against (the candidate side); each incoming record is matched against it.
features (list) – the feature list used to train model.
model (SKLearnModel or SparkMLModel) – the trained matcher. An sklearn/xgboost model predicts in-process with no Spark on the request path; a SparkML model has no in-process predict, so serving it runs one Spark job per match (supported for batch interoperability, but slower).
sparkly_index_path (str or Path, optional) – an existing built sparkly index directory, copied into the bundle. Omit it for a bundle whose lexical blocker is delex, that blocks with semantic alone, or that is matching-only (served via match(record, candidates=[...])). At least one blocker is recommended unless the bundle is used only in matching-only mode.
fill_na (float) – the NaN fill value for feature vectors. Required: it must equal the fill_na passed to the training-time featurize(...), because a mismatch silently changes NaN-valued features at serve time and cannot be validated against the model.
prebuilt_feature_table (MemmapDataFrame, optional) – the preprocessed corpus table a batch run already built (from build_preprocessed). Passing it skips re-scanning and re-preprocessing the whole corpus, the most expensive part of publishing. Must have been built from indexed_table with these features.
block_limit (int) –
how many candidates blocking retrieves per incoming record (default 50), and so how many pairs the model scores. Pinned in the manifest and read at load.

Deduplicating (serving corpus records against a corpus that contains them): serving has no self-match exclusion, so a record’s own row comes back at rank 1 from every blocker and eats one of the slots. The batch equivalent asked its blocker for K+1 candidates and dropped the self pair, so to serve the same K real candidates publish block_limit=K+1 and drop id1 == id2 from the results yourself. Getting this wrong costs exactly one candidate per record: it shows up as a small recall gap against the batch run, never as an error. 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 rather than re-publishing.
id_col (str) – the column an incoming record carries its own id in (default _id). Serving stamps it onto each result row as id2 when a request passes no explicit key, so a served row always says which record produced it. A record without this column is still matched; results are then keyed only by an explicit key (see RealtimeMatcher.match).
semantic_index_path (str or Path, optional) – an existing built (flat) semantic index directory. When given, serving fuses sparkly and semantic blocking. A lossy-codec index is refused (serving decodes flat postings only).
embed_provider (EmbeddingProvider or str, optional) – the embedder the corpus was built with: the same provider instance you passed to the batch embedding step (e.g. a SentenceTransformerProvider), or a model-id string. Its config is stored in the bundle, and the loader rebuilds one resident embedder shared by the semantic probe and the embedding store. None means every match call must supply a precomputed probe_embedding.
embed_fields (list, optional) – the record fields the serializer embedded, in order. Required when semantic_index_path is given.
serializer (Serializer, optional) – the serializer the corpus embeddings were built with (a DittoSerializer or PlainConcatSerializer). Its kind and config are stored so the resident probe formats each probe’s text identically. None uses DittoSerializer defaults. A custom serializer is refused (it cannot be reproduced per record).
nprobe (int) – semantic cells probed per record (recall/latency knob).
fusion_weights (dict, optional) – RRF source weights, e.g. {"sparkly": 0.5, "semantic": 0.5}.
fusion_kappa (int) – RRF kappa (rank damping).
cosine_feature (bool) – True when the model was trained with the semantic cosine appended as the last feature. Serving then appends the per-candidate cosine identically, which needs a cosine source: semantic_index_path (serving reuses the semantic blocker’s scores) and/or embedding_store (serving recomputes cosine from stored corpus vectors for the candidates the reuse path misses). With both, serving reuses where it can and recomputes only the gaps.
embedding_store (str/Path or Spark DataFrame, optional) – the corpus embedding table (candidate side): a parquet path or URI (local, or on any filesystem Spark can address), or a Spark DataFrame – see the note above on which of the two a given deployment can use. Needs an id column and an embedding column. When given, serving computes the cosine feature by fetching each candidate’s stored vector and dotting it with the probe’s embedding, so even a candidate no semantic blocker ranked gets its real trained cosine. Stored in the bundle. Must map each corpus id to exactly one vector (a duplicate id is refused at load). The probe is embedded with embed_provider / embed_fields / serializer (as for the semantic index); if embed_provider is None, every match must supply probe_embedding.
embedding_store_id_col (str) – the id and embedding column names in embedding_store (default _id / embedding).
embedding_store_col (str) – the id and embedding column names in embedding_store (default _id / embedding).
delex_program (BlockingProgram, optional) – the delex program to fuse. Delex has no cross-process resident form, so the bundle stores the raw corpus and the (unbuilt) program, and the loader rebuilds the resident probe once at load. Fusing delex therefore makes the loader require a SparkSession.
delex_corpus (Spark DataFrame or str/Path, optional) – the corpus delex indexes (the candidate side). Defaults to indexed_table. Stored in the bundle as parquet: a DataFrame is written there by Spark, while a path or URI to an existing parquet dataset is copied in without Spark – see the note above on which of the two a given deployment can use.
delex_id_col (str) – the id column of the delex corpus (default _id).
delex_optimize (bool) – run the delex plan optimizer / cost estimation at the load-time build. Default off (the default plan).
delex_estimate_cost (bool) – run the delex plan optimizer / cost estimation at the load-time build. Default off (the default plan).
enable_crash_recovery (bool) – write the bundle as resumable, checkpointed artifact stages so a crash mid-publish resumes from the last committed stage instead of rebuilding the whole bundle. Off (default) writes the bundle in one non-checkpointed pass. Requires checkpoint_dir.
checkpoint_dir (str or Path, optional) – required when enable_crash_recovery=True: where to keep the publish checkpoint. Any FileSystemStore-addressable URI (local, hdfs://, s3a://, …); reaching it needs a live SparkSession. Keep it 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. The checkpoint is never auto-deleted.
validate_input (bool) – with enable_crash_recovery (default True), pin the corpus row count and re-check it on resume: a resume after the corpus grew or shrank is refused (CheckpointMismatchError) rather than serving stale artifacts. Costs one count() over the corpus. Pass False to skip it (a same-count corpus swap is not detected either way).
dashboard (optional) – optional progress tracking on the shared dashboard (defer to the active session or env when None).
dashboard_port (optional) – optional progress tracking on the shared dashboard (defer to the active session or env when None).
open_browser (optional) – optional progress tracking on the shared dashboard (defer to the active session or env when None).

madmatcher_pro.serving.check_embedding_store(embedding_store, id_col='_id', embedding_col='embedding')¶

Validate a corpus embedding store before publishing it into a serving bundle.

Checks that the id and embedding columns are present and that the id column is unique (a 1:1 id-to-vector map). publish_serving_artifacts runs this automatically when an embedding_store is given, so a bad table fails at publish rather than at serve time; you can also call it directly to check a table up front. A duplicate id is rejected: it would make the resident store’s lookup ambiguous and diverge from the cosine feature the model was trained on.

Example:

check_embedding_store(my_embeddings_df, id_col="_id", embedding_col="embedding")
# raises ValueError if a column is missing or an id is repeated

Parameters:

embedding_store – pyspark DataFrame | str | pathlib.Path The corpus embedding table (one row per record), or a path to a local parquet dataset of it.
id_col – str The id column name (default _id).
embedding_col – str The embedding (vector) column name (default embedding).

Returns:

int: The number of rows in the store (one per corpus id).

Raises:

ValueError – a column is missing, or the id column has duplicates.

The resident matcher¶

class madmatcher_pro.serving.RealtimeMatcher(served, sparkly_probe, model, *, semantic_probe=None, delex_probe=None, cosine_feature=False, embed_store=None, fusion_weights=None, fusion_kappa=60, block_limit=50, threshold=0.5, id_col=None)¶

Bases: object

Matches one incoming record (or a batch of them) against a fixed set of your records, fast enough to answer live requests.

Create one with RealtimeMatcher.load("bundle_dir"), then call match(record).

classmethod load(artifacts_dir, *, spark=None, threshold=0.5, track=False, dashboard=None, dashboard_port=None, open_browser=None, id_col=None)¶

Load a saved matcher package from a folder and return a ready-to-use matcher.

This is the only supported way to create a matcher: it checks that your license includes the real-time add-on, then loads everything the matcher needs into memory. Creating a matcher any other way makes match refuse to run.

Example:

matcher = RealtimeMatcher.load("bundle_dir")
matcher.match({"name": "apple pie", "brand": "acme"})

Parameters:

artifacts_dir – str | Path A bundle written by publish_serving_artifacts (local or a Spark URI such as s3://… for the artifacts Spark reads).
spark – pyspark.sql.SparkSession | None Used to read the model bytes, rebuild a delex probe, and (for a SparkML model) run predict. None defaults to the active or newly created session. A SparkSession is required only for a delex or SparkML bundle.
threshold – float Default confidence threshold for match and match_batch.
track – bool Show this matcher’s throughput on the dashboard (a live Serving requests counter advanced per match and match_batch). Off by default. Finished by close.
dashboard – optional Dashboard resolution overrides for track (defer to the active session or env when None).
dashboard_port – optional Dashboard resolution overrides for track (defer to the active session or env when None).
open_browser – optional Dashboard resolution overrides for track (defer to the active session or env when None).

Returns:

RealtimeMatcher: A resident matcher ready to answer match(record).

Raises:

LicenseError – the base license is valid but lacks the realtime entitlement.
ValueError / FileNotFoundError – the bundle is incomplete, or its model width doesn’t match the pinned feature set.

match(record, *, candidates=None, threshold=None, probe_embedding=None, key=None)¶

Find which of your stored records match one incoming record, answered right away.

Returns a table with one row per matching record: the matched record’s id (id1), the incoming record’s key (id2), the match prediction, and a confidence score (highest confidence first).

Example:

matcher.match({"name": "apple pie", "brand": "acme"}, key="req-7")
# -> DataFrame [id1, id2, prediction, confidence], one row per match

Parameters:

record – dict | pandas.Series | 1-row pandas.DataFrame | pyspark.Row The incoming record as a column-to-value mapping.
key – any | None Value identifying this record in the results (id2). Use it when the record carries no id of its own (for example a request off a queue). None falls back to the record’s own id (the id_col given at load), then to None.
candidates – list[int] | None Bring your own candidates: when given, blocking is skipped and the record is matched only against these corpus ids (an id absent from the corpus is dropped; an empty or all-unknown list gives the empty result). A cosine-trained bundle needs an embedding store to produce the cosine feature this way. None (default) blocks as usual.
threshold – float | None Keep a candidate only when its match confidence is >= this. None uses the value set at load.
probe_embedding – numpy.ndarray | None A precomputed embedding for the semantic source or cosine store, skipping the resident embedder for this call. None embeds the record with the bundle’s embedder, and is required when none was loaded.

Returns:

pandas.DataFrame with columns [id1, id2, prediction, confidence]: One row per predicted match at or above the threshold, sorted by confidence descending then id1 ascending. id2 is this record’s key on every row, so a result stays attributable once it leaves the call.

Note

There is no self-match exclusion. When the corpus contains the records you are serving (deduplication), each record matches itself: drop the id1 == id2 row yourself, and publish the bundle with block_limit one higher than the number of real candidates you want, since the self-match occupies a candidate slot in every blocker.

match_batch(records, *, candidates=None, threshold=None, probe_embeddings=None, keys=None)¶

Match many incoming records at once, faster than calling match on each.

The results are exactly what you would get from match per record: one result table per input record, in the same order.

Example:

matcher.match_batch([
    {"name": "apple pie", "brand": "acme"},
    {"name": "banana bread", "brand": "acme"},
])
# -> [DataFrame, DataFrame]   # one [id1, id2, prediction, confidence] each

Parameters:

records – sequence of record mappings, or a pandas.DataFrame of rows The incoming records (each normalized as in match).
candidates – sequence | None None blocks every record; otherwise a sequence aligned with records, each element a bring-your-own id list for that record, or None to block it. Same cosine-store requirement as match.
threshold – float | None Confidence threshold applied to every record (None uses load’s).
probe_embeddings – sequence | None Optional precomputed embedding per record, aligned with records.
keys – sequence | None Optional key per record, aligned with records (as match’s key). None falls back per record to its own id under the id_col given at load.

Returns:

list[pandas.DataFrame]: One [id1, id2, prediction, confidence] frame per input record, in order.

close()¶

Release the files and memory this matcher is holding open.

Call this when you are done with the matcher, or before loading a new one, so it does not leave files open. Each part is closed independently, so a problem closing one part still releases the rest. Safe to call even if some parts were never opened.

Example:

matcher.close()

Returns:: None

Serving pools¶

class madmatcher_pro.serving.MatcherPool(matcher, *, n_workers=None, track=True, dashboard=None, dashboard_port=None, open_browser=None, job_id=None, max_queue=None, on_full='block')¶

Bases: _WorkerPool

Serve records through a loaded matcher on a resizable pool of worker threads, one record per call, in parallel.

Parameters:

matcher – RealtimeMatcher The loaded matcher to serve through.
n_workers – int | None Number of parallel workers (default: the machine’s CPU count, capped at 16). Resize any time with scale(n).
track – bool Show live throughput on the dashboard (a Serving requests counter plus requests and workers KPIs). On by default, best-effort, and gated on an active session (no cost when no dashboard is active). Finished by close.
dashboard – optional Dashboard resolution overrides (defer to the active session or env when None).
dashboard_port – optional Dashboard resolution overrides (defer to the active session or env when None).
open_browser – optional Dashboard resolution overrides (defer to the active session or env when None).
job_id – str | None Dashboard job id; defaults to a per-instance id.
max_queue – int | None Max requests waiting to be served. None = 8 per worker; 0 = unbounded (only for a caller bounding admission itself).
on_full – “block” | “reject” What submit does when the queue is full: wait for space (default), or raise QueueFullError so the caller can shed load.

Example:

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

map(records, *, max_in_flight=None)¶

Serve an iterable of records in parallel and yield each result in submission order, streaming (it starts yielding before the input is exhausted). At most max_in_flight requests are queued at once, so this works on an unbounded stream.

Example:

for matches in pool.map(records):
    write(matches)

Parameters:

records – iterable of records (as in match).
max_in_flight – int | None Max requests queued at once (default ~2 per worker).

Yields:

each record’s [id1, id2, prediction, confidence] frame, in input order.

imap_unordered(records, *, max_in_flight=None)¶

Like map, but yield each result as soon as it is ready (not in input order), for maximum streaming throughput. Bounded in-flight, so it works on an unbounded stream.

Example:

for matches in pool.imap_unordered(records):
    write(matches)

Parameters:

records – iterable of records (as in match).
max_in_flight – int | None Max requests queued at once (default ~2 per worker).

Yields:

each record’s [id1, id2, prediction, confidence] frame, in completion order.

close(*, wait=True, timeout=5.0)¶

Stop accepting new requests and shut down every worker. Requests already enqueued are still served; further submit raises. Safe to call more than once. (Using the pool as a with block calls this for you.)

Example:

pool.close()

Parameters:

wait – bool Join the worker threads before returning.
timeout – float Max seconds to wait for each worker’s join (when wait).

match(record, *, candidates=None, probe_embedding=None)¶

Submit a record and wait for its result (the blocking form of submit).

Example:

matches = pool.match({"name": "apple pie", "brand": "acme"})
# -> pandas.DataFrame [id1, id2, prediction, confidence]

Parameters:

record – the incoming record (as in RealtimeMatcher.match).
candidates – list[int] | None Match against these ids only, or None to search for candidates.
probe_embedding – numpy.ndarray | None A precomputed embedding for the record, or None to embed it.

Returns:

pandas.DataFrame [id1, id2, prediction, confidence] for the record.

scale(n)¶

Grow or shrink the pool to n workers, at any time while serving.

Scaling up starts new workers immediately. Scaling down is graceful: a retiring worker finishes its current unit (dropping no in-flight request) and then exits, so the effective count reaches n shortly after the call.

Example:

pool.scale(8)    # burst: 8 parallel workers
pool.scale(2)    # quiet: drop back to 2

Parameters:

n – int, >= 1 Desired pool size.

Returns:

self

Raises:

ValueError – n < 1.
RuntimeError – the pool has been closed.

submit(record, *, candidates=None, probe_embedding=None, timeout=None)¶

Enqueue one record; return a Future for its result.

Thread-safe: call from as many request threads as you like (a streaming source of requests). One failing request only fails its own Future.

The queue is bounded, so a producer faster than the pool gets backpressure rather than unbounded memory growth. When it is full this blocks until space frees (the default on_full="block") or raises QueueFullError (on_full="reject", for a caller that would rather shed load).

Example:

handle = pool.submit({"name": "apple pie", "brand": "acme"})
matches = handle.result()   # the [id1, id2, prediction, confidence] frame

Parameters:

record – dict | pandas.Series | 1-row pandas.DataFrame | pyspark.Row The incoming record (as in RealtimeMatcher.match).
candidates – list[int] | None Matching-only candidate ids for this record, or None to block it.
probe_embedding – numpy.ndarray | None Precomputed embedding for this record, or None to embed it.
timeout – float | None Max seconds to wait for queue space before raising QueueFullError. None waits indefinitely (under on_full="block").

Returns:

concurrent.futures.Future: Resolves to this record’s [id1, id2, prediction, confidence] frame (or the request’s own exception).

Raises:

RuntimeError – the pool has been closed.
QueueFullError – the queue is full and the policy is to reject (or the wait exceeded timeout).

property workers¶

The current number of live worker threads in the pool.

Example:

pool.workers
# -> 4

Returns:: int

class madmatcher_pro.serving.MicroBatchMatcher(matcher, *, n_workers=1, max_batch=32, max_delay_ms=10.0, track=True, dashboard=None, dashboard_port=None, open_browser=None, job_id=None, max_queue=None, on_full='block')¶

Bases: _WorkerPool

Coalesce concurrent single-record requests into RealtimeMatcher.match_batch calls, served by a resizable pool of worker threads, for throughput on a request stream.

A caller submit`s a record from any thread and gets a `Future. n_workers daemon drain threads share the queue, each grouping requests into windows (up to max_batch, or until max_delay_ms) and serving them with one match_batch, so a window shares one embedder pass and one predict. Separate windows run in parallel across the workers. Call scale(n) to add or remove workers at any time. It is an in-process helper: no service, no network hop, the caller owns transport.

Parameters:

matcher – RealtimeMatcher The loaded matcher to serve windows through.
n_workers – int, >= 1 Number of parallel drain threads sharing the queue (resize with scale(n)).
max_batch – int, >= 1 Max records coalesced into one match_batch window (per worker).
max_delay_ms – float Max time (ms) to wait for a window to fill before serving it.
track – bool Show live throughput on the dashboard (a Serving requests counter of total records served, plus batches, records, last batch size, and workers KPIs). On by default, best-effort, and gated on an active session (no cost when no dashboard is active). Finished by close.
dashboard – optional Dashboard resolution overrides (defer to the active session or env when None).
dashboard_port – optional Dashboard resolution overrides (defer to the active session or env when None).
open_browser – optional Dashboard resolution overrides (defer to the active session or env when None).
job_id – str | None Dashboard job id; defaults to a per-instance id so several batchers are distinguishable.
max_queue – int | None Max requests waiting to be coalesced. None = 8 per worker; 0 = unbounded.
on_full – “block” | “reject” What submit does when the queue is full: wait for space (default), or raise QueueFullError.

close(*, wait=True, timeout=5.0)¶

Example:

pool.close()

Parameters:

wait – bool Join the worker threads before returning.
timeout – float Max seconds to wait for each worker’s join (when wait).

match(record, *, candidates=None, probe_embedding=None)¶

Submit a record and wait for its result (the blocking form of submit).

Example:

matches = pool.match({"name": "apple pie", "brand": "acme"})
# -> pandas.DataFrame [id1, id2, prediction, confidence]

Parameters:

record – the incoming record (as in RealtimeMatcher.match).
candidates – list[int] | None Match against these ids only, or None to search for candidates.
probe_embedding – numpy.ndarray | None A precomputed embedding for the record, or None to embed it.

Returns:

pandas.DataFrame [id1, id2, prediction, confidence] for the record.

scale(n)¶

Grow or shrink the pool to n workers, at any time while serving.

Example:

pool.scale(8)    # burst: 8 parallel workers
pool.scale(2)    # quiet: drop back to 2

Parameters:

n – int, >= 1 Desired pool size.

Returns:

self

Raises:

ValueError – n < 1.
RuntimeError – the pool has been closed.

submit(record, *, candidates=None, probe_embedding=None, timeout=None)¶

Enqueue one record; return a Future for its result.

Thread-safe: call from as many request threads as you like (a streaming source of requests). One failing request only fails its own Future.

Example:

handle = pool.submit({"name": "apple pie", "brand": "acme"})
matches = handle.result()   # the [id1, id2, prediction, confidence] frame

Parameters:

record – dict | pandas.Series | 1-row pandas.DataFrame | pyspark.Row The incoming record (as in RealtimeMatcher.match).
candidates – list[int] | None Matching-only candidate ids for this record, or None to block it.
probe_embedding – numpy.ndarray | None Precomputed embedding for this record, or None to embed it.
timeout – float | None Max seconds to wait for queue space before raising QueueFullError. None waits indefinitely (under on_full="block").

Returns:

concurrent.futures.Future: Resolves to this record’s [id1, id2, prediction, confidence] frame (or the request’s own exception).

Raises:

RuntimeError – the pool has been closed.
QueueFullError – the queue is full and the policy is to reject (or the wait exceeded timeout).

property workers¶

The current number of live worker threads in the pool.

Example:

pool.workers
# -> 4

Returns:: int

class madmatcher_pro.serving.ProcessMatcherPool(bundle_path, *, n_procs=None, load_kwargs=None, loader=None, semantic_cache_mb=None, max_in_flight=None, on_full='block')¶

Bases: object

Serve records through a bundle on a pool of worker processes, for near-linear throughput across cores.

Each worker process loads the bundle once, so the pool takes a bundle path rather than a loaded matcher, and each process pays its own load cost (JVM, model, embedder). Size the pool against measured memory, not core count.

Parameters:

bundle_path – str | Path The published serving bundle each worker loads (as RealtimeMatcher.load).
n_procs – int | None Number of worker processes (default: the machine’s CPU count).
load_kwargs – dict | None Extra kwargs for RealtimeMatcher.load (e.g. threshold, id_col). spark is not picklable and not needed: a resident bundle loads without it.
loader – callable | None Advanced/testing: a picklable loader(bundle_path, **load_kwargs) that returns the per-process matcher. Defaults to RealtimeMatcher.load.
semantic_cache_mb – float | None Total memory the whole pool may spend caching semantic index cells, split evenly across the workers. None reads MM_SEMANTIC_CACHE_MB (default 1024); 0 disables caching. Only a bundle without the staged cell store uses this cache: a bundle published with staged_cells/ memory-maps one shared, read-only copy of the cells across all workers, and this budget is unused.
max_in_flight – int | None Max requests outstanding at once, so an unbounded producer cannot grow the parent’s heap. None = 4 per process; 0 = unbounded.
on_full – “block” | “reject” What submit does when that cap is reached: wait for a slot (default), or raise QueueFullError.

Example:

with ProcessMatcherPool("bundles/v1", n_procs=8,
                        load_kwargs={"threshold": 0.7}) as pool:
    for matches in pool.imap_unordered(records):   # served across processes
        handle(matches)

property procs¶

The number of worker processes in the pool.

Returns:: int

submit(record, *, candidates=None, probe_embedding=None, key=None, timeout=None)¶

Send one record to a worker process; return a Future for its result.

In-flight work is bounded: once max_in_flight requests are outstanding this waits for one to finish (the default on_full="block") or raises QueueFullError (on_full="reject"). The bound keeps an unbounded submit loop from growing the parent process’s heap with pending pickled records.

Example:

handle = pool.submit({"name": "apple pie"}, probe_embedding=v, key="req-1")
matches = handle.result()

Parameters:

record – dict | pandas.Series | 1-row pandas.DataFrame | pyspark.Row The incoming record (pickled to the worker; must be picklable).
candidates – list[int] | None Matching-only candidate ids, or None to block.
probe_embedding – numpy.ndarray | None Precomputed embedding for the record, or None to embed it.
key – any | None Value identifying this record in the results (id2).
timeout – float | None Max seconds to wait for a slot before raising QueueFullError.

Returns:

concurrent.futures.Future resolving to the [id1, id2, prediction, confidence] frame (or the request’s exception).

Raises:

RuntimeError – the pool has been closed.
QueueFullError – no slot free and the policy is to reject (or timeout hit).

property in_flight¶: Requests submitted but not yet finished.

property submitted¶: Requests submitted over this pool’s lifetime.

match(record, *, candidates=None, probe_embedding=None, key=None)¶: Send one record and wait for its result (the blocking form of submit).

map(records, *, max_in_flight=None)¶

Serve an iterable of records across the processes and yield each result in submission order, streaming, with a bounded number in flight (so an unbounded stream does not materialize). For per-record candidates or embeddings, use submit directly.

Yields:: each record’s [id1, id2, prediction, confidence] frame, in input order.

imap_unordered(records, *, max_in_flight=None)¶: Like map, but yield each result as soon as it is ready (not in input order), for maximum streaming throughput. Bounded in-flight.

close(*, wait=True)¶

Shut the pool down: release each worker’s matcher, then stop the processes.

Releasing each matcher frees resources (a Lucene reader, staged temp state) that killing the process would leave behind. Safe to call more than once.

Parameters:: wait – bool Wait for pending work to finish before returning.

exception madmatcher_pro.serving.QueueFullError¶

Bases: RuntimeError

The pool’s queue is full and its admission policy is to reject.

Signals load shedding: the caller decides whether to retry, drop, or push back on its own upstream. Only raised under on_full="reject"; the default policy blocks the producer instead.

Bundle refresh¶

class madmatcher_pro.serving.HotSwappableMatcher(artifacts_dir, *, spark=None, poll_interval=0.005, drain_timeout=30.0, track=True, dashboard=None, dashboard_port=None, open_browser=None, **load_kwargs)¶

Bases: object

A RealtimeMatcher you can atomically swap to a refreshed bundle.

Thread-safe: match runs concurrently with other match calls and with a swap; swap and close are serialized with each other.

Load the first bundle and start answering records.

Switch to a refreshed bundle later with swap while requests keep flowing.

Example:

live = HotSwappableMatcher("bundles/v1", threshold=0.7)
live.match({"name": "apple pie"})
# -> pandas.DataFrame [id1, id2, prediction, confidence]

Parameters:

artifacts_dir – str | Path The bundle to load first (as RealtimeMatcher.load).
spark – pyspark.sql.SparkSession | None Passed to load; needed for a delex or SparkML bundle.
poll_interval – float How often, in seconds, to re-check whether a retiring bundle’s outstanding requests have finished.
drain_timeout – float | None Seconds to wait for a replaced bundle’s outstanding requests to finish before giving up on closing it (None waits forever). A request that runs longer than this leaves the old bundle open (a bounded, one-matcher leak) rather than blocking future swaps or being torn down mid-request.
track – bool Show the swap lifecycle on the dashboard: a Hot-swap job counting versions swapped in, with each swap’s load/drain/live/failed steps on the event feed and the live bundle as a current bundle metric. On by default, finished by close. Controls only the swap job, not the wrapped matcher’s own request tracking.
dashboard – optional Dashboard resolution overrides for track.
dashboard_port – optional Dashboard resolution overrides for track.
open_browser – optional Dashboard resolution overrides for track.
**load_kwargs – Forwarded to RealtimeMatcher.load (e.g. threshold, id_col).

property version¶

The location (path) of the bundle currently being served.

Example:

live.version
# -> "bundles/v1"

Returns:: the artifacts directory of the bundle currently in use.
Return type:: str

property current¶

The live matcher object, for read-only inspection or tests.

Prefer match for real use: the matcher returned here may be replaced and closed at any time.

Returns:: the matcher currently answering requests.
Return type:: RealtimeMatcher

match(record, **kwargs)¶

Match one incoming record against the corpus and return its predicted matches.

Runs alongside other match calls and alongside a swap: a request that has already started finishes on the bundle it started on, even if a swap makes a newer bundle current partway through.

Example:

live = HotSwappableMatcher("bundles/v1", threshold=0.7)
live.match({"name": "apple pie"})
# -> pandas.DataFrame [id1, id2, prediction, confidence]

Parameters:

record – dict | pandas.Series | pandas.DataFrame | pyspark.sql.Row The incoming record to match (as RealtimeMatcher.match).
**kwargs – Forwarded to RealtimeMatcher.match (candidates, threshold, probe_embedding).

Returns:

pandas.DataFrame [id1, id2, prediction, confidence]: One row per candidate match for the record.

Raises:

RuntimeError – this matcher has been closed.

property undrained¶

Count of bundles left open because their requests had not finished.

Normally 0. A number that keeps climbing means requests are hanging, not that swapping is leaking.

swap(artifacts_dir)¶

Switch to a refreshed bundle with no downtime, then retire the old one once its outstanding requests finish.

The new bundle loads and warms up (including any build work) while the old one keeps serving, then requests switch over in one step. Loading re-checks the license and realtime entitlement, so a lapsed license stops future serving.

Example:

live = HotSwappableMatcher("bundles/v1")
live.swap("bundles/v2")   # v2 now serves; v1 retired once its requests finish

Parameters:

artifacts_dir – str | Path The refreshed bundle to load and make current.

Returns:

self, now serving the new bundle.

Return type:

HotSwappableMatcher

Raises:

RuntimeError – this matcher has been closed.
Exception – a failed load (invalid bundle, lost entitlement) raises and leaves the current bundle serving unchanged.

close()¶

Shut down and free the bundle currently being served.

Waits for any outstanding requests to finish first (up to drain_timeout) so it never pulls a bundle out from under a running request. Safe to call more than once.

class madmatcher_pro.serving.BundleWatcher(matcher, pointer_dir, *, poll_seconds=30.0, spark=None, on_swap=None)¶

Bases: object

Keep one serving replica on the current bundle, by polling the pointer.

Runs a daemon thread that checks the pointer every poll_seconds and, when the version changes, calls swap on the matcher you gave it, which loads the new bundle, drains in-flight requests, and closes the old one. Serving never stops.

A failed swap (a bad or half-published bundle) is logged and the replica keeps serving the version it has, so a broken bundle never takes a replica down.

Example:

hot = HotSwappableMatcher("bundles/v1", threshold=0.7)
watcher = BundleWatcher(hot, "bundles").start()
...                       # serve; new versions are picked up automatically
watcher.stop()

Parameters:

matcher – HotSwappableMatcher The live matcher to swap.
pointer_dir – str | pathlib.Path Where publish_bundle_version writes the pointer.
poll_seconds – float How often to check. Seconds to minutes is normal; the check is one small read.
spark – SparkSession | None Only needed for a non-local pointer_dir.
on_swap – callable | None Called as on_swap(version, bundle) after a successful swap.

property current_version¶: The version label this replica last swapped to (None before the first).

check_once()¶

Poll the pointer once and swap if it names a new version.

Exposed so a caller can drive the check from its own loop instead of running the polling thread.

Returns:: True if a swap happened.
Return type:: bool

start()¶

Begin polling in a daemon thread.

Returns:: self

stop(*, timeout=5.0)¶: Stop polling. Does not close the matcher; that is the caller’s responsibility.

madmatcher_pro.serving.publish_bundle_version(pointer_dir, bundle_path, *, version=None, spark=None, **info)¶

Point serving replicas at a bundle. They pick it up on their own next poll.

Call this only after the bundle is fully published: the pointer signals that the bundle is safe to load, so writing it early would send replicas at a half-written bundle.

Example:

publish_serving_artifacts("s3://.../bundles/v2", ...)
publish_bundle_version("s3://.../bundles", "s3://.../bundles/v2", version="v2")

Parameters:

pointer_dir – str | pathlib.Path Directory holding the pointer file. Every replica must be able to read it.
bundle_path – str | pathlib.Path The bundle replicas should serve.
version – str | None A label for logs and the pointer; defaults to the bundle’s directory name.
spark – SparkSession | None Only needed for a non-local pointer_dir.
**info – Extra fields recorded in the pointer (build id, row counts, and so on).

Returns:

the pointer that was written.

Return type:

dict

madmatcher_pro.serving.read_bundle_version(pointer_dir, *, spark=None)¶

The bundle replicas should currently serve, or None if no pointer is set.

Parameters:

pointer_dir – str | pathlib.Path
spark – SparkSession | None

Returns:

dict with bundle and version, or None.

Streaming runners¶

madmatcher_pro.serving.run_matching(source, sink, matcher, *, batch_size=None, passthrough=False, progress_every=0)¶

Stream every record from source through matcher into sink.

Holds only the current batch, so an unbounded source works. Not durable: for crash-resumable or multi-worker runs use run_matching_durable.

Example:

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

Parameters:

source – RecordSource | iterable of (key, record) Where records come from.
sink – ResultSink Where results go.
matcher – RealtimeMatcher | MatcherPool | ProcessMatcherPool Anything with match(record, …). A pool matches in parallel.
batch_size – int | None Use match_batch in groups of this size (one embed pass + one predict per group). None matches one record at a time.
passthrough – bool Also give the sink the incoming record, so it can store the original fields beside the match.
progress_every – int Log a line every N records (0 = silent).

Returns:

MatchRunStats

madmatcher_pro.serving.run_matching_durable(source, sink, matcher, *, checkpoint_dir, granularity='record', shard_size=1000, batch_size=None, passthrough=False, worker_id=None, spark=None, progress_every=0, max_units=None, exactly_once=False, lease_seconds=None)¶

Stream source through matcher into sink durably, claiming work so several workers can share one input without coordinating.

Re-running after a crash skips what already completed. Running the same command on several machines that can all see checkpoint_dir splits the work between them: each claims what it can, and together they cover the input.

Granularity for live streams: granularity=”record” (the default) claims one record at a time, so a record is processed the moment it arrives. granularity=”shard” batches claims for cheaper markers, but a shard is processed only once it has filled, so on a live stream a trailing partial shard would wait for records that may never come. Use “shard” only for bulk backfill of records that are all already present.

Delivery is at-least-once by default: a worker that dies after writing results but before its completion marker leaves the record claimable, and a later worker redoes it. exactly_once=True additionally skips any record already marked done, suppressing that redelivery. The remaining window (a crash between the sink write and the marker) is closed only by an idempotent sink that de-duplicates on (id1, id2).

Ordering: the source is read in order, so a worker reads past units it does not claim. That is cheap for a file source but wasteful for an expensive one, so give each machine its own source slice when reading is costly.

Output: at granularity=”record” the sink is flushed after every record (results must be durable before the completion marker), so a ParquetSink writes one part file per matched record. For compact bulk output prefer granularity=”shard” (one flush per shard) or a streaming/coalescing sink.

Example:

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

Parameters:

source – RecordSource | iterable of (key, record)
sink – ResultSink
matcher – RealtimeMatcher | MatcherPool | ProcessMatcherPool
checkpoint_dir – str | pathlib.Path Where claims and completion markers live. Must be reachable by every worker sharing the run, and not reused across different inputs (the manifest pins the configuration and refuses a mismatch).
granularity – “record” | “shard” Claim per record (safe for live streams) or per fixed-size group (cheaper markers, bulk only).
shard_size – int Records per shard, when granularity=”shard”.
batch_size – int | None Use match_batch in groups of this size (shard mode only).
passthrough – bool Also give the sink the incoming record.
worker_id – str | None Identifies this worker in claim files; defaults to host+pid.
spark – SparkSession | None Only needed for a non-local checkpoint_dir.
progress_every – int Log a line every N records (0 = silent).
max_units – int | None Stop after completing this many units (records, or shards in shard mode).
exactly_once – bool Skip records already marked done instead of re-processing them.
lease_seconds – float | None Seconds before an incomplete claim is assumed abandoned and reclaimable by another worker (None uses the store default). 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.

Returns:

MatchRunStats

Raises:

ValueError – checkpoint_dir was used for a different configuration.

class madmatcher_pro.serving.MatchRunStats¶

Bases: object

Counts and timing from a completed run.

A “unit” is whatever was claimed: one record at the default granularity, or one shard in shard mode. shards_done and shards_skipped are aliases of units_done and units_skipped.

records¶: Records matched by this worker.

matches¶: Match rows written by this worker.

units_done¶: Units this worker claimed and completed.

units_skipped¶: Units another worker had already done or claimed.

elapsed¶: Wall-clock seconds.

property shards_done¶: Alias of units_done.

property shards_skipped¶: Alias of units_skipped.

madmatcher_pro.serving.batched(source, size)¶

Group a source’s (key, record) pairs into lists of at most size.

Lazy: it holds one batch, not the whole stream, so an unbounded source can feed match_batch or a shard runner.

Example:

for chunk in batched(ParquetSource("in/"), 1000):
    ...   # chunk is a list of (key, record)

Parameters:

source – iterable of (key, record)
size – int, >= 1

Yields:

list[(key, record)]

Record sources¶

class madmatcher_pro.serving.RecordSource¶

Bases: object

Base source: iterate to get (key, record) pairs.

Subclasses implement __iter__. A key of None means the record has no id of its own and none was supplied; the orchestrator then assigns a sequence number, so a result is always attributable to its input.

Example:

class OneRecord(RecordSource):
    def __iter__(self):
        yield "only", {"name": "apple pie"}

close()¶: Release anything the source holds open. Safe to call more than once.

class madmatcher_pro.serving.IterableSource(records, *, id_col=None)¶

Bases: RecordSource

Wrap any Python iterable of records (or of (key, record) pairs).

The in-memory case, and the adapter for a caller who already has their own reader. A generator is consumed lazily, so an unbounded one is fine.

Example:

IterableSource([{"_id": 1, "name": "apple pie"}], id_col="_id")
IterableSource(stream_from_kafka())            # keys assigned downstream

Parameters:

records – iterable Records, or (key, record) 2-tuples.
id_col – str | None Column holding each record’s own id, used as its key.

class madmatcher_pro.serving.ParquetSource(path, *, id_col=None, columns=None, batch_rows=10000)¶

Bases: RecordSource

Stream records out of a parquet file or directory, in batches.

Reads through pyarrow one batch at a time, so a table far larger than memory is fine.

Example:

with ParquetSource("incoming/*.parquet", id_col="_id") as src:
    run_matching(src, sink, matcher)

Parameters:

path – str | pathlib.Path Parquet file or directory.
id_col – str | None Column holding each record’s own id, used as its key.
columns – list[str] | None Read only these columns (plus id_col). None reads all.
batch_rows – int Rows per read batch.

class madmatcher_pro.serving.CsvSource(path, *, id_col=None, chunk_rows=10000, **read_csv_kwargs)¶

Bases: RecordSource

Stream records out of a CSV, in chunks.

Example:

CsvSource("incoming.csv", id_col="_id")

Parameters:

path – str | pathlib.Path The CSV file.
id_col – str | None Column holding each record’s own id, used as its key.
chunk_rows – int Rows per read chunk.
**read_csv_kwargs – Passed to pandas.read_csv (dtype, sep, …).

class madmatcher_pro.serving.QueueSource(queue_=None, *, maxsize=1000, id_col=None)¶

Bases: RecordSource

Drain a queue.Queue that a producer thread fills.

A producer (a request handler or consumer) put`s records in and the orchestrator matches them as they arrive. Iteration ends when the producer calls `close(), so a consumer loop terminates cleanly rather than blocking forever.

Example:

src = QueueSource(maxsize=1000, id_col="_id")
threading.Thread(target=lambda: [src.put(r) for r in stream] or src.close()).start()
run_matching(src, sink, matcher)

Parameters:

queue – queue.Queue | None An existing queue to drain, or None to create one (see maxsize).
maxsize – int Bound for a created queue, so a fast producer gets backpressure instead of growing memory. 0 = unbounded.
id_col – str | None Column holding each record’s own id, used as its key.

put(record, *, key=None, timeout=None)¶

Offer one record to the source (blocks when the queue is full).

Parameters:

record – dict
key – any | None Its key; None falls back to id_col.
timeout – float | None Max seconds to wait for space.

close()¶: Signal that no more records are coming, ending iteration.

Result sinks¶

class madmatcher_pro.serving.ResultSink¶

Bases: object

Base sink: every hook is a no-op, so a subclass implements only write.

A sink must be safe to call from several worker threads at once (the pools serve in parallel), so write implementations should lock around shared state.

Example:

class CountingSink(ResultSink):
    def __init__(self): self.n = 0
    def write(self, result, record=None): self.n += len(result)

write(result, record=None)¶

Take one record’s match results.

Parameters:

result – pandas.DataFrame [id1, id2, prediction, confidence], possibly empty (a record with no match still arrives, so a sink can tell “no match” from “not processed”).
record – dict | None The incoming record, when the caller asked for passthrough.

Returns:

None

flush()¶: Push any buffered results to their destination. Called periodically and before close.

close()¶: Flush and release the destination. Safe to call more than once.

class madmatcher_pro.serving.MemorySink¶

Bases: ResultSink

Keep every result frame in memory.

Grows with the stream, so it suits bounded runs, tests, and interactive use. For an unbounded stream use CallbackSink or ParquetSink.

Example:

sink = MemorySink()
run_matching(source, sink, matcher)
sink.frames        # -> list[DataFrame]
sink.to_frame()    # -> one concatenated DataFrame

to_frame()¶

All results as one DataFrame (empty with the right columns if none).

Returns:: pandas.DataFrame [id1, id2, prediction, confidence]

class madmatcher_pro.serving.DataFrameSink¶

Bases: MemorySink

Accumulate results into one pandas.DataFrame, for handing straight on to the next step of a pipeline.

Example:

sink = DataFrameSink()
run_matching(source, sink, matcher)
df = sink.result        # -> one DataFrame of every match

property result¶: Every match so far as a single DataFrame.

class madmatcher_pro.serving.CallbackSink(fn)¶

Bases: ResultSink

Hand each result to your own function as it is produced.

For custom handling: post to an API, push to a queue, update a store. Holds nothing, so it works on an unbounded stream.

Example:

CallbackSink(lambda result, record: publish(result))

Parameters:: fn – callable Called as fn(result, record). A callable taking one argument is also accepted and receives just the result.

class madmatcher_pro.serving.ParquetSink(out_dir, *, batch_rows=10000, part_prefix=None, passthrough_cols=None, spark=None)¶

Bases: ResultSink

Append results to parquet, batched, through backend-agnostic IO.

Writes one part file per batch under out_dir (local, HDFS, S3, GCS, whatever FileSystemStore resolves). Each part is written to a .tmp sibling and renamed, so a reader never sees a half-written part. Holds at most batch_rows in memory, so an unbounded stream is fine.

Example:

with ParquetSink("out/matches", batch_rows=50_000) as sink:
    run_matching(source, sink, matcher)

Parameters:

out_dir – str | pathlib.Path Directory to write part files into (created if absent).
batch_rows – int Rows to buffer before writing a part.
part_prefix – str Filename prefix, so several writers into one dir do not collide. Defaults to a per-instance value.
passthrough_cols – list[str] | None Record fields to store beside each match. None writes results only; naming fields multiplies the bytes per match, so it is opt-in.
spark – SparkSession | None Only needed for a non-local out_dir.

class madmatcher_pro.serving.FanOutResultSink(sinks)¶

Bases: ResultSink

Send every result to several sinks at once.

Useful for “keep it in a DataFrame and also write it durably” in one configuration. Each sink is isolated: if one raises, the others still receive the result, and the error is re-raised afterward rather than silently swallowed.

Example:

sink = FanOutResultSink([DataFrameSink(), ParquetSink("out/matches")])

Parameters:: sinks – iterable of ResultSink

Batch-to-serving handoff¶

class madmatcher_pro.serving.BatchToServing(*, spool_dir, bundle_dir, id_col=None, checkpoint_dir=None, segment_rows=1000)¶

Bases: object

Coordinate the batch build -> live serving handoff, losing no record in between.

Example:

h = BatchToServing(spool_dir="spool/", bundle_dir="bundles/v1")

# while the batch build runs, records keep arriving
h.accept({"_id": 1, "name": "apple pie"})

# the build finishes; publish from its artifacts and go live
h.publish(lambda out: publish_serving_artifacts(
    out, indexed_table=corpus, features=features, model=model,
    sparkly_index_path=index_path,        # already built by the batch job
    prebuilt_feature_table=table_a,       # already preprocessed by featurize
    fill_na=0.0))
h.start_serving(matcher)
h.drain(sink)                             # the backlog, matched
h.match({"_id": 99, "name": "banana bread"})   # live from here on

Parameters:

spool_dir – str | pathlib.Path Where records accepted before serving is ready are buffered durably.
bundle_dir – str | pathlib.Path Where publish writes the bundle.
id_col – str | None Column holding a record’s own id, used as its key.
checkpoint_dir – str | pathlib.Path | None Makes the backlog drain resumable (per-record claims), so a crash mid-drain resumes rather than restarting the backlog. None drains without durability.
segment_rows – int Records per spool segment.

property state¶: One of accepting, publishing, draining, serving, closed.

property spool¶: The underlying RecordSpool (also a RecordSource).

property matcher¶: The live matcher once serving has started, else None.

accept(record, *, key=None)¶

Buffer one record durably before serving is ready.

Safe to call from many threads. Once serving has started, prefer match; accept still works (it spools), so a producer need not track which phase the handoff is in.

Parameters:

record – dict
key – any | None The record’s key. None falls back to id_col.

Returns:

The record’s key.

publish(publish_fn)¶

Build the serving bundle from the finished batch job’s artifacts.

publish_fn receives the output directory and performs the publish, typically by calling publish_serving_artifacts with the indices and preprocessed corpus your batch job already produced.

Example:

h.publish(lambda out: publish_serving_artifacts(
    out, indexed_table=corpus, features=features, model=model,
    sparkly_index_path=index_path, prebuilt_feature_table=table_a,
    fill_na=0.0))

Parameters:: publish_fn – callable Called as publish_fn(bundle_dir).
Returns:: the bundle directory.
Return type:: str

start_serving(matcher)¶

Go live with a loaded matcher (or pool).

Parameters:: matcher – RealtimeMatcher | MatcherPool | ProcessMatcherPool | HotSwappableMatcher
Returns:: self

drain(sink, *, batch_size=None, passthrough=False, clear=True, **kwargs)¶

Match everything spooled during the build, into sink.

Uses the durable runner when a checkpoint_dir was given, so an interrupted drain resumes per record rather than restarting the backlog.

Parameters:

sink – ResultSink
batch_size – int | None Match in groups of this size.
passthrough – bool Also give the sink each incoming record.
clear – bool Delete drained spool segments afterwards. False keeps them (to re-drain into a second sink, or to inspect).
**kwargs – Passed to the underlying runner.

Returns:

MatchRunStats

match(record, *, key=None, **kwargs)¶

Match one record live, once serving has started.

Before serving has started this raises; use accept to buffer the record until the bundle is ready.

Parameters:

record – dict
key – any | None The record’s key, used to stamp id2 on the result.

Returns:

pandas.DataFrame [id1, id2, prediction, confidence]

Raises:

RuntimeError – serving has not started yet.

close()¶

Stop accepting and seal the spool.

Does not close the matcher; the caller owns its lifecycle, which may outlive this handoff.

class madmatcher_pro.serving.RecordSpool(directory, *, segment_rows=1000, id_col=None)¶

Bases: RecordSource

A durable append-and-drain buffer of (key, record) pairs.

Safe for many threads to append to at once. Also a RecordSource, so it plugs straight into run_matching / run_matching_durable.

Example:

spool = RecordSpool("spool/incoming")
spool.append({"name": "apple pie"}, key="req-1")   # while batch is running
...
spool.close_segment()                  # stop accepting, seal what is buffered
run_matching(spool, sink, matcher)     # drain it once the bundle exists

Parameters:

directory – str | pathlib.Path Where segment files live (created if absent).
segment_rows – int Records per segment before it is sealed and a new one started. Smaller segments become readable sooner; larger ones cost fewer files.
id_col – str | None Column holding a record’s own id, used as its key when append is given none.

append(record, *, key=None)¶

Buffer one record durably.

The record is flushed to the OS on every append, so a process crash loses at most what the OS had not yet written.

Parameters:

record – dict
key – any | None The record’s key. None falls back to id_col, then to a generated key.

Returns:

The record’s key.

close_segment()¶: Seal the segment being written, making it readable. Idempotent.

close()¶: Seal and stop writing. Safe to call more than once.

property appended¶: Records appended by this instance (not the total on disk).

pending()¶

How many records are readable right now (in sealed segments).

Returns:: int

source_for(segments)¶

A one-shot RecordSource over exactly segments (a snapshot from segments()).

Draining over a fixed snapshot lets a drain match precisely the segments it will clear, so a concurrent append that seals a new segment mid-drain is neither double-matched nor lost. A segment deleted meanwhile is skipped.

Parameters:: segments – list[pathlib.Path]
Returns:: RecordSource

segments(*, sealed_only=True)¶

Segment files, oldest first.

Parameters:: sealed_only – bool Only complete segments (the default). False also lists the open one, for recovering a spool whose writer died without sealing.
Returns:: list[pathlib.Path]

recover_open_segments()¶

Seal segments left open by a writer that died, so they can be drained.

A crash leaves the in-progress segment as .open, which the reader skips. This seals any such segment (except the one this instance is writing) so its records are recovered; a torn final line is skipped on read.

Returns:: segments recovered.
Return type:: int

clear_drained(segments=None)¶

Delete sealed segments once their records have been matched.

Call this only once the results are durable: it is the point of no return for the backlog. run_matching_durable records per-record completion markers, so a drain interrupted before this is safe to re-run.

Parameters:: segments – list[pathlib.Path] | None The specific segments to delete, typically the snapshot a drain took at its start so a concurrent append that sealed a new segment is not deleted before it is matched. None deletes every currently sealed segment.
Returns:: segments deleted.
Return type:: int

Durable claims¶

Filesystem-only per-unit claims behind run_matching_durable; use these directly only for a custom durable runner.

class madmatcher_pro.serving.RecordClaimStore(root, *, fs, worker_id, lease_seconds=1800.0)¶

Bases: _BaseClaimStore

Claim work one record at a time.

The right granularity for a live stream, where records arrive singly and waiting for a group to fill would stall them. Costs two small files per record (a claim and a completion marker); for a bulk backfill of many already-present records, use ShardClaimStore.

Example:

store = RecordClaimStore("ckpt/run-7", fs=fs, worker_id="host-1")
if store.claim(key):
    ...                       # match and write results
    store.complete(key, matches=3)

class madmatcher_pro.serving.ShardClaimStore(root, *, fs, worker_id, lease_seconds=1800.0)¶

Bases: _BaseClaimStore

Claim work a fixed-size group of records at a time.

Cheap markers for bulk backfill. Not for a live stream: a shard is processed only once it has filled, so a trailing partial shard waits for records that may never come.

Example:

store = ShardClaimStore("ckpt/backfill", fs=fs, worker_id="host-1")
if store.claim(12):           # shard index
    ...
    store.complete(12, records=1000)

Advanced: resident probes and feature state¶

The pieces RealtimeMatcher composes, usable directly for a custom serving layer.

class madmatcher_pro.serving.ResidentSparklyProbe(index, query_spec, limit, fill_keys=frozenset({}))¶

Bases: ResidentProbe

Keyword (word-based) search over your records, kept loaded in memory so it can quickly find candidate records that look similar to one incoming record, with no Spark needed.

Example:

probe = ResidentSparklyProbe.load("bundle/blocking/sparkly", limit=50)
probe.probe({"name": "apple pie"})
# -> array([12, 42, 7])

class madmatcher_pro.serving.ResidentSemanticProbe(*, centroids, dim, quantizer, target_dim, id_col, postings_dir, provider, fields, serializer_cfg=None, col_types=None, nprobe=32, limit=50, staged_cells=None)¶

Bases: ResidentProbe

Meaning-based (embedding) search over your records, one record at a time, in process.

Given an incoming record, this finds the stored records whose meaning is most similar to it (not just those sharing the same words) and returns them with a similarity score. Serving builds one from a persisted semantic index at publish time; you normally use it through RealtimeMatcher, but it can be used directly.

Example:

probe = ResidentSemanticProbe.load(
    "bundle/blocking/semantic",
    provider=my_embedder, fields=["name", "description"])
ids, scores = probe.probe({"name": "apple pie"})

With a large index that uses an approximate (HNSW) coarse quantizer, a record near a region boundary may occasionally return a slightly different candidate set than the batch engine; the similarity scores are always exact.

class madmatcher_pro.serving.ResidentDelexProbe(nodes, sink, search_cols, in_schema, *, limit=None)¶

Bases: ResidentProbe

Runs a Delex blocking program (a set of matching rules) in memory to find the candidate records for one incoming record, with no Spark job on the lookup path.

Serving builds one from your records at publish time; you normally use it through RealtimeMatcher, but it can be used directly.

Example:

probe = ResidentDelexProbe.build(corpus_spark_df, blocking_program)
probe.probe({"name": "apple pie"})   # -> array([12, 87, 103])

class madmatcher_pro.serving.ResidentEmbeddingStore(store, dim, *, provider=None, fields=None, serializer_cfg=None, col_types=None, target_dim=None, staging_dir=None)¶

Bases: object

A loaded table of your records’ embedding vectors, keyed by id, that computes the semantic-similarity (cosine) feature for a record against a set of candidates.

Serving builds one from an embedding table at publish time; you normally use it only through RealtimeMatcher, but it can be used directly.

Example:

store = ResidentEmbeddingStore.load(
    "bundle/embedding_store", provider=my_embedder, fields=["name"])
store.cosine_for_ids([12, 99], record={"name": "apple pie"})

classmethod load(embedding_dir, *, id_col='_id', embedding_col='embedding', provider=None, fields=None, serializer_cfg=None, col_types=None, target_dim=None, dest_dir=None)¶

Load an embedding table (a folder of (id, vector) rows) into memory.

Example:

store = ResidentEmbeddingStore.load(
    "bundle/embedding_store",
    id_col="_id", embedding_col="embedding",
    provider=my_embedder, fields=["name", "brand"])

Uses the ready-to-mmap table publish_serving_artifacts writes into the bundle: it is mapped read-only, so several matchers over the same bundle (for example a ProcessMatcherPool’s worker processes) share one set of physical pages. A bundle published without that table falls back to staging a private copy and warns; re-publish to get the shared table.

Parameters:

embedding_dir – str | pathlib.Path The bundle’s blocking/embedding_store folder. A direct path to a parquet embedding table also works (it is then staged privately).
id_col – str Name of the id column in that table.
embedding_col – str Name of the vector column in that table.
provider – an embedding provider, or None Used to turn an incoming record into a vector. If None, every call to cosine_for_ids must pass a precomputed probe_embedding.
fields – list[str] | None The record fields to build the incoming record’s embedding text from.
serializer_cfg – dict | None How to format that text (from the manifest); None uses the default.
col_types – dict | None Column-type hints so an incoming value is formatted like the stored records; None is fine for plain text fields.
target_dim – int | None Truncate an incoming embedding to this width before scoring (for Matryoshka embeddings); None keeps the full width.
dest_dir – str | None Where to stage the table when a private copy is needed; None uses a temp dir that close cleans up. Unused when the bundle already holds the shared table.

Returns:

ResidentEmbeddingStore: Ready to answer cosine_for_ids.

serialize(record)¶

Turn one record into the text its embedding is built from (the same way the stored records were turned into text).

Example:

store.serialize({"name": "Apple Pie", "brand": "Acme"})
# -> "[COL] name [VAL] apple pie [COL] brand [VAL] acme"

Parameters:

record – dict The incoming record as a column->value mapping.

Returns:

str: The embedding text for that record.

cosine_for_ids(ids, *, record=None, probe_embedding=None, fill=0.0)¶

Semantic-similarity score of an incoming record against each of the given record ids, as a number from -1 to 1 (higher means more similar).

The incoming record is turned into a vector (embedded from record, or taken from probe_embedding), and each id’s stored vector is compared to it. An id that isn’t in the store gets fill and a one-time warning is logged (add its vector to the embedding table and re-publish to fix).

Example:

store.cosine_for_ids([12, 42, 99], record={"name": "apple pie"})
# -> array([0.83, 0.61, 0.  ])   # id 99 wasn't in the store -> fill (0.0)

Parameters:

ids – sequence of int The record ids to score against the incoming record.
record – dict | None The incoming record to embed. Required unless probe_embedding is given.
probe_embedding – numpy.ndarray | None A precomputed vector for the incoming record, skipping the embedder.
fill – float The value used for an id the store doesn’t have a vector for.

Returns:

numpy.ndarray of float, one per id in ids (same order): The similarity for each id, or fill where the store has no vector.

close()¶

Free the loaded embedding table and delete its temporary files. Safe to call more than once.

Example:

store.close()

Returns:: None

class madmatcher_pro.serving.ServedState(corpus, features, fill_na=0.0, column_types=None)¶

Bases: object

Your saved record-comparison state, loaded into memory and ready to compare an incoming record against your records while serving.

You normally get one from ServedState.load and hand it to featurize_one.

Example:

served = ServedState.load("bundle/feature_state")
served.known_ids([1, 2, 999999])  # -> array([1, 2])

classmethod load(dir_path, fill_na=0.0)¶

Load the saved record-comparison state (from publish_feature_state) back into memory so it can score incoming records.

Example:

served = ServedState.load("bundle/feature_state")
# -> a ready-to-use ServedState

Parameters:

dir_path – str | pathlib.Path Folder the state was saved to by publish_feature_state.
fill_na – float Value to use for a comparison that can’t be computed (e.g. a field the incoming record is missing).

Returns:

ServedState: The loaded state, ready for known_ids and featurize_one.

known_ids(ids)¶

Keep only the record ids that actually exist in your data, dropping any unknown id so it is skipped instead of causing an error later.

Example:

served.known_ids([1, 2, 999999])
# -> array([1, 2])

Parameters:

ids – sequence of int Candidate record ids to check against your data.

Returns:

numpy.ndarray of int64: The ids from ids that are present (same order), unknowns removed.

madmatcher_pro.serving.featurize_one(probe_record, candidate_ids, served)¶

Turn one incoming record plus a set of candidate records into the comparison feature vectors a trained matcher scores (one row of numbers per candidate).

Example:

fvs = featurize_one({"name": "apple pie"}, [12, 42], served)
# -> (2, n_features) float32 array

Parameters:

probe_record – dict (field name -> value) The incoming record to compare against the candidates.
candidate_ids – sequence of int Ids of your records to compare the incoming record against.
served – ServedState The loaded state from ServedState.load.

Returns:

numpy.ndarray of shape (len(candidate_ids), n_features), float32: One comparison feature vector per candidate id, in the same order.

madmatcher_pro.serving.publish_feature_state(A, B, features, out_dir, prebuilt_table_a=None)¶

Build and save the record-comparison state for your data, so the real-time serving layer can load it later and score incoming records without reprocessing the whole dataset.

Example:

publish_feature_state(corpus_df, None, features, "bundle/feature_state")
# writes the reusable state into bundle/feature_state/ ; returns None

Parameters:

A – Spark DataFrame Your records (the indexed data that incoming records are compared against).
B – Spark DataFrame, or None A second table to compare A against; pass None for a single table (the common case).
features – list of feature objects The record comparisons to compute (e.g. from create_features, or the ones a trained matcher already carries).
out_dir – str | pathlib.Path Folder to write the saved state into (created if it does not exist).
prebuilt_table_a – MemmapDataFrame | None An already-preprocessed corpus table, to avoid rebuilding it. A batch featurize run produces exactly this (via build_preprocessed), so a pipeline that goes on to publish can pass it straight through. It must have been built from A with these same features, or serving would score against different state than it advertises. Note that a batch-built table was registered with the SparkContext (addFile) by the featurize run that made it, so reusing it here keeps that registration – one more reason to publish in its own Spark application rather than as a tail stage.

Returns:

None

madmatcher_pro.serving.reciprocal_rank_fusion_local(sources, limit, *, kappa=60, cosine_fill=0.0)¶

Merge several blockers’ candidate lists for one record into a single ranked list, so a record found by more than one blocker ranks higher.

Example:

fused_ids, scores = reciprocal_rank_fusion_local(
    [{"ids": [12, 42], "scores": [3.1, 2.0], "weight": 0.5, "semantic": False},
     {"ids": [42, 7],  "scores": [0.9, 0.8], "weight": 0.5, "semantic": True}],
    limit=50)
# fused_ids: array([42, 12, 7]); scores: the meaning-similarity per id

Parameters:

sources – list of dict One entry per blocker, each with keys: ids (its candidate ids), scores (each id’s ranking strength, or None if the blocker gives no order), weight (how much to weight this blocker, default 0.5), semantic (True if scores are meaning-similarity values to carry through), and optional shared_rank (True to treat every id as equally ranked, for an unordered blocker).
limit – int The most candidates to keep after merging.
kappa – int A rank-smoothing constant (larger flattens the contribution of rank).
cosine_fill – float The similarity value recorded for a candidate no meaning-based blocker ranked.

Returns:

tuple (numpy.ndarray of int ids, numpy.ndarray of float | None): The merged candidate ids, best first (at most limit), and each id’s meaning-similarity score. The second value is None when no blocker was a meaning-based one.

Probe serializers¶

How an incoming record’s fields become the one text a resident embedder embeds, matching the batch-time corpus serialization.

class madmatcher_pro.serving.ProbeSerializer¶

Bases: ABC

Serialize one probe record to embedding text. Subclasses differ only in how each field is formatted and how the parts join; the base handles the shared walk: iterate fields in order, convert each value to text (using the column’s Spark type from col_types for parity with the corpus), drop null or uncastable values, and lower-case when self.lower.

class madmatcher_pro.serving.DittoProbeSerializer(*, col_token='[COL]', val_token='[VAL]', lower=True)¶

Bases: ProbeSerializer

Formats a record as [COL] <field> [VAL] <value> for each field, joined by spaces, dropping empty fields. This is the serving twin of the batch semantic.serialize.DittoSerializer, so a probe’s text matches how the corpus was embedded.

Example:

DittoProbeSerializer().serialize(
    {"name": "Apple Pie", "brand": "Acme"}, ["name", "brand"])
# -> "[COL] name [VAL] apple pie [COL] brand [VAL] acme"

Create a Ditto serializer, optionally customizing the marker tokens and whether values are lower-cased.

Example:

DittoProbeSerializer(col_token="[COL]", val_token="[VAL]", lower=True)

Parameters:

col_token – str The marker placed before each field name (keyword-only).
val_token – str The marker placed before each field value (keyword-only).
lower – bool Whether to lower-case each value before formatting (keyword-only).

Returns:

None

class madmatcher_pro.serving.PlainConcatProbeSerializer(*, sep=' ', lower=True)¶

Bases: ProbeSerializer

Joins the field values directly with a separator, with no markers, keeping empty strings and dropping only null fields. This is the serving twin of the batch semantic.serialize.PlainConcatSerializer, so a probe’s text matches how the corpus was embedded.

Example:

PlainConcatProbeSerializer(sep=" ").serialize({"name": "Apple Pie"}, ["name"])
# -> "apple pie"

Create a plain-concatenation serializer, optionally setting the separator and whether values are lower-cased.

Example:

PlainConcatProbeSerializer(sep=" ", lower=True)

Parameters:

sep – str The string placed between field values (keyword-only).
lower – bool Whether to lower-case each value before joining (keyword-only).

Returns:

None

madmatcher_pro.serving.register_serializer(cls)¶

Record a serializer class under its kind name so it can be saved and reloaded later. Use as a decorator on a ProbeSerializer subclass.

Example:

@register_serializer
class MyScheme(ProbeSerializer):
    kind = "my_scheme"
# serializer_from_manifest({"kind": "my_scheme"}) can now rebuild it

Parameters:

cls – type A ProbeSerializer subclass that sets a non-empty kind class attribute (an empty or missing kind is rejected).

Returns:

type: The same class, unchanged, so it works as a decorator.