How can I filter an rows in column of ArrayType(StringType) against items in another column in a separate dataframe using pyspark?

I have two dataframes. The structure is as below:

df1:

where column B is an array (string)

df2:

where there is only a single value per row

I want to create a new column in df1, which is filtered using df2. If a value in the array is present in df1 then return True, otherwise False.

for example:

There is no maximum array length in df1 and there are about ~1000 rows of df2.

How can I create this boolean column using PySpark?

Thanks!

Most my error are such like column not iterable.

Use transform spark higher built functions.

• Get the distinct list of destinations from df2 and do cross join to get this list added to df1
• Then use Higher order functions array_contains() to look for each element in colb if it exists in destination array

Example:

from pyspark.sql.functions import *
df = spark.createDataFrame([('1',['Tokyo', 'Singapore']),('2',['Tokyo', 'New York', 'Paris']),('3',['Paris'])],['cola','colb'])

df1 = spark.createDataFrame([('Paris',),('New York',)], ['Destination'])

df1 = df1.groupBy(lit("1")).agg(collect_set(col("Destination")).alias("dst")).drop('1')

df2 = df.crossJoin(df1)

df2.withColumn("new col", expr("""transform(colb,x ->array_contains(dst,x))""")).show()
#+----+------------------------+-----------------+-------------------+
#|cola|colb                    |dst              |new col            |
#+----+------------------------+-----------------+-------------------+
#|1   |[Tokyo, Singapore]      |[Paris, New York]|[false, false]     |
#|2   |[Tokyo, New York, Paris]|[Paris, New York]|[false, true, true]|
#|3   |[Paris]                 |[Paris, New York]|[true]             |
#+----+------------------------+-----------------+-------------------+

You can use array_contains to join the 2 dataframes, aggregate matching Destination with the collect_list, and finally transform the Column B to create the boolean array.

** Assuming Column A is an identifier. (If not, please reply in comment)
<br/>
** transform is only available in Pyspark 3.1+. If you are using lower version of Pyspark, you need to use SQL built-in F.expr('tranform...').

from pyspark.sql import functions as F

df = (df1.join(df2, on=F.array_contains(df1['Column B'], df2.Destination), how='left')
      .groupby('Column A')
      .agg(*[F.first(x).alias(x) for x in df1.columns if x != 'Column A'], 
           F.collect_list('Destination').alias('new column')))

At this point, the df should look like this.

+---------+--------------------+---------+---------+-----------------+
| Column A|            Column B| Column C| Column D|       new column|
+---------+--------------------+---------+---------+-----------------+
|        1|  [Tokyo, Singapore]|  4 hours|    apple|               []|
|        2|[Tokyo, New York,...|1.5 hours|   banana|[Paris, New York]|
|        3|             [Paris]|  2 hours|   orange|          [Paris]|
+---------+--------------------+---------+---------+-----------------+

Then use transform and condition with array_contains, turn the Column B into the boolean array.

df = df.withColumn('new column', F.transform('Column B', lambda x: F.array_contains('new column', x)))

All together.

df = (df1.join(df2, on=F.array_contains(df1['Column B'], df2.Destination), how='left')
      .groupby('Column A')
      .agg(*[F.first(x).alias(x) for x in df1.columns if x != 'Column A'], 
           F.collect_list('Destination').alias('new column'))
      .withColumn('new column', F.transform('Column B', lambda x: F.array_contains('new column', x))))