title: How to optimize searching RDD contents in another RDD
body: <p>I have a problem where I need to search the contents of an RDD in another RDD.</p>
  
  <p>This question is different from <a href="https://stackoverflow.com/questions/43938672/efficient-string-matching-in-apache-spark">Efficient string matching in Apache Spark</a>, as I am searching for an exact match and I don't need the overhead of using the ML stack.</p>
  
  <p>I am new to spark and I want to know which of these methods is more efficient or if there is another way.
  I have a keyword file like the below sample (in production it might reach up to 200 lines)</p>
  
  <p><strong>Sample keywords file</strong></p>
  
  <pre><code>0.47uF 25V X7R 10% -TDK C2012X7R1E474K125AA
  20pF-50V NPO/COG - AVX- 08055A200JAT2A
  </code></pre>
  
  <p>and I have another file(tab separated)from which I need to find matches(in production I have up to 80 <strong>Million</strong> line) </p>
  
  <pre><code>C2012X7R1E474K125AA Conn M12 Circular PIN 5 POS Screw ST Cable Mount 5 Terminal 1 Port
  </code></pre>
  
  <p><strong>First method</strong>
  I defined a <code>UDF</code> and looped through keywords for each line</p>
  
  <pre><code>keywords = sc.textFile("keys")
  part_description = sc.textFile("part_description")
  
  
  def build_regex(keywords):
      res = '('
      for key in keywords:
          res += '(?&lt;!\\\s)%s(?!\\\s)|' % re.escape(key)
      res = res[0:len(res) - 1] + ')'
      return r'%s' % res
  
  
  def get_matching_string(line, regex):
      matches = re.findall(regex, line, re.IGNORECASE)
      matches = list(set(matches))
      return list(set(matches)) if matches else None
  
  
  def find_matching_regex(line):
      result = list()
      for keyGroup in keys:
          matches = get_matching_string(line, keyGroup)
          if matches:
              result.append(str(keyGroup) + '~~' + str(matches) + '~~' + str(len(matches)))
      if len(result) &gt; 0:
          return result
  
  
  def split_row(list):
      try:
          return Row(list[0], list[1])
      except:
          return None
  
  
  keys_rdd = keywords.map(lambda keywords: build_regex(keywords.replace(',', ' ').replace('-', ' ').split(' ')))
  keys = keys_rdd.collect()
  
  sc.broadcast(keys)
  
  part_description = part_description.map(lambda item: item.split('\t'))
  df = part_description.map(lambda list: split_row(list)).filter(lambda x: x).toDF(
      ["part_number", "description"])
  
  find_regex = udf(lambda line: find_matching_regex(line), ArrayType(StringType()))
  
  df = df.withColumn('matched', find_regex(df['part_number']))
  
  df = df.filter(df.matched.isNotNull())
  
  df.write.save(path=job_id, format='csv', mode='append', sep='\t')
  </code></pre>
  
  <p><strong>Second method</strong>
  I thought I can do more parallel processing (instead of looping through keys like above) I did cartersian product between keys and lines, splitted and exploded the keys then compared each key to the part column</p>
  
  <pre><code>df = part_description.cartesian(keywords)
  
      df = df.map(lambda tuple: (tuple[0].split('\t'), tuple[1])).map(
          lambda tuple: (tuple[0][0], tuple[0][1], tuple[1]))
  
      df = df.toDF(['part_number', 'description', 'keywords'])
  
      df = df.withColumn('single_keyword', explode(split(F.col("keywords"), "\s+"))).where('keywords != ""')
  
      df = df.withColumn('matched_part_number', (df['part_number'] == df['single_keyword']))
  
      df = df.filter(df['matched_part_number'] == F.lit(True))
  
      df.write.save(path='part_number_search', format='csv', mode='append', sep='\t')
  </code></pre>
  
  <p>Are these the correct ways to do this? Is there anything I can do to process these data faster?</p>