title: Matching Company names in Python through Cosine Similarity, TF-IDF and pyspark
body: <p>I'm trying to match company names from 2 lists together, in order to check if a company in list A is indeed listed in list B. As company names a written in all kinds of different forms, I leaned toward a match using the cosing similarity. 
  For this, I followed the note on this blog made by Ran Tavory : <a href="https://medium.com/@rantav/large-scale-matrix-multiplication-with-pyspark-or-how-to-match-two-large-datasets-of-company-1be4b1b2871e" rel="nofollow noreferrer">Link Here</a></p>
  
  <p>Here is the general outline:</p>
  
  <blockquote>
    <ol>
    <li>Calculate TF-IDF matrices on the driver.</li>
    <li>Parallelize matrix A; Broadcast matrix B</li>
    <li>Each worker now flatMaps its chunk of work by multiplying its chunk of matrix A with the entire matrix B. So if a worker operates on
    A[0:99] then it would multiply these hundred rows and return the
    result of, say A[13] matches a name found in B[21]. Multiplication is
    done using numpy.</li>
    <li>The driver would collect back all the results from the different workers and match the indices (A[13] and B[21]) to the actual names in
    the original dataset — and we’re done!</li>
    </ol>
  </blockquote>
  
  <p>I was able to run the exact code described in the note, but one part of it seems kind of odd : 
  <code>b_mat_dist = broadcast_matrix(a_mat)</code></p>
  
  <p>When broadcasting a_mat as well as parallelize a_mat, I get a logical result of a perfect match for every company names (as we're looking in the same source).</p>
  
  <p>When I try broadcasting the b_mat : b_mat_dist = broadcast_matrix(b_mat), I get the following error : <em><code>Incompatible dimension for X and Y matrices: X.shape[1] == 56710 while Y.shape[1] == 2418</code></em></p>
  
  <p>Any help would be greatly appreciated!
  Thanks in advance !</p>
  
  <p>Here is my code :</p>
  
  <pre><code>import pandas as pd
  from sklearn.feature_extraction.text import TfidfVectorizer
  from sklearn.metrics.pairwise import cosine_similarity
  from pyspark.sql import SQLContext, SparkSession
  from pyspark import SparkContext
  from scipy.sparse import csr_matrix
  vectorizer = TfidfVectorizer()
  
  if 'sc' in locals():
      sc.stop()
  
  sc = SparkContext("local", "Simple App")
  
  pd.set_option('display.max_colwidth', -1)
  RefB =  pd.read_excel('Ref.xlsx')
  ToMatchB =  pd.read_excel('ToMatch.xlsx')
  
  Ref = RefB['CLT_company_name']
  ToMatch = ToMatchB ['Name1']
  
  a_mat = vectorizer.fit_transform(Ref)
  b_mat = vectorizer.fit_transform(ToMatch)
  
  def find_matches_in_submatrix(sources, targets, inputs_start_index,
                                threshold=.8):
      cosimilarities = cosine_similarity(sources, targets)
      for i, cosimilarity in enumerate(cosimilarities):
          cosimilarity = cosimilarity.flatten()
          # Find the best match by using argsort()[-1]
          target_index = cosimilarity.argsort()[-1]
          source_index = inputs_start_index + i
          similarity = cosimilarity[target_index]
          if cosimilarity[target_index] &gt;= threshold:
              yield (source_index, target_index, similarity)
  
  def broadcast_matrix(mat):
      bcast = sc.broadcast((mat.data, mat.indices, mat.indptr))
      (data, indices, indptr) = bcast.value
      bcast_mat = csr_matrix((data, indices, indptr), shape=mat.shape)
      return bcast_mat
  
  def parallelize_matrix(scipy_mat, rows_per_chunk=100):
      [rows, cols] = scipy_mat.shape
      i = 0
      submatrices = []
      while i &lt; rows:
          current_chunk_size = min(rows_per_chunk, rows - i)
          submat = scipy_mat[i:i + current_chunk_size]
          submatrices.append((i, (submat.data, submat.indices, 
                                  submat.indptr),
                              (current_chunk_size, cols)))
          i += current_chunk_size
      return sc.parallelize(submatrices)
  
  a_mat_para = parallelize_matrix(a_mat, rows_per_chunk=100)
  b_mat_dist = broadcast_matrix(b_mat)
  results = a_mat_para.flatMap(
          lambda submatrix:
          find_matches_in_submatrix(csr_matrix(submatrix[1],
                                               shape=submatrix[2]),
                                     b_mat_dist,
                                     submatrix[0]))
  </code></pre>