Как сравнить записи из фреймов данных PySpark

Question

Я хотел бы сравнить 2 фрейма данных и вытащить записи, основанные на условиях ниже 3.

1. Если запись совпадает, то «SAME» должно появиться в новом столбце FLAG,
2. Если запись не совпадает, если она из df1 (предположим, № 66), в столбце FLAG должно появиться «DF1».
3. Если запись не совпадает, если она из df2 (предположим, № 77), в столбце FLAG должно появиться «DF2».Здесь всю ЗАПИСЬ нужно рассмотреть и проверить.Мудрое сравнение.
   Также мне нужно проверить, как это, для миллионов записей с использованием кода PySpark.

df1:

No,Name,Sal,Address,Dept,Join_Date
11,Sam,1000,ind,IT,2/11/2019
22,Tom,2000,usa,HR,2/11/2019
33,Kom,3500,uk,IT,2/11/2019
44,Nom,4000,can,HR,2/11/2019
55,Vom,5000,mex,IT,2/11/2019
66,XYZ,5000,mex,IT,2/11/2019

df2:

No,Name,Sal,Address,Dept,Join_Date
11,Sam,1000,ind,IT,2/11/2019
22,Tom,2000,usa,HR,2/11/2019
33,Kom,3000,uk,IT,2/11/2019
44,Nom,4000,can,HR,2/11/2019
55,Xom,5000,mex,IT,2/11/2019
77,XYZ,5000,mex,IT,2/11/2019

Ожидаемый вывод:

No,Name,Sal,Address,Dept,Join_Date,FLAG
11,Sam,1000,ind,IT,2/11/2019,SAME
22,Tom,2000,usa,HR,2/11/2019,SAME
33,Kom,3500,uk,IT,2/11/2019,DF1
33,Kom,3000,uk,IT,2/11/2019,DF2
44,Nom,4000,can,HR,2/11/2019,SAME
55,Vom,5000,mex,IT,2/11/2019,DF1
55,Xom,5000,mex,IT,2/11/2019,DF2
66,XYZ,5000,mex,IT,2/11/2019,DF1
77,XYZ,5000,mex,IT,2/11/2019,DF2

Я загрузил входные данные, как показано ниже, но не понял, как поступить.

df1 = pd.read_csv("D:\\inputs\\file1.csv")

df2 = pd.read_csv("D:\\inputs\\file2.csv")

Любая помощь приветствуется.Спасибо.

Answer 1

# Requisite packages to import
import sys
from pyspark.sql.functions import lit, count, col, when
from pyspark.sql.window import Window

# Create the two dataframes
df1 = sqlContext.createDataFrame([(11,'Sam',1000,'ind','IT','2/11/2019'),(22,'Tom',2000,'usa','HR','2/11/2019'),
                                 (33,'Kom',3500,'uk','IT','2/11/2019'),(44,'Nom',4000,'can','HR','2/11/2019'),
                                 (55,'Vom',5000,'mex','IT','2/11/2019'),(66,'XYZ',5000,'mex','IT','2/11/2019')],
                                 ['No','Name','Sal','Address','Dept','Join_Date']) 
df2 = sqlContext.createDataFrame([(11,'Sam',1000,'ind','IT','2/11/2019'),(22,'Tom',2000,'usa','HR','2/11/2019'),
                                  (33,'Kom',3000,'uk','IT','2/11/2019'),(44,'Nom',4000,'can','HR','2/11/2019'),
                                  (55,'Xom',5000,'mex','IT','2/11/2019'),(77,'XYZ',5000,'mex','IT','2/11/2019')],
                                  ['No','Name','Sal','Address','Dept','Join_Date']) 
df1 = df1.withColumn('FLAG',lit('DF1'))
df2 = df2.withColumn('FLAG',lit('DF2'))

# Concatenate the two DataFrames, to create one big dataframe.
df = df1.union(df2)

Используйте функцию window , чтобы проверить, является ли число одинаковых строк больше 1, и если это действительно так, то пометьте столбец FLAG как SAME, иначе оставьте его таким, как онявляется.Наконец, отбросьте дубликаты.

