Фильтр на объединенном фрейме данных не работает в pyspark

Question

У меня есть фрейм данных с 3 следующими столбцами

• student_id
• name
• timestamp

Есть несколько строк дляидентификатор студента с разными именами вместе с отметкой времени, когда запись фактически обновлена. Я хотел получить 2 разных кадра данных.

• unique_data (строки всех идентификаторов студентов с последней отметкой времени для этого идентификатора студента)

• duplicate_data (все строки из входного фрейма данных, кроме приведенного вышеупомянутые строки unique_data)

У меня есть следующий код, который генерирует 2 кадра данных

input_frame.show()
+----------+----------+---------+
|student_id|name      |timestamp|
+----------+----------+---------+
|        s1|testuser  |       t1|
|        s1|sampleuser|       t2|
|        s2|test123   |       t1|
|        s2|sample123 |       t2|
+----------+----------+---------+

# Assuming t2 > t1

unique_data = input_frame.sort(sf.desc(timestamp))drop_duplicates("student_id")
unique_data.show()
+----------+----------+---------+
|student_id|name      |timestamp|
+----------+----------+---------+
|        s1|sampleuser|       t2|
|        s2|sample123 |       t2|
+----------+----------+---------+

input_frame = input_frame.alias('input_frame')
unique_frame = unique_frame.alias('unique_frame')

joined_data = input_frame.join(unique_data, input_frame["student_id"] == unique_data["student_id"], how="left")
joined_data.show()
+----------+----------+---------+----------+----------+---------+
|student_id|name      |timestamp|student_id|name      |timestamp|
+----------+----------+---------+----------+----------+---------+
|        s1|testuser  |       t1|        s1|sampleuser|       t2|
|        s1|sampleuser|       t2|        s1|sampleuser|       t2|
|        s2|test123   |       t1|        s2|sample123 |       t2|
|        s2|sample123 |       t2|        s2|sample123 |       t2|
+----------+----------+---------+----------+----------+---------+



duplicate_data = joined_data.filter(input_frame["timestamp"] != unique_data["timestamp"]).select("input_frame.*")
duplicate_data.show()
+----------+----+---------+
|student_id|name|timestamp|
+----------+----+---------+
+----------+----+---------+

Answer 1

Мы должны использовать псевдоним в условии фильтра, так как кадры имеют похожие столбцы

from pyspark.sql import functions as sf

input_frame = input_frame.alias('input_frame')
unique_frame = unique_frame.alias('unique_frame')

duplicate_data = joined_data.filter(sf.col("input_frame.timestamp") != sf.col("unique_data.timestamp")).select("input_frame.*")
duplicate_data.show()
+----------+----------+---------+
|student_id|name      |timestamp|
+----------+----------+---------+
|        s1|testuser  |       t1|
|        s2|test123   |       t1|
+----------+----------+---------+

Answer 2

для unique_data["timestamp"], если вы хотите взять весь столбец, спарк не знает, о какой строке вы говорите. вам разрешено делать что-то вроде:

duplicate_data = joined_data.filter((joined_data.timestamp != unique_data.collect()[0]['timestamp'])), который говорит, что join_data.timestamp не равен первой строке unique_data, [row 0] [timestamp]. или вы можете перебирать каждую строку unique_data и проверять, равны они или нет.