Pyspark - 基于数据框中的 2 列的不同记录

Posted 2023-04-17

技术标签:

【中文标题】Pyspark - 基于数据框中的 2 列的不同记录【英文标题】：Pyspark - distinct records based on 2 columns in dataframe 【发布时间】：2018-06-14 19:42:26 【问题描述】：

我有 2 个数据框，比如 df1 和 df2。

df1 数据来自数据库，df2 是我从客户那里收到的新数据。我需要处理新数据，并根据是新记录还是要更新的现有记录执行UPSERTs。

样本数据输出：

df1= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 01","NJ"),("xxx2","81A01","TERR NAME 01","NJ"),("xxx3","81A01","TERR NAME 01","NJ"),("xxx4","81A01","TERR NAME 01","CA"),("xx5","81A01","TERR NAME 01","ME")], ["zip_code","territory_code","territory_name","state"])
df2= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 55","NY"),("xxx2","81A01","TERR NAME 55","NY"),("x103","81A01","TERR NAME 01","NJ")], ["zip_code","territory_code","territory_name","state"])

df1.show()
+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 01|   NJ|
|    xxx2|         81A01|  TERR NAME 01|   NJ|
|    xxx3|         81A01|  TERR NAME 01|   NJ|
|    xxx4|         81A01|  TERR NAME 01|   CA|
|    xxx5|         81A01|  TERR NAME 01|   ME|
+---------------------------------------------

# Print out information about this data
df2.show()
+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 55|   NY|
|    xxx2|         81A01|  TERR NAME 55|   NY|
|    x103|         81A01|  TERR NAME 01|   NJ|
+---------------------------------------------

预期结果：我需要将 df2 数据帧与 df1 进行比较。根据上述比较创建 2 个新数据集，即要更新的记录和要附加/插入到数据库的记录。

如果 zip_code 和 territory_code 相同，则为 UPDATE，否则为 INSERT 到数据库。

例如： INSERT 的新数据帧输出：

 +--------+--------------+--------------+-----+
 |zip_code|territory_code|territory_name|state|
 +--------+--------------+--------------+-----+
 |    x103|         81A01|  TERR NAME 01|   NJ|
 +---------------------------------------------

更新的新数据框：

+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 55|   NY|
|    xxx2|         81A01|  TERR NAME 55|   NY|
+---------------------------------------------

有人可以帮帮我吗？我正在使用 AWS Glue。

更新：解决方案（使用连接和减去）

df3 = df1.join(df2, (df1.zip_code == df2.zip_code_new) & (df1.territory_code == df2.territory_code_new))
df5=df3.drop("zip_code", "territory_code", "territory_name", "state")
df5.show()

+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|        x103|             81A01|      TERR NAME 01|       NJ|
+------------+------------------+------------------+---------+

df4=df2.subtract(df5)
df4.show()

+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|    xxx1    |         81A01    |  TERR NAME 55    |   NY    |
|    xxx2    |         81A01    |  TERR NAME 55    |   NY    |
+------------------------------------------------------------+

对于 RDS 数据库更新，我使用 pymysql/Mysqldb：

db = MySQLdb.connect("xxxx.rds.amazonaws.com", "username", "password", "databasename")
cursor = db.cursor()

#cursor.execute("REPLACE INTO table SELECT * FROM table_stg")
insertQry = "INSERT INTO table VALUES('xxx1','81A01','TERR NAME 55','NY') ON DUPLICATE KEY UPDATE territory_name='TERR NAME 55', state='NY'"
n=cursor.execute(insertQry)
db.commit()
cursor.fetchall()
db.close()

谢谢

【问题讨论】：

下次请把你的问题写得更好。它（也许仍然）很难理解。也许您还应该提供一个数据集（我的意思是 4-5 个样本数据），以便其他人可以测试您的代码。我想，我已经提供了足够的信息，并且我的问题很清楚，不确定是什么导致了反对票。无论如何，我将编辑问题并使其更短。 【参考方案1】：

这是一个解决方案草图：

将两个帧投影到您的唯一密钥（邮政编码和领土）上

使用 spark 数据帧 api 计算两个数据帧之间的交集和差异。见此链接：How to obtain the difference between two DataFrames?

更新键的交集

插入差异（在新数据框内，而不是在现有数据内）

在 scala 中这看起来像这样——在 python 中应该非常相似：

import org.apache.spark.sql.SparkSession

case class ZipTerr(zip_code: String, territory_code: String, 
    territory_name: String, state:String)

case class Key(zip_code: String, territory_code: String)

val spark: SparkSession

