Sparkly (lexical blocking)

TF/IDF (Lucene) blocking. madmatcher_pro.sparkly has an empty __init__; import these symbols from their submodules, as shown below. The :members: on each class is scoped to its user-facing methods (the classes also carry internal plumbing that is not part of the supported API).

Index

class madmatcher_pro.sparkly.index.lucene_index.LuceneIndex(index_path, config, delete_if_exists=True, index_build_chunk_size=2500)

Bases: Index

A PyLucene-backed inverted index for TF/IDF (BM25) blocking.

Build (or extend) the index from a Spark or pandas DataFrame with upsert_docs() (optionally crash-recoverable), then bulk-search it with Searcher. The field/analyzer setup and id column come from the IndexConfig passed at construction; ANALYZERS lists the built-in analyzer names an IndexConfig field can reference. Requires PyLucene (a JVM build).

Parameters:

  • index_path (Path or str) – Directory the index is written to / read from.

  • config (IndexConfig) – Field/analyzer/id configuration; frozen on build.

  • delete_if_exists (bool, default True) – Clear any existing index at index_path first.

  • index_build_chunk_size (int, default 2500) – Records per build chunk/partition; also sets the Spark partition count (df_size // this) and the local-vs-distributed dispatch pivot.

upsert_docs(df, disable_distributed=False, force_distributed=False, show_progress_bar=False, *, dashboard=None, dashboard_port=None, open_browser=None, enable_crash_recovery=False, checkpoint_dir=None, n_groups=10, rows_per_group=None, validate_input=True)

Build the index, or incrementally add documents to an already-built one, indexing df according to the index config (self.config). A pandas or Spark DataFrame is accepted; a sufficiently large Spark DataFrame is built in parallel.

Parameters:

  • df (pd.DataFrame or pyspark DataFrame) – the table that will be indexed, if a pyspark DataFrame is provided, the build will be done in parallel for suffciently large tables

  • disable_distributed (bool, default=False) – disable using spark for building the index even for large tables

  • force_distributed (bool, default=False) – force using spark for building the index even for smaller tables

  • show_progress_bar (bool, default=False) – show the progress bar in addition to debug logs

  • dashboard (bool, optional) – for a parallel/distributed build, launch (or join) the best-effort web dashboard and track documents-indexed progress on it. None (default) defers to the active MadMatcherSession, else the env/default (on); pass True/False to override, and the global opt-out MADMATCHER_DASHBOARD=0 turns it off. Shares one server with any blocking jobs in the process; a launch failure is swallowed and never blocks the build. Small single-threaded builds are tracked too (a single Building index phase, no merge).

  • dashboard_port (int, optional) – port for the shared dashboard server (scans upward if busy); only the first job in the process decides the port. None defers to the active session, else 4050.

  • open_browser (bool, optional) – open the dashboard in a browser when the build starts (default False).

  • enable_crash_recovery (bool, default=False) – make the build resumable at segment granularity (checkpoint_dir then required). When False the arguments below are ignored and the normal all-or-nothing build runs. Covers the initial build only: requires a Spark DataFrame and a not-yet-built index (an upsert into a built index is incremental and runs the normal path).

  • checkpoint_dir (str, optional) – where the staged segments and commit markers are stored. Re-running with the same checkpoint_dir skips committed segments. May be local/hdfs/s3a.

  • n_groups (int, default=10) – split the input into this many hash-groups, each a separately checkpointed segment.

  • rows_per_group (int, optional) – if set, derive n_groups as ceil(num_records / rows_per_group).

  • validate_input (bool, default=True) – record the input row count and refuse to resume if it changed (a resumable build assumes a stable input).

get_full_query_spec(cross_fields=False)

get a query spec that uses all indexed columns

Parameters:

cross_fields (bool, default = False) – if True return <FIELD> -> <CONCAT FIELD> in the query spec if FIELD is used to create CONCAT_FIELD else just return <FIELD> -> <FIELD> and <CONCAT_FIELD> -> <CONCAT_FIELD> pairs

Return type:

QuerySpec

class madmatcher_pro.sparkly.index_config.IndexConfig(*, store_vectors=False, id_col='_id', weighted_queries=False)

Bases: object

Configuration for a sparkly (Lucene) index.

Declares which fields are indexed and with which analyzer(s), the id column, the BM25 similarity, and a few build options. Add analyzed fields with add_field / concatenated fields with add_concat_field, then pass the config to LuceneIndex. Call freeze() for an immutable copy (the index freezes its config on build).

Parameters:

  • store_vectors (bool, default False) – Store term vectors in the index (needed by some query specs).

  • id_col (str, default '_id') – The record id column; must be an integer/long, unique per row.

  • weighted_queries (bool, default False) – Generate per-field weighted queries instead of a flat disjunction.

add_field(field, analyzers)

Add a new field to be indexed with this config

Parameters:

  • field (str) – The name of the field (column) in the table to index

  • analyzers (set, list or tuple of str) – The analyzers used to index the field. Each must be a built-in analyzer name (see LuceneIndex.ANALYZERS); an unknown name raises ValueError when the index is built.

remove_field(field)

remove a field from the config

Parameters:

field (str) – the field to be removed from the config

Returns:

True if the field existed else False

Return type:

bool

add_concat_field(field, concat_fields, analyzers)

Add a new concat field to be indexed with this config

Parameters:

  • field (str) – The name of the new (derived) field to index – created by concatenating concat_fields, not an existing column.

  • concat_fields (set, list or tuple of str) – The columns concatenated (space-joined) to form field. Each must exist in the indexed table; and to search on this concat field, each must also exist in the search table, since the query value is rebuilt from them for each search record.

  • analyzers (set, list or tuple of str) – The analyzers used to index the field. Each must be a built-in analyzer name (see LuceneIndex.ANALYZERS); an unknown name raises ValueError when the index is built.

Query spec

class madmatcher_pro.sparkly.query_generator.QuerySpec(*args, **kwargs)

Bases: dict

A query specification: which indexed fields each search field is matched against, plus optional per-field boosts and filters.

A spec maps search_field -> {indexed_field, ...}: each key is a column read from the search table (B), and its values are the <field>.<analyzer> fields in the index (built on table A) that the column is matched against. search() runs it against every search record. You usually get one from get_full_query_spec() (covering all indexed fields), then optionally tune it:

  • boost_map – weight each (search_field, indexed_field) pair (default 1.0); a higher weight makes that field contribute more to the score.

  • filter – restrict scoring to specific (search_field, indexed_field) pairs.

property filter

The set of (search_field, indexed_field) pairs to filter on. Every pair set here must also be used for scoring in the spec (else RuntimeError); empty (the default) applies no filter.

property boost_map

The boosting weights for each (search_field, indexed_field) pair (a pair not in the map has weight 1.0; a higher weight makes that pair contribute more to the score). Create one by assigning a dict of (search_field, indexed_field) -> float – the keys are the same pairs the spec scores, the values must be floats:

spec = searcher.get_full_query_spec()
spec.boost_map = {
    ('name', 'name.standard'): 2.0,   # weight the standard match on name higher
    ('name', 'name.3gram'):    1.0,
}

Index optimization and config generation (Sparkly Auto)

class madmatcher_pro.sparkly.index_optimizer.index_optimizer.IndexOptimizer(is_dedupe, scorer=None, conf=0.99, init_top_k=10, max_combination_size=3, opt_query_limit=250, sample_size=10000, use_early_pruning=True, search_chunk_size=50)

Bases: object

Sparkly Auto: automatically choose the search fields and analyzers (the query spec) for an index, instead of specifying them by hand (“Sparkly Manual”).

Given a built index and a sample of the search table, optimize() scores candidate field/analyzer combinations and returns the best-performing QuerySpec; make_index_config() produces a starting IndexConfig to build that index from. Sparkly Auto requires every field of the indexed table (except the id column) to also exist in the search table.

Parameters:

  • is_dedupe (bool) – Should be true if the search table == the indexed table

  • scorer (QueryScorer, optional) – the class that will be used to score the queries during optimization. If not provided defaults to AUCQueryScorer

  • conf (float, in [0, 1), default=.99) – the confidence score cut off during optimization

  • init_top_k (int, > 0, default=10) – top-k candidates retrieved per query when scoring a candidate query spec

  • max_combination_size (int, > 0, default=3) – maximum number of (field, analyzer) pairs combined into a single query spec

  • opt_query_limit (int, > 0, default=250) – candidate limit used for the trial searches run during optimization

  • sample_size (int, > 0, default=10000) – number of search records sampled to score candidate query specs

  • use_early_pruning (bool, default=True) – prune clearly-underperforming candidates early instead of scoring every candidate on the full sample

  • search_chunk_size (int, > 0, default=50) – records per trial-search slice during optimization; raise it to trade driver memory for fewer trial rounds

make_index_config(df, id_col='_id', *, show_progress=False)

create the starting index config which can then be used to for optimization throws out any columns where the average number of whitespace delimited tokens are >= 50

Parameters:

  • df (pyspark.sql.DataFrame) – the dataframe that we want to generate a config for

  • id_col (str) – the unique id column for the records in the dataframe

  • show_progress (bool, default=False) – show a coarse two-step progress bar (token counting, then config assembly). Not crash-recoverable: this is a single Spark job with no intermediate state worth checkpointing; a crash just re-runs it.

Return type:

IndexConfig

optimize(index, search_df, *, enable_crash_recovery=False, checkpoint_dir=None, show_progress=False, validate_input=True, dashboard=None, dashboard_port=None, open_browser=None)
Parameters:

  • index (Index) – the index that will have an optimzed query spec created for it

  • search_df (pyspark.sql.DataFrame:) – the records that will be used to choose the query spec

  • enable_crash_recovery (bool, default=False) – make optimization resumable at candidate granularity (checkpoint_dir then required). When False the arguments below are ignored and the normal Wilcoxon-pruned optimization runs.

  • checkpoint_dir (str, optional) – where the persisted sample, per-candidate score cache and commit markers are stored. Re-running with the same checkpoint_dir skips already-scored candidates. May be local/hdfs/s3a.

  • show_progress (bool, default=False) – show a per-candidate progress bar.

  • validate_input (bool, default=True) – record the input row count and refuse to resume if it changed (a resumable run assumes a stable input).

  • dashboard (bool | None)

  • dashboard_port (int | None)

  • open_browser (bool | None)

Returns:

a query spec optimized for searching for search_df using index

Return type:

QuerySpec

Table validation

madmatcher_pro.sparkly.utils.check_tables_manual(table_a, id_col_table_a, table_b, id_col_table_b, embedding_col=None)

Check that table_a and table_b have valid id columns.

Parameters:

  • table_a (Union[pd.DataFrame, sql.DataFrame]) – The table A to be indexed.

  • id_col_table_a (str) – The column name of the id column in table A.

  • table_b (Union[pd.DataFrame, sql.DataFrame]) – The table B to be searched.

  • id_col_table_b (str) – The column name of the id column in table B.

  • embedding_col (str | None)

Raises:

ValueError – If table_a or table_b do not have a valid id column.

madmatcher_pro.sparkly.utils.check_tables_auto(table_a, id_col_table_a, table_b, id_col_table_b, embedding_col=None)

Check that table_a and table_b have valid id columns. Check that table_b columns are a supserset of table_a columns.

Parameters:

  • table_a (Union[pd.DataFrame, sql.DataFrame]) – The table A to be indexed.

  • id_col_table_a (str) – The column name of the id column in table A.

  • table_b (Union[pd.DataFrame, sql.DataFrame]) – The table B to be searched.

  • id_col_table_b (str) – The column name of the id column in table B.

  • embedding_col (str | None)

Raises:

  • ValueError – If table_a or table_b do not have a valid id column.

  • ValueError – If table_b columns are not a supserset of table_a columns.