我最后做的是：

val editedDF = Seq(("A",3),("B",3)).toDF("id","timestamp")
val messageDF = Seq(("A",2),("B",2),("A",2),("A",3),("B",4),("A",2),("B",2),("c",9)).toDF("id","timestamp")

最后我使用了这个连接：

// Filter out the edited meesages.
    val editedFilteredDF  = messageDF.join(editedDF,
(editedDF("id") === messageDF("id")) && (editedDF("timestamp") > messageDF("timestamp")),
joinType="left_anti")

结果是：

editedFilteredDF.show()
+---+---------+
| id|timestamp|
+---+---------+
|  A|        3|
|  B|        4|
|  c|        9|
+---+---------+

您可以聚合消息表，将其与已编辑的表联接并筛选

import pyspark.sql.functions as F

# Test dataframe

tst=sqlContext.createDataFrame([('A',2),('B',2),('A',2),('A',3),('B',4),('A',2),('B',2),('c',9)],schema=("id","count"))
tst1 = sqlContext.createDataFrame([('A',4),('B',1)],schema=("id","val"))

# Aggregate and join

tst_g=tst.groupby('id').agg(F.max('count').alias('count'))
tst_j= tst_g.join(tst1,tst_g.id==tst1.id,'left')

# Filter result

tst_f = tst_j.where((F.col('count')>=F.col('val'))|(F.col('val').isNull()))

结果是：

tst_j.show()

+---+-----+----+----+
| id|count|  id| val|
+---+-----+----+----+
|  c|    9|null|null|
|  B|    4|   B|   1|
|  A|    3|   A|   4|
+---+-----+----+----+
 tst_f.show()
+---+-----+----+----+
| id|count|  id| val|
+---+-----+----+----+
|  c|    9|null|null|
|  B|    4|   B|   1|
+---+-----+----+----+

最后，可以删除不相关的列。
如果需要完整的数据，那么可以将update表与message表连接起来，并执行相同的操作。如果更新表很小，那么出于性能原因考虑使用广播连接。

# Approach to join with full table

# Test dataframe

tst=sqlContext.createDataFrame([('A',2),('B',2),('A',2),('A',3),('B',4),('A',2),('B',2),('c',9)],schema=("id","count"))
tst1 = sqlContext.createDataFrame([('A',4),('B',1)],schema=("id","val"))

# %%

# join with the full table

tst_j= tst.join(tst1,tst.id==tst1.id,'left')

# Filter result

tst_f = tst_j.where((F.col('count')>=F.col('val'))|(F.col('val').isNull()))

提示：如果不想在结果中包含两个id列，可以将join语法更改为tst.join（tst1，on=“id”，how='left'）