val newData = spark.createDataFrame(List(
  ZipTerr("xxx1", "81A01", "TERR NAME 01", "NJ"),
  ZipTerr("xxx2", "81A01", "TERR NAME 01", "NJ"),
  ZipTerr("xxx3", "81A01", "TERR NAME 01", "NJ"),
  ZipTerr("xxx4", "81A01", "TERR NAME 01", "CA"),
  ZipTerr("xx5","81A01","TERR NAME 01","ME")
))

val oldData = spark.createDataFrame(List(
  ZipTerr("xxx1","81A01","TERR NAME 55","NY"),
  ZipTerr("xxx2","81A01","TERR NAME 55","NY"),
  ZipTerr("x103","81A01","TERR NAME 01","NJ")
))

val newKeys = newData.map(z => Key(z.getAs("zip_code"), z.getAs("territory_code")))
val oldKeys = oldData.map(z => Key(z.getAs("zip_code"), z.getAs("territory_code")))

val keysToInsert = newKeys.except(oldKeys)
val keysToUpdate = newKeys.intersect(oldKeys)

这有帮助吗？

注意：变量的名称表明您正在使用胶合动态框架。然而，您正在使用 sqlContext.createDataFrame 函数为它们分配普通的 spark 数据帧。

【讨论】：

谢谢@user152468，我会试试这个并随时通知您。有没有办法通过加入和删除不需要的列来根据keyToInsert / keyToUpdate中的键从newData / oldData中提取整个记录？抱歉，如果它是基础知识，但我仍在学习 pyspark。您好 user152468，根据您的建议以及其他参考，我能够获得所需的结果。我已经更新了问题中的代码和结果。只是一个注释，认为 except 被替换为减法，就我谷歌搜索而言，pyspark 中没有相交。谢谢。嗨@YuvaKumar。很高兴你做到了。 Pyspark 在语言结构上似乎与普通 spark 大不相同。【参考方案2】：

为了清楚起见，我在这里用代码 sn-ps 复制解决方案：

df1= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 01","NJ"),("xxx2","81A01","TERR NAME 01","NJ"),("xxx3","81A01","TERR NAME 01","NJ"),("xxx4","81A01","TERR NAME 01","CA"),("xx5","81A01","TERR NAME 01","ME")], ["zip_code","territory_code","territory_name","state"])
df2= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 55","NY"),("xxx2","81A01","TERR NAME 55","NY"),("x103","81A01","TERR NAME 01","NJ")], ["zip_code_new","territory_code_new","territory_name_new","state"])

df1.show()
+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 01|   NJ|
|    xxx2|         81A01|  TERR NAME 01|   NJ|
|    xxx3|         81A01|  TERR NAME 01|   NJ|
|    xxx4|         81A01|  TERR NAME 01|   CA|
|    xxx5|         81A01|  TERR NAME 01|   ME|
+---------------------------------------------

# Print out information about this data
df2.show()
+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|    xxx1    |         81A01    |  TERR NAME 55    |   NY    |
|    xxx2    |         81A01    |  TERR NAME 55    |   NY    |
|    x103    |         81A01    |  TERR NAME 01    |   NJ    |
+------------------------------------------------------------+

获取新记录，可以使用“append”操作将其插入到mysql中

df3 = df1.join(df2, (df1.zip_code == df2.zip_code_new) & (df1.territory_code == df2.territory_code_new))
df5=df3.drop("zip_code", "territory_code", "territory_name", "state")
df5.show()

+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|        x103|             81A01|      TERR NAME 01|       NJ|
+------------+------------------+------------------+---------+

然后获取剩余的需要更新到mysql数据库的记录。我们可以使用arr = df1.collect()，后跟for r in arr:，以防纯python需要，否则我们可以使用pandas迭代器处理每条记录。

df4=df2.subtract(df5)
df4.show()

+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|    xxx1    |         81A01    |  TERR NAME 55    |   NY    |
|    xxx2    |         81A01    |  TERR NAME 55    |   NY    |
+------------------------------------------------------------+

希望这可以帮助有需要的人。请让我知道在上述情况下是否有更好的数据帧迭代方法。谢谢

【讨论】：

以上是关于Pyspark - 基于数据框中的 2 列的不同记录的主要内容，如果未能解决你的问题，请参考以下文章

如何更改pyspark数据框中列的顺序？

如何通过 Pyspark 中同一数据框中另一列的正则表达式值过滤数据框中的一列

Pyspark：如何将现有非空列的元组列表作为数据框中的列值之一返回

pyspark 数据框中所有列的总计数为零

将重复记录合并到 pyspark 数据框中的单个记录中

从 Pyspark 中的数据框中计算空值和非空值