estimate_u_using_random_sampling spark seed

[Brian Head (brianfhead)]

Hi,

I am converting my code from using DuckDB to Spark. I've been able to get everything to work except pretty much the same except the 'estimate_u_using_random_sampling' function. When using DuckDB I use the seed like:

linker.training.estimate_u_using_random_sampling(max_pairs=1e9, seed=1911)

But, when I convert to Spark this results in the following error:

Py4JJavaError: An error occurred while calling o3892.saveAsTable. : java.util.NoSuchElementException: key not found: 587

I've tried it several times and the "key" number changes each time. LLMs were not helpful in identifying the problem. I have also not been successful in finding another discussion on this topic.

However, I've figured out I can remove the seed and it will run, like:

linker.training.estimate_u_using_random_sampling(max_pairs=1e9)

Is it not possible or does it require different something different to set a seed for reproducibility with spark?

I am using splink version 4.0.1.

Appreciate any help.

Thank you,
Brian

[Robin Linacre (RobinL)]

I've seen this before when people are doing several spark transforms such as monotonically increasing id prior to bringing the data into Splink. The solution has been to save the data out to e.g. parquet before importing to Splink i.e the data frame coming into splijk should be loaded directly from disk.

(I'm not sure this will fix your problem but it's a good place to start)

[Brian Head (brianfhead)]

Thank you for the quick response, Robin. I do have a lot of data transformations being done in spark so I was hopeful. However, I've written the data to a table on disk and then re-read it back into spark and get the same error. Any other ideas/suggestions?

[Brian Head (brianfhead)]

Thank you, Robin. I had tried this but get the error:

TypeError: Linker.__init__() got an unexpected keyword argument 'break_lineage_method

I didn't previously mention because I didn't want to conflate issues. I did add some imports to my script from the page you linked to above hoping it might be related to not having imported something. I also get errors for the last two imports:

from .dataframe import SparkDataFrame from splink.internals.jar_location import get_scala_udfs

I'm a bit confused on the second one in particular because functions I thought were in the UDFs (e.g., DamerauLevenshteinAtThresholds) are working later in the script.

Any suggestions on these issues? I looked through the change log but didn't see anything in the newer versions that seemed to address them (I'm using 4.0.1).

[Robin Linacre (RobinL)]

The break lineage argument is on the database API, not the linker. On the other, I think we set up the udfs for users automatically in databricks, so if they're working you prob don't need to worry

[Brian Head (brianfhead)]

I was putting the break_lineage_method in the Linker argument like here: https://moj-analytical-services.github.io/splink/topic_guides/performance/optimising_spark.html?h=break_lineage_method#summary

Like this:
linker = Linker( [df_1, df_2] ,settings # ,db_api=DuckDBAPI() ,db_api=SparkAPI(spark_session=spark) ,break_lineage_method="parquet" ,input_table_aliases=["df_left", "df_right"] )
Are you saying I should put it somewhere else? Apologies if I've missed it somewhere else.

[Robin Linacre (RobinL)]

Ah - sorry those docs are out of date. It's moved to the database API i.e where you do
db_api = SparkAPI(spark_session=spark, break_lineage_method="parquet")

[Brian Head (brianfhead)]

Thank you, Robin. This seems to have resolved my issue.

[Robin Linacre (RobinL)]

Great. Thanks for confirming. I've also now updated the docs above to Splink 4 syntax - thanks for noticing that.