my_window = Window.partitionBy('No','Name','Sal','Address','Dept','Join_Date').rowsBetween(-sys.maxsize, sys.maxsize)
df = df.withColumn('FLAG', when((count('*').over(my_window) > 1),'SAME').otherwise(col('FLAG'))).dropDuplicates()
df.show()
+---+----+----+-------+----+---------+----+
| No|Name| Sal|Address|Dept|Join_Date|FLAG|
+---+----+----+-------+----+---------+----+
| 33| Kom|3000|     uk|  IT|2/11/2019| DF2|
| 44| Nom|4000|    can|  HR|2/11/2019|SAME|
| 22| Tom|2000|    usa|  HR|2/11/2019|SAME|
| 77| XYZ|5000|    mex|  IT|2/11/2019| DF2|
| 55| Xom|5000|    mex|  IT|2/11/2019| DF2|
| 11| Sam|1000|    ind|  IT|2/11/2019|SAME|
| 66| XYZ|5000|    mex|  IT|2/11/2019| DF1|
| 55| Vom|5000|    mex|  IT|2/11/2019| DF1|
| 33| Kom|3500|     uk|  IT|2/11/2019| DF1|
+---+----+----+-------+----+---------+----+

Answer 2

Я думаю, что вы можете решить вашу проблему с помощью создания временных столбцов для указания источника и join.Тогда вам нужно только проверить условия, то есть, присутствуют ли оба источника или есть только один и какой.

Рассмотрим следующий код:

from pyspark.sql.functions import *


df1= sqlContext.createDataFrame([(11,'Sam',1000,'ind','IT','2/11/2019'),\
(22,'Tom',2000,'usa','HR','2/11/2019'),(33,'Kom',3500,'uk','IT','2/11/2019'),\
(44,'Nom',4000,'can','HR','2/11/2019'),(55,'Vom',5000,'mex','IT','2/11/2019'),\
(66,'XYZ',5000,'mex','IT','2/11/2019')], \
["No","Name","Sal","Address","Dept","Join_Date"])

df2= sqlContext.createDataFrame([(11,'Sam',1000,'ind','IT','2/11/2019'),\
(22,'Tom',2000,'usa','HR','2/11/2019'),(33,'Kom',3000,'uk','IT','2/11/2019'),\
(44,'Nom',4000,'can','HR','2/11/2019'),(55,'Xom',5000,'mex','IT','2/11/2019'),\
(77,'XYZ',5000,'mex','IT','2/11/2019')], \
["No","Name","Sal","Address","Dept","Join_Date"])
#creation of your example dataframes

df1 = df1.withColumn("Source1", lit("DF1"))
df2 = df2.withColumn("Source2", lit("DF2"))
#temporary columns to refer the origin later

df1.join(df2, ["No","Name","Sal","Address","Dept","Join_Date"],"full")\
#full join on all columns, but source is only set if record appears in original dataframe\
.withColumn("FLAG",when(col("Source1").isNotNull() & col("Source2").isNotNull(), "SAME")\
#condition if record appears in both dataframes\
.otherwise(when(col("Source1").isNotNull(), "DF1").otherwise("DF2")))\
#condition if record appears in one dataframe\
.drop("Source1","Source2").show() #remove temporary columns and show result

Вывод:

+---+----+----+-------+----+---------+----+
| No|Name| Sal|Address|Dept|Join_Date|FLAG|
+---+----+----+-------+----+---------+----+
| 33| Kom|3000|     uk|  IT|2/11/2019| DF2|
| 44| Nom|4000|    can|  HR|2/11/2019|SAME|
| 22| Tom|2000|    usa|  HR|2/11/2019|SAME|
| 77| XYZ|5000|    mex|  IT|2/11/2019| DF2|
| 55| Xom|5000|    mex|  IT|2/11/2019| DF2|
| 11| Sam|1000|    ind|  IT|2/11/2019|SAME|
| 66| XYZ|5000|    mex|  IT|2/11/2019| DF1|
| 55| Vom|5000|    mex|  IT|2/11/2019| DF1|
| 33| Kom|3500|     uk|  IT|2/11/2019| DF1|
+---+----+----+-------+----+---------+----+