如何使用 pySpark 比较两个 CSV 文件并验证是否存在

Question

所以，我有一个像这样的 input.csv：

First_Name  Last_Name   Birthdate   Gender  Email_ID        Mobile
Smit        Will        21-04-1974  M       da1@gmail.com   5224521452
Bob         Builder     14-03-1992  M       ad4@gmail.com   2452586253

还有包含更多记录的 Database.csv：

First_Name  Last_Name   Birthdate   Gender  Email_ID        Mobile
Bob         Micheles    10-04-1982  M       ya4@gmail.com   7845214525
Will        Smith       21-04-1974  M       da1@gmail.com   9874521452
Emma        Watson      21-08-1989  F       emma@gmail.com  5748214563
Emma        Smit        21-08-1999  F       da1@gmail.com   9874521452
bob         robison     14-03-1992  M       za@gmail.com    2452586253

df_DataBase = spark.read.csv("DataBase.csv",inferSchema=True,header=True) 我的预期结果是：

1. Bob Builder 与 Bob robison 相同，只是他的 Last_Name 和 Email_ID 不同
2. Smit Will 和 Will Smith 只是 姓名 相同，而 手机号码 不同。 最后打印它们是否存在于现有输入文件中，如下所示：

注意：如果电子邮件、电话和生日不匹配，则此人不同。

如果我们能做到这一点，那么使用 pyspark 我会很棒。

Answer 1

您可以尝试以下方法：

ip = spark.read.csv("input.csv")
db = spark.read.csv("database.csv")
#condition if person is same
person_exists = [((col('a.Email_id') == col('b.Email_id')) | (col('a.Mobile') == col('b.Mobile')) | (col('a.Birthdate') == col('b.Birthdate'))) ]

#people existing in db
existing_persons = 
ip.alias('a').join(db.alias('b'),person_exists,"inner").select([col('a.'+x) for x in a.columns])

#people not existing in db
non_existing = ip.subtract(existing_persons)

#add a column to indicate if same person or not
existing_persons = existing_persons.withColumn('Same_Person',lit('Yes'))
non_existing = non_existing.withColumn('Same_Person',lit('No'))