Scoring a provided blocked_id_pairs table

[medwar99]

I'm wondering if there is existing functionality within Splink to score a provided table of edges?

e.g. if I provide a table like __splink__blocked_id_pairs:

match_key	join_key_l	join_key_r
0	set_A_1	set_B_2
...

a linker could then expand that to __splink__df_comparison_vectors, and score them to produce a df_predict table only for those ID pairs?

This is useful to me where the records exist within the linker object's input_tables already, and I have a target list of edges I'd like to generate a df_predict for. I'd like to avoid pulling the data out to records and then using compare_two_records(), I'd just like to score this table of candidate edges.

I've bastardised the inference.predict() function to kind of do what I want by not computing the blocked_id_pairs and instead taking it in as an argument, but:

1. I'm not sure if this already exists in other functionality,
2. Not sure if it's a good approach (I might be abusing the CTEPipeline?),
3. I'm forcing the pipeline to not use the cache, as I was seeing it re-use previous df_predict even when I change the input ID pairs table,
4. The match_key column isn't actually needed for this. Likewise I believe the intent is to move away from the join_key and towards USING(unique_id, source_dataset)? In either case I kept them just to get the function running.

The most important parts are:

• not computing blocked_id_pairs, taking it in as an argument
• having the final output table not named df_predict, but still retaining it's structure

Any insights would be welcome :)

Related: #2822

---

# splink.internals.linker_components.inference
    def score_edges(
            self,
            blocked_id_pairs,
            threshold_match_probability: float = None,
            threshold_match_weight: float = None,
            materialise_after_computing_term_frequencies: bool = True
        ) -> SplinkDataFrame:
        """Create a dataframe of scored pairwise comparisons using the parameters
        of the linkage model.

        This takes in a table mimicking __splink__blocked_id_pairs, and only scores these pairs.

        Examples:
            ```py
            linker = linker(df, "saved_settings.json", db_api=db_api)
            df = pd.DataFrame({'match_key':[0], 'join_key_l': ['set_A_1'], 'join_key_r': ['set_B_2']})

            my_id_pairs = linker.table_management.register_table(df, "blocked_id_pairs")
            splink_df = linker.inference.score_edges(my_id_pairs)
            splink_df.as_pandas_dataframe(limit=5)
            ```
        Returns:
            SplinkDataFrame: A SplinkDataFrame of the scored pairwise comparisons.
        """
        
        pipeline = CTEPipeline([blocked_id_pairs])
        pipeline.enqueue_sql("SELECT * FROM blocked_id_pairs", "__splink__blocked_id_pairs")
        blocked_pairs = self._linker._db_api.sql_pipeline_to_splink_dataframe(pipeline)
        
        pipeline = CTEPipeline()
        
        if (
            materialise_after_computing_term_frequencies
            or self._linker._sql_dialect.sql_dialect_str == "duckdb"
        ):
            df_concat_with_tf = compute_df_concat_with_tf(self._linker, pipeline)
            pipeline = CTEPipeline([df_concat_with_tf])
        else:
            pipeline = enqueue_df_concat_with_tf(self._linker, pipeline)

        if (
            len(self._linker._input_tables_dict) == 2
            and self._linker._settings_obj._link_type == "link_only"
        ):
            sqls = split_df_concat_with_tf_into_two_tables_sqls(
                "__splink__df_concat_with_tf",
                self._linker._settings_obj.column_info_settings.source_dataset_column_name,
            )
            pipeline.enqueue_list_of_sqls(sqls)

        pipeline = CTEPipeline([blocked_pairs, df_concat_with_tf])
        
        sqls = compute_comparison_vector_values_from_id_pairs_sqls(
            self._linker._settings_obj._columns_to_select_for_blocking,
            self._linker._settings_obj._columns_to_select_for_comparison_vector_values,
            input_tablename_l="__splink__df_concat_with_tf",
            input_tablename_r="__splink__df_concat_with_tf",
            source_dataset_input_column=self._linker._settings_obj.column_info_settings.source_dataset_input_column,
            unique_id_input_column=self._linker._settings_obj.column_info_settings.unique_id_input_column
        )
        pipeline.enqueue_list_of_sqls(sqls)


        sqls = predict_from_comparison_vectors_sqls_using_settings(
            self._linker._settings_obj,
            threshold_match_probability,
            threshold_match_weight,
            sql_infinity_expression=self._linker._infinity_expression,
        )
        pipeline.enqueue_list_of_sqls(sqls)

        pipeline.enqueue_sql("SELECT * FROM __splink__df_predict", "scored_edges")

        scored_edges = self._linker._db_api.sql_pipeline_to_splink_dataframe(pipeline, use_cache=False)

        self._linker._predict_warning()

        blocked_pairs.drop_table_from_database_and_remove_from_cache()

        return scored_edges

[RobinL]

Yeah - I don't think there's an easy/clean way to do this in Splink 4.

In Splink 5 (forthcoming), we deliberately make each part of the workflow (computing tf tables, computing df_concat_with_tf, blocking, predict) more like a lego block, so this kind of thing is explicitly supported - see here.

[medwar99]

No problem, the above currently works for my use case in a rough-and-ready manner.

I like the lego approach, we're currently wrapping some Splink 4 pipelines with Airflow and this more modular approach should play even nicer with the DAGs there. Will give the dev branch a nose and eagerly await Splink 5 release! 😄