Spark (используя pyspark) использует значение в одном фрейме данных (структурированная потоковая передача) для запроса статического фрейма данных и слияния строки из второго df с первым

Question

У меня есть структурированный потоковый фрейм данных, который я хочу для каждой строки взять значение в одном столбце (в данном случае временную метку, например 1525670700) и использовать это значение для запроса другого статического фрейма данных для ближайшей временной метки к этому значению и объединить одну строку, возвращенную из этого запроса, со структурным потоком данных, например,

my_row_to_merge_df = weather_df.filter(weather_df.timestamp_unix > 1525670700).sort(col('timestamp_unix').asc()).limit(1).show()

С этим я хочу сделать что-то вроде:

joined_df = streaming_df.merge(function_to_return_row_from_other_df(col('timestamp')))

Проблема с использованием объединения в этом случае заключается в том, что временные метки могут не совпадать идеально, но это нормально, если они отличаются на несколько секунд или даже на несколько минут.

Итак, я не уверен, какую операцию я мог бы сделать, чтобы получить этот результат.

Answer 1

Вы можете округлить отметку времени до необходимого вам уровня значимости:

# Assuming timestamp is in seconds
timestamp_rounded = int(timestamp - (timestamp % int(minutes * 60)))

Где "минуты" - это уровень, на который вы хотите округлить. Например, если вы выберете минуты = 5, вы будете округлять (уменьшать) метку времени каждые 5 минут.

Вы можете добавить новый столбец, как показано здесь: Как выполнить математическую операцию с двумя столбцами в кадре данных, используя pyspark