我有 2 个数据帧,例如 df1
和 df2
。
df1
数据来自数据库,df2
是我从客户端收到的新数据。我需要处理新数据,并根据它是新记录还是要更新的现有记录来执行 UPSERT
。
数据输出示例:
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相同,则为更新,否则为数据库插入。
例如: 用于插入的新数据帧输出:
+--------+--------------+--------------+-----+
|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()
谢谢
最佳答案
这是一个解决方案草图:
将两个框架投影到您的唯一键(邮政编码和地区)
使用 Spark DataFrame 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 数据帧。
关于python - Pyspark - 基于数据框中的 2 列的不同记录,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50864788/