Compare two different columns from two different pyspark dataframe

Question

I'm trying to compare two different columns which are in two different data frames, and if I found a match I'm returning value 1 else None -

df1 =

df2 =

df1 (Expected_Output) =

I have tried the below code -

def getImpact(row):
match = df2.filter(df2.second_key == row)
if match.count() > 0:
    return 1
return None

udf_sol = udf(lambda x: getImpact(x), IntegerType())
df1 = df1.withcolumn('impact',udf_sol(df1.first_key))

But getting below error - TypeError: cannot pickle '_thread.RLock' object

Can anyone help me to achieve the expected output as shown above?

Thanks

Answer 1

Assuming first_key and second_key are unique , you can opt for a join across the dataframes -

More examples and explanation can be found here

from pyspark import SparkContext
from pyspark.sql import SQLContext
from functools import reduce
import pyspark.sql.functions as F

from pyspark.sql import Window


data_list1 = [
    ("abcd","Key1")
    ,("jkasd","Key2")
    ,("oigoa","Key3")
    ,("ad","Key4")
    ,("bas","Key5")
    ,("lkalsjf","Key6")
    ,("bsawva","Key7")
]

data_list2 = [
    ("cashj","Key1",10)
    ,("ax","Key11",12)
    ,("safa","Key5",21)
    ,("safasf","Key6",78)
    ,("vasv","Key3",4)
    ,("wgaga","Key8",0)
    ,("saasfas","Key7",10)
]

sparkDF1 = sql.createDataFrame(data_list1,['data','first_key'])
sparkDF2 = sql.createDataFrame(data_list2,['temp_data','second_key','frinks'])


>>> sparkDF1
+-------+---------+
|   data|first_key|
+-------+---------+
|   abcd|     Key1|
|  jkasd|     Key2|
|  oigoa|     Key3|
|     ad|     Key4|
|    bas|     Key5|
|lkalsjf|     Key6|
| bsawva|     Key7|
+-------+---------+

>>> sparkDF2
+---------+----------+------+
|temp_data|second_key|frinks|
+---------+----------+------+
|    cashj|      Key1|    10|
|       ax|     Key11|    12|
|     safa|      Key5|    21|
|   safasf|      Key6|    78|
|     vasv|      Key3|     4|
|    wgaga|      Key8|     0|
|  saasfas|      Key7|    10|
+---------+----------+------+

#### Joining the dataframes on common columns 
finalDF = sparkDF1.join(
                sparkDF2
             ,(sparkDF1['first_key'] == sparkDF2['second_key'])
            ,'left'
).select(sparkDF1['*'],sparkDF2['frinks']).orderBy('frinks')


### Identifying impact if the frinks value is Null or Not
finalDF = finalDF.withColumn('impact',F.when(F.col('frinks').isNull(),0).otherwise(1))

>>> finalDF.show()

+-------+---------+------+------+
|   data|first_key|frinks|impact|
+-------+---------+------+------+
|  jkasd|     Key2|  null|     0|
|     ad|     Key4|  null|     0|
|  oigoa|     Key3|     4|     1|
|   abcd|     Key1|    10|     1|
| bsawva|     Key7|    10|     1|
|    bas|     Key5|    21|     1|
|lkalsjf|     Key6|    78|     1|
+-------+---------+------+------+