它在pyspark中很复杂。请参见下面的代码和逻辑

df1 =df1.withColumn('index', row_number().over(Window.partitionBy().orderBy('my_id','parent_id')))#Create index. This can be avaoide if you have a nuique key
df =(df.withColumn('index', row_number().over(Window.partitionBy().orderBy('my_id','parent_id')))#Create index. This can be avaoide if you have a nuique key

# Rename columns to make them similar with df1

     .withColumnRenamed('attribute_1','Attribute_1').withColumnRenamed('attribute_2','attribute2')
.withColumnRenamed('attribute_3','attr_3').withColumnRenamed('attribute_3','attr_3'))

s =(df1.drop('attribute_5').unionByName(df.drop('attribute_4')).orderBy('my_id','parent_id','index')#Union the two dfs and sort bt index
    .withColumn('change',array('attribute_1','attribute2', 'attr_3'))#Create an array of columns being invetsigated for change
    .withColumn('cols',split(lit('attribute_1,attribute2, attr_3'),'\,'))#Introduce list of column names being invetsigated
    .withColumn('change1',last('change').over(Window.partitionBy('index').orderBy('my_id','parent_id')))#For every index,put changes side by side
    .where(col('change')!=col('change1'))#Filter where changes are not the same
    .withColumn('change2', expr("transform(change,(c,i)->change[i]!=change1[i])"))#create boolean of chnages
    .withColumn('faulty_attr',expr('filter(cols,(x,j)->(change2[j]))'))#Leverage arrray functions to filter columns that didnt change as expected
    .drop('index','change','cols','change1' ,'change2')#drop unwated columns
   )
s.show(truncate=False)

+-----+---------+-----------+----------+-------+---------------------+
|my_id|parent_id|Attribute_1|attribute2|attr_3 |faulty_attr          |
+-----+---------+-----------+----------+-------+---------------------+
|ABB  |DEG      |A          |908.8     |Accept |[ attr_3]            |
|ABB  |DEG      |C-         |378.8     |Accept |[attribute_1]        |
|ABC  |DEF      |A-         |478.8     |Decline|[attribute2,  attr_3]|
+-----+---------+-----------+----------+-------+---------------------+

假设A引用第一个dict，B引用第二个dict：

**选项1：**使用 Dataframe 比较

df1 = pd.DataFrame.from_dict(A)
df2 = pd.DataFrame.from_dict(B)

diffs = df1.where(df1.values==df2.values).isnull()
df1['faulty_attr'] = diffs.dot(diffs.columns)

### resulting output:

### my_id parent_id attribute_1  attribute_2 attribute_3  attribute_4             faulty_attr

### 0   ABC       DEF          A-        378.8      Accept        False  attribute_2attribute_3

### 1   ABS       DES          A-        388.8      Accept        False

### 2   ABB       DEG           A        908.8     Decline         True             attribute_3

### 3   ABB       DEG          B-        378.8      Accept        False             attribute_1

### 4   APP       DRE          C-        370.8      Accept         True

**选项2：**使用DeepDiff

from deepdiff import DeepDiff

print(DeepDiff(A, B, ignore_order=False).pretty())

### resulting output:

### Value of root['attribute_1'][3] changed from "B-" to "C-".

### Value of root['attribute_2'][0] changed from 378.8 to 478.8.

### Value of root['attribute_3'][0] changed from "Accept" to "Decline".

### Value of root['attribute_3'][2] changed from "Decline" to "Accept"

**选项3：**使用DataFrame合并

df1 = pd.DataFrame.from_dict(A)
df2 = pd.DataFrame.from_dict(B)

comparison_df = df1.merge(df2, indicator=True, how='outer')
print(comparison_df[comparison_df['_merge'] == 'right_only'])

### resulting output:

### my_id parent_id attribute_1  attribute_2 attribute_3  attribute_4      _merge

### 5   ABC       DEF          A-        478.8     Decline        False  right_only

### 6   ABB       DEG           A        908.8      Accept         True  right_only

### 7   ABB       DEG          C-        378.8      Accept        False  right_only