我有两个数据框 df1(旧) 和 df2(新)。我想比较 df2 和 df1 并找出新增的行、删除的行、更新的行以及被更新的列的名称。
以下是我编写的代码:
from pyspark.sql.functions import col, array, when, array_remove, lit
data1 = [("James","rob","Smith","36636","M",3000),
("Michael","Rose","jim","40288","M",4000),
("Robert","dunkin","Williams","42114","M",4000),
("Maria","Anne","Jones","39192","F",4000),
("Jen","Mary","Brown","60563","F",-1)
]
data2 = [("James","rob","Smith","36636","M",3000),
("Robert","dunkin","Williams","42114","M",2000),
("Maria","Anne","Jones","72712","F",3000),
("Yesh","Reddy","Brown","75234","M",3000),
("Jen","Mary","Brown","60563","F",-1)
]
schema = StructType([ \
StructField("firstname",StringType(),True), \
StructField("middlename",StringType(),True), \
StructField("lastname",StringType(),True), \
StructField("id", StringType(), True), \
StructField("gender", StringType(), True), \
StructField("salary", IntegerType(), True) \
])
df1 = spark.createDataFrame(data=data1,schema=schema)
df2 = spark.createDataFrame(data=data2,schema=schema)
conditions_ = [when(df1[c]!=df2[c], lit(c)).otherwise("") for c in df1.columns if c not in ['firstname','middlename','lastname']]
select_expr =[
col("firstname"),col("middlename"),col("lastname"),
*[df2[c] for c in df2.columns if c not in ['firstname','middlename','lastname']],
array_remove(array(*conditions_), "").alias("updated_columns")
]
df1.join(df2, ["firstname","middlename","lastname"],"inner").select(*select_expr).show()
这是我得到的输出。
+---------+----------+--------+-----+------+------+---------------+
|firstname|middlename|lastname| id|gender|salary|updated_columns|
+---------+----------+--------+-----+------+------+---------------+
| James| rob| Smith|36636| M| 3000| []|
| Robert| dunkin|Williams|42114| M| 2000| [salary]|
| Maria| Anne| Jones|72712| F| 3000| [id, salary]|
| Jen| Mary| Brown|60563| F| -1| []|
+---------+----------+--------+-----+------+------+---------------+
这里是我期望得到的输出结果
+---------+----------+--------+-----+------+------+---------------+-----------------+
|firstname|middlename|lastname| id|gender|salary|updated_columns| status|
+---------+----------+--------+-----+------+------+---------------+-----------------+
| James| rob| Smith|36636| M| 3000| []| unchanged|
| Robert| dunkin|Williams|42114| M| 2000| [salary]| updated|
| Michael| Rose| jim|40288| M| 4000| []| deleted|
| Maria| Anne| Jones|72712| F| 3000| [id, salary]| updated|
| Yesh| Reddy| Brown|75234| M| 3000| []| added|
| Jen| Mary| Brown|60563| F| -1| []| unchanged|
+---------+----------+--------+-----+------+------+---------------+-----------------+
我知道可以使用左反 join 分别查找添加和删除的行。但是,我正在寻找更新现有 join 的方法,以获得上述输出。