PySpark: как объединить 2 DataFrame с одинаковыми столбцами и создать новый DataFrame с уникальными и более новыми строками

Question

У меня есть 2 DataFrames, и я должен объединить оба и создать новый DataFrame с уникальными и последними строками, например,

DataFrame 1:

+------+-------+---------+-----+-------+
|col A |col B  |col C    |col D|col E  |
+------+-------+---------+-----+-------+
|city  |prod 1 |9/29/2017|358  |193    |
|city  |prod 2 |8/25/2017|50   |201    |
|city  |prod 3 |9/9/2017 |236  |169    |
|city  |prod 4 |9/28/2017|358  |193    |
|city  |prod 5 |8/24/2017|50   |201    |
|city  |prod 6 |9/8/2017 |236  |169    |
+------+-------+---------+-----+-------+

DataFrame 2:

+------+-------+---------+-----+-------+
|col A |col B  |col C    |col D|col E  |
+------+-------+---------+-----+-------+
|city  |prod 1 |9/29/2018|359  |197    |
|city  |prod 2 |8/25/2018|51   |209    |
|city  |prod 3 |9/9/2018 |237  |181    |
|city  |prod 7 |9/28/2017|358  |193    |
|city  |prod 8 |8/24/2017|50   |201    |
|city  |prod 9 |9/8/2017 |236  |169    |
+------+-------+---------+-----+-------+

Мне нужно объединить 2 DataFrames от col A и col B, и результирующий DataFrame должен выглядеть следующим образом.

Обратите внимание, что prod 1, 2 и 3 присутствует в обоих DataFrame, но взят из DataFrame 2, поскольку он имеет самую последнюю дату (столбец C)

DataFrame 3:

+------+-------+---------+-----+-------+
|col A |col B  |col C    |col D|col E  |
+------+-------+---------+-----+-------+
|city  |prod 1 |9/29/2018|359  |197    |
|city  |prod 2 |8/25/2018|51   |209    |
|city  |prod 3 |9/9/2018 |237  |181    |
|city  |prod 4 |9/28/2017|358  |193    |
|city  |prod 5 |8/24/2017|50   |201    |
|city  |prod 6 |9/8/2017 |236  |169    |
|city  |prod 7 |9/28/2017|358  |193    |
|city  |prod 8 |8/24/2017|50   |201    |
|city  |prod 9 |9/8/2017 |236  |169    |
+------+-------+---------+-----+-------+

Пожалуйста, дайте мне знать, как я могу этого достичь.

Answer 1

Давайте предположим, что ваши фреймы данных имеют только первые четыре строки (ради пространства!)

df_1.show()
+-----+------+---------+-----+-----+
|col A| col B|    col C|col D|col E|
+-----+------+---------+-----+-----+
| city|prod 1|9/29/2017|  358|  193|
| city|prod 2|8/25/2017|   50|  201|
| city|prod 3| 9/9/2017|  236|  169|
| city|prod 4|9/28/2017|  358|  193|
+-----+------+---------+-----+-----+

df_2.show()
+-----+------+---------+-----+-----+
|col A| col B|    col C|col D|col E|
+-----+------+---------+-----+-----+
| city|prod 1|9/29/2018|  359|  197|
| city|prod 2|8/25/2018|   51|  209|
| city|prod 3| 9/9/2018|  237|  181|
| city|prod 4|9/28/2017|  358|  193|
+-----+------+---------+-----+-----+

Тогда один из способов добиться того, что вы хотите, заключается в следующем:

from pyspark.sql import Window
## change parameters in w, for partition and order
w = Window.partitionBy('col A', 'col B').orderBy('col C').rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
## union just appends the dataframes
df = df_1.union(df_2)
## generate a feature which for latest row is True
df = df.withColumn('latest_record', F.col("col C")==F.last("col C").over(w))
## keep those rows in which the new column is True, and remove duplicates
df = df.filter(F.col('latest_record')).select('col A', 'col B', 'col C', 'col D', 'col E').distinct()
df.sort('col B').show()
+-----+------+---------+-----+-----+
|col A| col B|    col C|col D|col E|
+-----+------+---------+-----+-----+
| city|prod 1|9/29/2018|  359|  197|
| city|prod 2|8/25/2018|   51|  209|
| city|prod 3| 9/9/2018|  237|  181|
| city|prod 4|9/28/2017|  358|  193|
+-----+------+---------+-----+-----+