title: Fast comparison between pyspark dataframe and a list of elements?
body: <p>Good Afternoon everyone! I have an issue in my project. I need to read each element of a pyspark dataframe and use each of these together another each element in a list..</p>
  <p>This takes a lot of time 'cause the dataframe could be composed by 100 to 180k rows and the lists could be composed by 100-400 elements.. How can I speed the process? I tried to create something to speed up the process by the time is horrible:</p>
  <pre><code># this is a function that takes a weighted score between two of the famouse distances for text similarity
  def triplet_weighted_avg_score(word1, word2):
      levenshstein = fuzz.ratio(word1, word2)/100 # levenshstein distance
      s = SequenceMatcher(None); s.set_seq1(word1); s.set_seq2(word2); edit = s.ratio() # edit distance
      #jaro_ = jaro.jaro_winkler_metric(word1, word2)
      
      return 0.6*levenshstein+0.40*edit #+0.20*jaro_
  
  # this function iterates through a dataset of comuni and check which element of the list of unigrams, bigrams and trigrams over a text is more similar, later I am appending the value
  def find_comune(df, unigrams, bigrams, trigrams):
      edited_text = []
      
      for row in df.collect():
          for unigram in unigrams:
              score = triplet_weighted_avg_score(row['Comune'].lower(), unigram)
              if score &gt; 0.90:
                  edited_text.append([row['Comune'], unigram, score])
          
          for bigram in bigrams:
              score = triplet_weighted_avg_score(row['Comune'].lower(), bigram)
              if score &gt; 0.90:
                  edited_text.append([row['Comune'], bigram, score])
                  
          for trigram in trigrams:
              score = triplet_weighted_avg_score(row['Comune'].lower(), trigram)
              if score &gt; 0.90:
                  edited_text.append([row['Comune'], unigram, score])
      
      return sorted(edited_text, key = lambda x: x[2])
  
  # this function iterates through a dataset of surnames and check which element of the list of unigrams and bigrams over a text is more similar, later I am appending the value
  def find_cognome(df, unigrams, bigrams):
      edited_text = []
      
      for row in df.collect():
          for unigram in unigrams:
              score = triplet_weighted_avg_score(row['Cognome'].lower(), unigram)
              if score &gt; 0.90:
                  edited_text.append([row['Cognome'], unigram, score])
          
          for bigram in bigrams:
              score = triplet_weighted_avg_score(row['Cognome'].lower(), bigram)
              if score &gt; 0.90:
                  edited_text.append([row['Cognome'], bigram, score])
      
      return sorted(edited_text, key = lambda x: x[2])
  
  # this function iterates through a dataset of names and check which element of the list of unigrams and bigrams over a text is more similar, later I am appending the value
  def find_nome(df, unigrams, bigrams):
      edited_text = []
      
      for row in df.collect():
          for unigram in unigrams:
              score = triplet_weighted_avg_score(row['Nome'].lower(), unigram)
              if score &gt; 0.90:
                  edited_text.append([row['Nome'], row['Sesso'], unigram, score])
          
          for bigram in bigrams:
              score = triplet_weighted_avg_score(row['Nome'].lower(), bigram)
              if score &gt; 0.90:
                  edited_text.append([row['Nome'], row['Sesso'], bigram, score])
      
      return sorted(edited_text, key = lambda x: x[2])
  </code></pre>
  <p>above there are the functions used, below the main code:</p>
  <pre><code>for column_empty in [&quot;COMUNE_DI_RESIDENZA&quot;, &quot;COGNOME_E_NOME&quot;]:
      
      if df_spark_transl.first()[column_empty] == &quot;NULL&quot;:
          if column_empty == &quot;COMUNE_DI_RESIDENZA&quot;:
              comuni = find_comune(df_comuni, unigrams, bigrams, trigrams)
          
          if column_empty == &quot;COGNOME_E_NOME&quot;:
              cognomi = find_cognome(df_cognomi, unigrams, bigrams)
              nomi = find_nome(df_nomi_e_sesso, unigrams, bigrams)
  </code></pre>