The problem is as follow:

'genre', 'top_tags' (250 rows)
----------------
Action, Array('bleeding', 'dying', 'guns', ...) - can hold up to 50k max. (avg is 4000)
Drama,  Array('crying', 'hard life', 'street') 

And another table with movies, a genre and its associated tags

'movie', 'genre', 'tags'. (DataFrame size, around 23M Rows)
------------------------
M1        Action,  'guns', 'dying', 'bleeding', 'outside', 'worldwide'.  approx ~10 records for each movie

I want to iterate every movie and try to extend its genre similarity by comparing tags. NO Fuzzy algorithm, just exact match.

I want to return a DataFrame (the same movie, genre, tags dataframe) with new column called potentially_related_genres and a list of genres.

As I see it, I have two options:

1. crossJoin and compare two columns using UDF but this will be bad since it will lad me to iterate 5,750,000,000 rows. (crossJoin output)

2. pickle the results (do a collect() on the first dataframe (250 rows) and use right after within a UDF with all the logic, this by calling the original dataframe and use withColumn

DF.withColumn('potentially_related_genres', my_udf('genre', 'tags'))

The issue with the approach is that passing to the driver is a pretty big collect ( remember the fat column top_tags ). and transfer this to all of the workers to be used. (the pickling and unpickling)

Any suggestion ?

Thanks in advance.