title: Filter Spark DataFrame using keywords from another DataFrame
body: <p>I have a large dataset of news articles loaded into a PySpark DataFrame. I am interested in filtering that DataFrame down to the set of articles that contain certain words of interest in their body text. At the moment the list of keywords is small, but I would like to store them in a DataFrame anyway as that list may expand in the future. Consider the following small example:</p>
  
  <pre><code>from pyspark.sql import SparkSession
  spark = SparkSession.builder.getOrCreate()
  
  article_df = [{'source': 'a', 'body': 'Seattle is in Washington.'},
                  {'source': 'b', 'body': 'Los Angeles is in California'},
                  {'source': 'a', 'body': 'Banana is a fruit'}]
  article_data = spark.createDataFrame(article_data)
  
  keyword_data = [{'city': 'Seattle', 'state': 'Washington'},
                  {'city': 'Los Angeles', 'state': 'California'}]
  keyword_df = spark.createDataFrame(keyword_data)
  </code></pre>
  
  <p>This gives us the following DataFrames:</p>
  
  <pre><code>+--------------------+------+
  |                body|source|
  +--------------------+------+
  |Seattle is in Was...|     a|
  |Los Angeles is in...|     b|
  |   Banana is a fruit|     a|
  +--------------------+------+
  
  +-----------+----------+
  |       city|     state|
  +-----------+----------+
  |    Seattle|Washington|
  |Los Angeles|California|
  +-----------+----------+
  </code></pre>
  
  <p>As a first pass, I would like to filter down <code>article_df</code> so that it only contains articles whose <code>body</code> string contains any of the strings in <code>keyword_df['city']</code>. I'd also like to filter it down to articles that contain both a string from <code>keyword_df['city']</code> and the corresponding entry (same row) in <code>keyword_df['state']</code>. How can I accomplish this?</p>
  
  <p>I have managed to do this with a manually defined list of keywords:</p>
  
  <pre><code>from pyspark.sql.functions import udf
  from pyspark.sql.types import BooleanType
  def city_filter(x):
      cities = ['Seattle', 'Los Angeles']
      x = x.lower()
      return any(s.lower() in x for s in cities)
  filterUDF = udf(city_filter, BooleanType())
  </code></pre>
  
  <p>Then <code>article_df.filter(filterUDF(article_df.body)).show()</code> gives the desired result:</p>
  
  <pre><code>+--------------------+------+
  |                body|source|
  +--------------------+------+
  |Seattle is in Was...|     a|
  |Los Angeles is in...|     b|
  +--------------------+------+
  </code></pre>
  
  <p>How can I implement this filter without having to manually define the list of keywords (or tuples of keyword pairs)? Should I need to use a UDF for this?</p>