# Semantic Blocking Tutorial *Version 1.0, July 2, 2026* This tutorial describes how to perform **semantic (dense) blocking** for entity matching with `madmatcher-pro`. We first discuss how semantic blocking works and how it relates to the lexical blocking you may already know from Sparkly and Delex, and then provide a step-by-step guide that walks you through embedding your records, building a dense index, searching it, and combining it with a lexical blocker. You should read the "How Semantic Blocking Works" section first, because it defines the terms used later. Like Sparkly, semantic blocking uses Spark, which can run on a single machine or a cluster. On a single machine Spark treats each CPU core as a worker. For this guide we use Spark on a local machine, so a "Spark cluster" here means all of your local cores. --- ## How Semantic Blocking Works Blocking reduces the search space for matching: instead of comparing every record in one table against every record in another, we find, for each record, a small set of likely matches called **candidates**. Lexical blockers (Sparkly's BM25, Delex's rules) find candidates by **shared tokens** — two records are candidates if they share enough words or n-grams. Semantic blocking finds candidates by **meaning**: it turns each record into a vector (an **embedding**) so that records with similar meaning sit close together, then retrieves each record's nearest neighbors. This catches matches that lexical blocking misses — "`Sony WH-1000XM5 headphones`" and "`Sony wireless over-ear noise-cancelling headset`" share almost no tokens but describe the same product. Because the two approaches catch different matches, they are strongest **combined**, which this guide also covers. We will use the same terminology as the other blocking tutorials: - The **indexing table** (table A) is the table an index is built over; its records supply the candidates. - The **search table** (table B) is the table we search *for*: for each of its records we retrieve nearest candidates from table A. - A **candidate** is a possible match retrieved for a record. - A **record** (or **tuple**) is a row; a **value** is one column's data in that row. Either of your two tables can play either role; the index is built once over the indexing table and then reused for every search. For **deduplication** (finding matches within a single table), the indexing and search tables are the same table (covered in Step 5). For this tutorial we will assume two small product tables, A and B, as illustrated below. **Table A (Indexing Table)** | _id | name | description | |-----|------|-------------| | 0 | Sony WH-1000XM5 headphones | over-ear noise cancelling, 30h battery | | 1 | Logitech MX Master 3S | ergonomic wireless mouse | | … | … | … | **Table B (Search Table)** | _id | name | description | |-----|------|-------------| | 0 | Sony wireless over-ear noise-cancelling headset | bluetooth headphones, long battery life | | 1 | wireless ergonomic mouse, USB-C | precision scroll wheel | | … | … | … | Consider record 0 in Table B, "`Sony wireless over-ear noise-cancelling headset`". It shares almost no words with record 0 in Table A, "`Sony WH-1000XM5 headphones`", so a purely lexical blocker might not pair them. Semantic blocking embeds both records into vectors that land close together because they *mean* the same thing, so record 0 from Table A is retrieved as a candidate for record 0 in Table B. That is the kind of match this guide is built to find. Semantic blocking has three moving parts, and this guide follows them in order: 1. **Embedding** turns each record's text into a vector. You embed each table once and reuse the vectors — building or retuning the index never re-embeds. 2. **The index** organizes the vectors so that each query's nearest neighbors can be found quickly, without comparing it against every record. 3. **The search** retrieves, for each search record, its nearest table-A candidates. The output is one row per search record, with the same shape the lexical blockers produce: | Column | Type | Meaning | |---|---|---| | `id2` | long | the search record's id | | `id1_list` | array<long> | its candidate ids from table A, best first | | `scores` | array<float> | the cosine score for each candidate; `scores[i]` goes with `id1_list[i]` | | `search_time` | float | always `0.0` for semantic; present only so the schema matches the lexical blockers' output (search runs as one batch, so there is no per-query time to report) | | `mm_sem_scores` | array<float> | the per-pair cosine, carried so the matcher can reuse it as a feature | Because the shape matches the lexical blockers, a semantic candidate set can be fed straight into MatchFlow featurization, or fused with a lexical candidate set first (Step 6). --- ## Crash Recovery and Progress Tracking Semantic blocking is one of `madmatcher-pro`'s premium features. Two capabilities apply to its long-running operations, both opt-in (off by default): - **Crash recovery** lets a long-running job resume after a failure instead of restarting from the beginning. In semantic blocking it applies to the three heavy steps: **embedding** (`create_embeddings`), the **index build** (`SemanticIndex.upsert_docs`), and the **search** (`SemanticSearcher.search`). Each of those steps below has a "Crash recovery (premium)" note showing how to enable it. - **Progress tracking** shows your operations live on a web dashboard while they run. You launch it once near the top of your script. The [Reference: Crash Recovery and Progress Tracking](#reference-crash-recovery-and-progress-tracking) section at the end covers how to launch and read the dashboard, and documents every argument in full. --- ## Step-by-Step Walkthrough This section walks through writing a Python script to perform semantic blocking, in the following steps: 1. Creating a Spark Session 2. Loading in Data 3. Embedding the Records 4. Building the Index 5. Searching 6. Fusing Lexical and Dense Blocking 7. Saving Blocking Output A complete, runnable version of everything below ships with the package as `scripts/demos/semantic/demo_fusion_blocking.py` (embedding through fusion); a semantic-only variant without Step 6 is `scripts/demos/semantic/demo_semantic_blocking.py`. Run one first to confirm your install, then read on to understand what each part does. ### Step 1: Creating a Spark Session Before we can use semantic blocking, we need to set up a `SparkSession`. A `SparkSession` defines the location of the Spark master (driver node), a name, and other optional settings. For this guide, we set the master (driver node) to our local machine and give it the name 'Semantic Blocking' to identify it. To create the `SparkSession`, we write the following in our Python script: ```python from pyspark.sql import SparkSession spark = SparkSession.builder\ .master('local[*]')\ .appName('Semantic Blocking')\ .getOrCreate() ``` From now on, whenever we need Spark operations in our script, we use this `SparkSession` instance, which we called "spark" above. If you think you will need to configure more options, such as storage limits or other `SparkSession` settings, see the [SparkSession builder documentation](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/spark_session.html). You can also turn on live **progress tracking**: create a `MadMatcherSession` near the top of your script, and every operation that follows appears on a small web dashboard. See [Viewing progress on the dashboard](#viewing-progress-on-the-dashboard) in the reference section for how to launch it, the builder options, and the read-only viewer. ### Step 2: Loading in Data **Required columns.** Semantic blocking makes few assumptions about your schema, with two requirements: - Each table must have an **id column**: a column that has a value for every record, whose values are unique and are 32- or 64-bit integers. It can be named anything (the demo uses `_id`), and you pass its name to the operations in the later steps. - Each table must have the **text column(s) you intend to embed** (`name`, `description`, and so on). You choose these fields in Step 3. The indexing table (A) and the search table (B) do not have to share the same columns, but the fields you embed on each side should describe the same kind of information so their vectors are comparable. The tables start on disk and are loaded into Spark DataFrames. We use the **`abt_buy`** demo dataset that ships with the package: two product tables to link, `table_a` (the indexing table) and `table_b` (the search table), each with columns `_id` (a unique long id), `name`, `description`, and `price`, plus `gold` (the true `id1`/`id2` matches, used only to score recall at the end). Parquet files carry their own schema, so no options are needed; to load CSVs instead, use `spark.read.csv(..., header=True)` and rename columns with `withColumnRenamed` as needed (see the Spark API for more). ```python import pyspark.sql.functions as F table_a = spark.read.parquet("abt_buy/table_a.parquet") # indexing table; gold id1 lives here table_b = spark.read.parquet("abt_buy/table_b.parquet") # search table; gold id2 lives here gold = spark.read.parquet("abt_buy/gold.parquet") ``` **Validate the tables first.** Immediately after loading, confirm the id columns are well-formed with `check_tables_manual`, so a schema problem is caught here rather than after you have spent time embedding: ```python from madmatcher_pro.sparkly.utils import check_tables_manual check_tables_manual(table_a=table_a, id_col_table_a="_id", table_b=table_b, id_col_table_b="_id") ``` `check_tables_manual` verifies that the named id column exists in each table and that its values are 32- or 64-bit integers, present for every row, and unique; it raises a `ValueError` if any of those does not hold. Note your id column names for the later steps — every operation takes an `id_col` argument (here `"_id"`). This guide loads the demo tables from Parquet and keeps the data step lean. For the full data-loading and preprocessing guidance — loading CSV vs Parquet, the CSV `header` option, inspecting a schema with `printSchema`, and renaming columns with `withColumnRenamed` — see **Step 2 of the Sparkly tutorial**; it applies unchanged to the tables you use here. ### Step 3: Embedding the Records **Choosing the fields to embed.** First decide which columns describe each record. Embedding all of a table's text columns is a good default; here we take every string column except the id: ```python from pyspark.sql import types as T fields = [f.name for f in table_a.schema.fields if f.name != "_id" and isinstance(f.dataType, T.StringType)] ``` Here `fields` is read from table A's schema and reused for table B because the demo tables share a schema. If your A and B columns differ, choose the embed fields for each table separately, picking columns that describe the same kind of information on each side so the two sets of vectors stay comparable. **How records become text.** Before a record is embedded, its `fields` are serialized into a single string. The default scheme marks each field so the model can tell them apart. For a record with `name = "Sony WH-1000XM5 headphones"` and `description = "wireless over-ear noise-cancelling headset"`, the serialized text is: ```text [COL] name [VAL] sony wh-1000xm5 headphones [COL] description [VAL] wireless over-ear noise-cancelling headset ``` Empty fields are dropped, and the text is lowercased (most models are uncased). You can change this by passing a different `serializer` to `create_embeddings`: `DittoSerializer` (the default, shown above) or `PlainConcatSerializer` (the field values joined by spaces, with no markers). **Prerequisites.** Semantic blocking installs with the `[semantic]` extra: `pip install "madmatcher-pro[semantic]"`. The first run downloads the model (here `BAAI/bge-small-en-v1.5`) from HuggingFace, so it needs network access the first time. **Embedding.** `create_embeddings` serializes the chosen `fields`, embeds each record, and writes an id-keyed vector table `(_id, embedding)` to `out_path`. You give it a **provider** — here a `SentenceTransformerProvider` wrapping any sentence-transformers model id. Embed each table once; the vectors are reused when you build or retune the index. ```python from madmatcher_pro.semantic import SentenceTransformerProvider, create_embeddings provider = SentenceTransformerProvider(model_id="BAAI/bge-small-en-v1.5") vec_a = create_embeddings(df=table_a, fields=fields, provider=provider, out_path="work/emb_a") vec_b = create_embeddings(df=table_b, fields=fields, provider=provider, out_path="work/emb_b") ``` `SentenceTransformerProvider` is one of several built-in providers. Other built-in providers, hosted APIs, passing a bare model-id string, and writing your own provider are all covered in the **Appendix**. **Crash recovery (premium).** Make the embedding resumable by setting `enable_crash_recovery=True` and a durable `checkpoint_dir`: ```python vec_a = create_embeddings( df=table_a, fields=fields, provider=provider, out_path="work/emb_a", enable_crash_recovery=True, checkpoint_dir="ckpt/emb_a", ) ``` If the job is interrupted, run the same call again with the same `checkpoint_dir`; the records already embedded are skipped. `create_embeddings`'s crash-recovery arguments — `enable_crash_recovery`, `checkpoint_dir`, `n_groups`, `rows_per_group`, `validate_input`, and `show_progress` — are documented in the reference section (its other arguments, such as the input table, `fields`, `provider`, and `out_path`, are shown above). ### Step 4: Building the Index `SemanticIndex` builds and persists a dense index over the embedding column. You create the index object, then build it with `upsert_docs`, which fits the index to your vectors and writes it under the path you give: ```python from madmatcher_pro.semantic import SemanticIndex index = SemanticIndex(index_path="work/index", id_col="_id") index = index.upsert_docs(df=vec_a) ``` To reopen a built index in a later script, use `SemanticIndex.load(index_path="work/index")` instead of rebuilding it. **Crash recovery (premium).** Enable it the same way; the build then runs in committed groups and resumes from where it stopped: ```python index = SemanticIndex(index_path="work/index", id_col="_id") index = index.upsert_docs(df=vec_a, enable_crash_recovery=True, checkpoint_dir="ckpt/index") ``` The index's internal model is **fitted once**, on the first run, and stored in the checkpoint. A resume reuses that stored model rather than refitting it, so every group is built consistently before and after a crash. Because a resume reuses what is already in the `checkpoint_dir`, re-running never rebuilds from scratch — to build a **new** index (over different data or with different settings), first delete the existing index directory and its `checkpoint_dir`, then run the build again. `upsert_docs`'s crash-recovery arguments — `enable_crash_recovery`, `checkpoint_dir`, `n_groups`, `rows_per_group`, and `validate_input` — are documented in the reference section. (Its first argument is the vector DataFrame to index, shown above.) ### Step 5: Searching `SemanticSearcher` searches a built index. You construct it from the index, build a `SemanticQuerySpec` for the recall knobs, then call `search` to retrieve each record's nearest candidates: ```python from madmatcher_pro.semantic import SemanticSearcher, SemanticQuerySpec spec = SemanticQuerySpec(nprobe=32) searcher = SemanticSearcher(index=index) candidates = searcher.search(search_df=vec_b, spec=spec, limit=20, id_col="_id") candidates.show() ``` `search` takes: - **`search_df`** — the query vector table (table B's embeddings): one row per record to find candidates for. - **`spec`** — a `SemanticQuerySpec` carrying the recall knobs (below). - **`limit`** — the number of candidates to return per query record. - **`id_col`** (default `"_id"`) — the name of the id column in `search_df`. It also takes `exclude_self` (see Deduplication below) and the crash-recovery arguments (in the reference section). `SemanticQuerySpec` takes: - **`nprobe`** (default `32`) — the search-thoroughness dial, and the knob that matters. Higher values search more of the index per query, raising recall at some cost in speed. Raise it until recall stops improving. - **`rerank_window`** (default `200`) and **`ef_search`** (default `None`) — advanced knobs the standard setup does not use; leave them at their defaults. Each row holds a query id and its ranked candidate list, as described in "How Semantic Blocking Works". To score recall against the gold matches, explode the lists into pairs and intersect: ```python pairs = candidates.select(F.explode("id1_list").alias("a"), F.col("id2").alias("b")) recall = gold.intersect(pairs).count() / gold.count() print(f"recall: {recall:.4f}") ``` **Deduplication.** When the indexing table *is* the search table (you are finding duplicates within one table), every record retrieves itself at cosine 1.0. Pass `exclude_self=True` to drop each record from its own candidate list; the parallel `scores` stay aligned automatically: ```python candidates = searcher.search(search_df=vec_a, spec=spec, limit=20, id_col="_id", exclude_self=True) ``` **Crash recovery (premium).** As with the other steps: ```python candidates = searcher.search( search_df=vec_b, spec=spec, limit=20, id_col="_id", enable_crash_recovery=True, checkpoint_dir="ckpt/search", ) ``` `search`'s crash-recovery arguments — `enable_crash_recovery`, `checkpoint_dir`, `n_groups`, `rows_per_group`, and `validate_input` — are documented in the reference section. (Its search table, `spec`, `limit`, `id_col`, and `exclude_self` are shown above.) ### Step 6: Fusing Lexical and Dense Blocking Semantic and lexical blocking catch different matches, so fusing them usually beats either alone. `reciprocal_rank_fusion` combines any number of candidate tables by weighted Reciprocal Rank Fusion. You wrap each blocker's output in a `CandidateSource`, which takes three arguments: - **`name`** — a label for that blocker (for example `"lexical"` or `"semantic"`), carried into the fused output so you can see which source found each pair. - **`candidates`** — that blocker's candidate table. - **`weight`** — its relative influence in the fused score. The weights across your sources should sum to 1.0 (here two sources at 0.5 each, i.e. equal weight). First produce a lexical candidate set to fuse with the semantic one — here Sparkly BM25 over the same fields: ```python from madmatcher_pro.sparkly.index_config import IndexConfig from madmatcher_pro.sparkly.index.lucene_index import LuceneIndex from madmatcher_pro.sparkly.search import Searcher cfg = IndexConfig(id_col="_id") for col in ["name", "description"]: cfg.add_field(field=col, analyzers=["3gram"]) lidx = LuceneIndex(index_path="work/lex", config=cfg, delete_if_exists=True) lidx.upsert_docs(df=table_a) lexical_searcher = Searcher(index=lidx) lexical = lexical_searcher.search(search_df=table_b, query_spec=lidx.get_full_query_spec(), limit=20, id_col="_id") ``` The lexical half here is the Sparkly blocker — `IndexConfig`, `LuceneIndex`, `Searcher`, the analyzers, and `get_full_query_spec`. See the Sparkly tutorial for what those calls do; this guide uses them only to produce a second candidate set to fuse. Then wrap each candidate set in a `CandidateSource` and fuse them: ```python from madmatcher_pro.semantic import reciprocal_rank_fusion, CandidateSource lexical_source = CandidateSource(name="lexical", candidates=lexical.select("id2", "id1_list", "scores"), weight=0.5) semantic_source = CandidateSource(name="semantic", candidates=candidates, weight=0.5) fused = reciprocal_rank_fusion(sources=[lexical_source, semantic_source], limit=20) ``` On `abt_buy` at `limit=20`, the fused recall meets or beats both endpoints — for example lexical ≈ 0.93, semantic ≈ 0.95, fusion ≈ 0.96 (exact numbers depend on the model and knobs). `fused` has the same contract as either input, so it feeds MatchFlow featurization directly. ### Step 7: Saving Blocking Output The candidate DataFrame is a normal Spark DataFrame. Write it wherever your matcher reads from: ```python candidates.write.mode("overwrite").parquet("output/semantic_candidates") ``` ## Closing Notes You embedded both tables once, built a dense index over the indexing table, searched it for each search record, and (optionally) fused the result with a lexical blocker. The output is the standard candidate contract, ready for MatchFlow. From here you can raise `nprobe` for more recall or fuse more sources. When your data grows large, the **Appendix** covers shrinking the vectors and searching in bounded memory. --- ## Reference: Crash Recovery and Progress Tracking This is the complete reference for the crash recovery used in the steps above. It explains how the pieces fit together, what happens when a job crashes, and exactly what each argument does and does not do. ### How it works When `enable_crash_recovery` is off (the default), each operation runs exactly as described earlier, and none of the arguments here have any effect. When it is on, the operation runs in groups: 1. **Split.** The input is divided into `n_groups` groups using a deterministic hash of each row's id, so a given row always lands in the same group. 2. **Process and write.** Each group is processed on its own and its output is written under `checkpoint_dir`. 3. **Commit.** After a group's output is safely written, a small commit marker is written for it. The marker, not the data, is what records the group as done. 4. **Resume.** Running the same call again with the same `checkpoint_dir` reads the markers, skips committed groups, processes only the rest, and returns the union of all groups. A first run processes every group; a resume processes only the uncommitted ones; a run where everything is already committed does no processing. The result is the same set of rows a single uninterrupted run would produce. Recovery changes how the work is scheduled and committed, not what the answer is. On the first run the operation also records identifying facts (the input row count, and a fingerprint of the run's configuration) in a `manifest.json` under `checkpoint_dir`, and re-checks them on every resume; a resume whose facts no longer match is refused, so results from two different inputs are never mixed. The index build has one extra guarantee worth restating: its internal model is **fitted once**, on the first run, and stored in the checkpoint. A resume reuses it rather than refitting, so every group is built consistently. ### What happens when a job crashes Nothing is left corrupted. A committed group is safe, because its marker is written only after its output is durable. A group still being written when the crash hit has no marker, so it is treated as not done. To recover, **run the same operation again with the same `checkpoint_dir`**. There is no separate resume command and no cleanup step. Re-running is **idempotent**: a committed group is never reprocessed, an interrupted group is redone and overwrites any partial output, and an operation that already finished does no work. Two things matter for a full pipeline: - Each operation is idempotent only against its **own** checkpoint. Give each long step you want to resume its **own** `checkpoint_dir`. A step run without crash recovery re-runs from the start. - The three semantic steps are separate operations. In particular, embed each table to a durable `out_path` and build the index and search from that stored vector table; recovery covers each operation's checkpointed work, not the writes you do between them. ### Crash-recovery arguments Not every operation takes every argument. All three take `enable_crash_recovery`, `checkpoint_dir`, `n_groups`, `rows_per_group`, and `validate_input`; `create_embeddings` additionally takes `show_progress`, and `SemanticSearcher.search` additionally takes `exclude_self`. Each argument is noted with the operations it applies to. **`enable_crash_recovery`** (boolean, default `False`; all three steps) Turns the resumable path on. When on, `checkpoint_dir` is required. *It does not:* change anything when off (the operation and its output are exactly as in the base steps above). It does not make a successful run faster; recovery adds a small, fixed cost (writing each group and its marker) in exchange for being able to resume. **`checkpoint_dir`** (string, default `None`; required when recovery is on; all three steps) The durable location for each group's output and commit markers. Reuse the same path to resume. It accepts a local path or a remote URL (`file://`, `hdfs://`, `s3a://`, and so on). *It does not:* work as worker-local scratch space; it must be reachable by every node and survive the crash. It is not cleaned up for you; delete it once the job is done. It holds one operation's checkpoint only — pointing it at a different operation or a changed config is refused, not overwritten. **Splitting the work: set `n_groups` or `rows_per_group`, not both.** These two arguments both control how the input is divided into resumable groups, and you set at most one of them. **`rows_per_group` is usually the better choice** — it bounds how much work you redo after a crash regardless of how large the input is. Use **`n_groups`** for a simple fixed split, which suits small or fixed-size inputs. If you set both, `rows_per_group` wins (it overrides `n_groups`). All three steps accept both. **`n_groups`** (integer, default `10` for embed and search, `0` (auto) for the index build; all three steps) How many groups the work is split into, which sets the recovery and commit granularity. *It does not:* change the result, only how the work is chunked, and it is not the number of Spark tasks or partitions. It is fixed once a checkpoint exists: a resume reuses the recorded value, so to change it you start a fresh `checkpoint_dir`. **`rows_per_group`** (integer, default `None`; all three steps) An alternative to `n_groups` that sizes groups by work instead of by count: `n_groups = ceil(row_count / rows_per_group)`. This bounds how much you redo after a crash regardless of input size. *It does not:* produce exactly `rows_per_group` rows per group (the hash spreads rows roughly evenly). If both are given, `rows_per_group` wins. It costs one count of the input on the first run. **`validate_input`** (boolean, default `True`; all three steps) Records the input row count on the first run and re-checks it on every resume, refusing (with `CheckpointMismatchError`) if it changed. This guards against a resume silently dropping rows. *It does not:* compare the input's contents; it is a row-count check, so a change that keeps the same count is not detected. The configuration fingerprint is always checked regardless, so a changed config is refused even with `validate_input=False`. **`show_progress`** (boolean, default `False`; `create_embeddings` only) Prints a per-group progress bar to the console as groups commit. *It does not:* affect results or recovery, and it is separate from the web dashboard. ### Viewing progress on the dashboard Progress tracking shows each operation live on a small web dashboard. To turn it on, create a `MadMatcherSession` near the top of your script. Creating the session launches the dashboard once and tracks every operation that follows, so you do not pass a flag to each call. ```python from madmatcher_pro import MadMatcherSession with MadMatcherSession.builder.open_browser(True).getOrCreate() as session: print(f"Dashboard: {session.url}") # run the embedding, index build, and search below here; each appears as a job # leaving the block shuts the dashboard down ``` The session builder takes `.dashboard()` (tracking on or off, default on), `.port()` (default `4050`), `.open_browser()` (default `False`), and `.host()` (default `127.0.0.1`, loopback); `.getOrCreate()` returns the already-active session if one exists, or launches and activates a new one. `session.url` is the dashboard address once it is up. This works the same whether you run your script with `python your_script.py` or submit it with `spark-submit your_script.py`. The examples in this guide use `python ...` for brevity; `spark-submit ...` is equivalent. The dashboard runs inside the Spark driver, so where you reach it depends on where the driver runs: - **Driver on your machine** (plain `python`, or `spark-submit --deploy-mode client`): the dashboard is at `http://localhost:4050` on your machine. - **Driver on another host** (a remote gateway you launched from, or `spark-submit --deploy-mode cluster`): the driver, and so the dashboard, runs somewhere other than your laptop. If you can SSH to that host, the simplest and most secure way to reach the dashboard is to port-forward it — `ssh -L 4050:localhost:4050 user@driver-host`, then open `http://localhost:4050` on your laptop; this works with the default bind and exposes nothing on the network. If you cannot tunnel (for example the driver landed on an arbitrary cluster node), instead bind it with `MadMatcherSession.builder.host("0.0.0.0")` and reach it over the network — the dashboard has no built-in authentication, so secure it at the network layer. Either way, the read-only viewer below works from any machine that can read the checkpoint directory. On the dashboard, each operation (`create_embeddings`, `upsert_docs`, `search`) appears as a job with a live progress bar, throughput, and ETA. When you resume a job, the bar opens at the work already committed, so the ETA reflects only the remaining work. If a job fails, the page shows the error and reminds you to re-run with the same `checkpoint_dir`. To watch or inspect a job from any machine, regardless of how it was launched, point the read-only viewer at a checkpoint directory: ```bash python -m madmatcher_pro.reliability.dashboard ckpt/blocking ``` It reads the checkpoint and serves the same dashboard, read-only. Its arguments are: - **The checkpoint directory** (one or more) is the same `checkpoint_dir` you passed to the operation. Pass several to watch several jobs at once. Start the operation first, then launch the viewer: it resolves each checkpoint directory once, at startup, so a directory with no run in it yet (no `manifest.json`) is skipped and shows nothing until you restart the viewer. - **`--port`** is the port to serve on (default 4050; it scans upward if that port is busy). - **`--poll`** is how often, in seconds, the viewer re-reads the checkpoint directory for new progress (default 2.0). - **`--no-browser`** keeps it from opening a browser when it starts (it opens one by default). The dashboard is best-effort: if it cannot start, for example because the port is busy or the host is headless, your job still runs. The three dashboard arguments below can be passed on any operation, or set on the session builder with `.dashboard()`, `.port()`, `.open_browser()`, and `.host()`; they resolve in the order explicit call argument, then active session, then default. **`dashboard`** (boolean, default unset) Turns tracking on or off for a single call. Unset means "use the active session, or on if there is none". *It does not:* affect results or recovery. The environment variable `MADMATCHER_DASHBOARD=0` (also `false`, `no`, `off`) turns tracking off everywhere and overrides this argument. **`dashboard_port`** (integer, default `4050`) The port the dashboard tries first; if it is busy, the server scans upward for a free one. *It does not:* guarantee that exact port (it may land on the next free one), and only the first job in a process chooses the port; later jobs share that server. **`open_browser`** (boolean, default `False`) Opens the dashboard in your browser when it launches. *It does not:* do anything on a headless host with no browser; the job and dashboard still run. --- ## Appendix ### Embedding providers Pass any of these as the `provider` to `create_embeddings` (or pass a model-id string, which resolves to one). Each produces a fixed-width vector per record. | Provider | What it is | |----------|------------| | `SentenceTransformerProvider(model_id)` | Runs a local sentence-transformers model on CPU or GPU. The most flexible option; `model_id` is any sentence-transformers model name or local path. | | `FastEmbedProvider(model_id="BAAI/bge-small-en-v1.5")` | A lighter local model with less setup than a full deep-learning stack. | | `OpenAIEmbeddingProvider(model_id, dimensions=None)` | Hosted embeddings through the OpenAI API — no local model to run. `dimensions` optionally shortens Matryoshka (`text-embedding-3-*`) vectors. | You can also pass a bare model-id string in place of a provider (for example `"BAAI/bge-small-en-v1.5"`); it is resolved to a built-in provider for you. **Choosing a model.** Start with a small, fast, general-purpose model such as `BAAI/bge-small-en-v1.5`. If recall on your data falls short, move to a larger model — larger models usually embed better, at the cost of more time and memory. Whatever you choose, embed table A and table B with the **same** model so their vectors are comparable. ### Bringing your own provider Any embedding backend works — a local model, a hosted API such as OpenAI or Google Gemini, or an internal model service. Subclass `EmbeddingProvider` and implement three members: - **`dim`** — a property returning the output vector width. - **`is_mrl`** — a property returning whether the model is Matryoshka. - **`embed(texts)`** — the method MadMatcher calls, once per partition. It receives the serialized record texts as a pandas `Series` and must return a `(len(texts), dim)` float32 array, one row per text, in the same order. `embed` is *your* code: inside it you talk to whatever model or service you use and reshape its output into that array. There is no single method every backend has — a local sentence-transformers model exposes `model.encode(...)`, the OpenAI client exposes `client.embeddings.create(...)`, and a REST service is a plain HTTP call. The two built-in providers below are the exact classes MadMatcher ships (lightly trimmed); copy whichever shape matches your backend and swap in your own model or SDK. A **local model** — load it once per executor and encode each partition: ```python import numpy as np import pandas as pd from madmatcher_pro.semantic import EmbeddingProvider, create_embeddings class MyLocalProvider(EmbeddingProvider): def __init__(self, model_id, *, batch_size=256): self.model_id = model_id self.batch_size = batch_size self._model = None # the loaded model; built lazily below def _model_obj(self): # Load on first use so the model is created once per executor, never on the # driver. `_model` is whatever object your library returns for the model. if self._model is None: from sentence_transformers import SentenceTransformer self._model = SentenceTransformer(self.model_id) return self._model def __getstate__(self): # A loaded model is not picklable; drop it so Spark ships only the config # (model_id, batch_size) and each executor loads its own copy. state = self.__dict__.copy() state["_model"] = None return state @property def dim(self) -> int: return int(self._model_obj().get_sentence_embedding_dimension()) @property def is_mrl(self) -> bool: return False # True only for Matryoshka models def embed(self, texts: pd.Series) -> np.ndarray: vectors = self._model_obj().encode( texts.tolist(), batch_size=self.batch_size, convert_to_numpy=True) return np.asarray(vectors, dtype=np.float32) # shape (len(texts), dim) ``` A **hosted API** — here `_client` is the vendor's SDK object, and the real request is `client.embeddings.create(...)`: ```python class MyHostedProvider(EmbeddingProvider): def __init__(self, model_id, *, batch_size=512): self.model_id = model_id self.batch_size = batch_size self._client = None # the API client; built lazily below def _client_obj(self): # `_client` is a handle to your API — here the OpenAI SDK, which reads its # key from the OPENAI_API_KEY environment variable. Built once per executor. if self._client is None: import openai self._client = openai.OpenAI() return self._client def __getstate__(self): state = self.__dict__.copy() state["_client"] = None # don't ship the client from the driver return state @property def dim(self) -> int: return 1536 # e.g. text-embedding-3-small's width @property def is_mrl(self) -> bool: return False def embed(self, texts: pd.Series) -> np.ndarray: client = self._client_obj() items = texts.tolist() vectors = [] for start in range(0, len(items), self.batch_size): # batch to stay under limits resp = client.embeddings.create( model=self.model_id, input=items[start:start + self.batch_size]) vectors.extend(d.embedding for d in resp.data) # the API preserves input order return np.asarray(vectors, dtype=np.float32) vec_a = create_embeddings(df=table_a, fields=fields, provider=MyHostedProvider(model_id="text-embedding-3-small"), out_path="work/emb_a") ``` Two patterns recur in both: the model or client is built **lazily** (in `_model_obj`/`_client_obj`, not `__init__`) so it is created on each executor rather than shipped from the driver, and `__getstate__` drops it before pickling for the same reason. `dim` and `is_mrl` describe your model; `embed` does the work. ### Serializers The `serializer` argument to `create_embeddings` controls how a record's fields become the one text that is embedded (see Step 3). | Serializer | Output | |------------|--------| | `DittoSerializer` (default) | `[COL] [VAL] ` per field, in order, with empty fields dropped and the text lowercased. The markers help the model tell fields apart. | | `PlainConcatSerializer` | The field values joined by spaces, with no markers. | ### Scaling to large data Two tools help when the tables are large. **Shrink Matryoshka embeddings.** A Matryoshka (MRL) model's vectors can be truncated to a shorter prefix and re-normalized with almost no recall loss, which shrinks the index and speeds up search. `shrink_mrl` does this on a vector column: ```python from madmatcher_pro.semantic import shrink_mrl vec_a_256 = shrink_mrl(df=vec_a, target_dim=256, embedding_col="embedding") ``` Or build the index with `target_dim=` to shrink on intake and keep the query side aligned automatically: ```python index = SemanticIndex(index_path="work/index", id_col="_id", target_dim=256) index = index.upsert_docs(df=vec_a) ``` **When the index does not fit in memory.** `run_chunked_search` builds and searches in bounded memory: it partitions the corpus into chunks, builds and searches each chunk within the memory budget, and folds the per-chunk results into a global top-`limit`. Give it an `available_bytes` memory budget, an `index_root`, and a `checkpoint_dir`; it is resumable and result-preserving (the same answer as a single-index search when its per-chunk keep count is at least `limit`). --- For the full API, see the reference at `https://docs.madmatcher.ai/madmatcher-pro/{version}/madmatcher_pro.semantic.html` (`{version}` is your installed release). For help, email support@madmatcher.ai or visit [madmatcher.ai](https://madmatcher.ai).