验证一个 pyspark Dataframe 中的行数据与另一个 Dataframe 匹配

Question

我有 2 个 Pyspark 数据框 df1,df2。 df1 和 df2 都包含数百万条记录。

df1 是这样的：

+-------------------+--------+--------+
|               name|state   | pincode|
+-------------------+--------+--------+
|  CYBEX INTERNATION| HOUSTON| 00530  |
|        FLUID POWER| MEDWAY | 02053  |
|   REFINERY SYSTEMS| FRANCE | 072234 |
|    K N ENTERPRISES| MUMBAI | 100010 |
+-------------------+--------+--------+

df2 是这样的：

+--------------------+--------+--------+
|               name |state   | pincode|
+--------------------+--------+--------+
|FLUID POWER PVT LTD | MEDWAY | 02053  |
|  CYBEX INTERNATION | HOUSTON| 02356  |
|REFINERY SYSTEMS LTD| MUMBAI | 072234 |
+--------------------+--------+--------+

所以，我想根据名称状态检查是否在 df2 上找到了 df1，并且 Pincode 和输出应该是否经过验证，即行找到它将是 1，否则 0 和 df 将是

+-------------------+--------+--------+--------- --+
|               name|state   | pincode|  Validated |
+-------------------+--------+--------+---------- -+
|  CYBEX INTERNATION| HOUSTON| 00530  |     0      |
|        FLUID POWER| MEDWAY | 02053  |     1      |
|   REFINERY SYSTEMS| FRANCE | 072234 |     0      |
|    K N ENTERPRISES| MUMBAI | 100010 |     0      |
+-------------------+--------+--------+------------+

在 df1 Pincode 的第 1 行的第一种情况下，如此验证的任何 df2 Pincode 列都不匹配 = 0
在 df1 Pincode 的第 2 行匹配的第二种情况下，状态也匹配，对于 name 列，我使用 Levenshtein 匹配列名，最后一行被验证 = 1
在第 3 行 Pincode 匹配但状态不匹配且已验证 = 0
在第 4 个 Pincode 中不存在并且已验证 = 0

我在嵌套 if 中迭代数据时使用 Pandas dataFrame 进行了尝试，但数据太大，迭代不是一个好的选择。

我希望使用 pyspark 并利用并行处理来加快进程，例如：

df_final = df1.withColumn('validated', if some_expression == True THEN 1,ELSE 0)

但无法弄清楚 some_expression，以及如何在另一个 df2 上检查整个 df1 验证给定列且无需任何迭代。

我经历了不同的火花问题和类似的问题，但没有一个对我有帮助。 任何帮助将不胜感激。如果有任何信息不清楚，请发表评论。

Answer 1

将levenshtein-distance 与左连接一起使用，您可以执行以下操作：

join_condition = (col("df1.pincode") == col("df2.pincode")) \
                 & (levenshtein(col("df1.name"), col("df2.name")) <= 10) \
                 & (col("df1.state") == col("df2.state"))

result_df = df1.alias("df1").join(df2.alias("df2"), join_condition , "left")

result_df.select("df1.*",
              when(col("df2.name").isNotNull(), lit(1)).otherwise(lit(0)).alias("validated")
              ).show()

#+-----------------+-------+-------+---------+
#|             name|  state|pincode|validated|
#+-----------------+-------+-------+---------+
#|CYBEX INTERNATION|HOUSTON|  00530|        0|
#|      FLUID POWER| MEDWAY|  02053|        1|
#| REFINERY SYSTEMS| FRANCE| 072234|        0|
#|  K N ENTERPRISES| MUMBAI| 100010|        0|
#+-----------------+-------+-------+---------+