Эффективное определение max (date) в pyspark без группового вызова или вызова collect ()

Question

1) Мне нужно использовать date_diff() в моем коде, чтобы найти разницу между столбцом Date и Max(Date)

Что я сейчас использую

from pyspark.sql import functions as F
max_date = df.select(F.max(df['date'])).collect()[0][0]
df = df.withColumn('period_difference', F.datediff(F.lit(max_date), df['date']))

Это займет около 6 минут, чтобы закончить. Есть ли лучший эффективный способ сделать это?

2) Есть ли способ получить разницу, возвращаемую date_diff() в месяцах? В пандах вы можете использовать:

import numpy as np
df['period_difference'] = ((max(df['date']) - df['date']) / np.timedelta64(1, 'M'))

Я просто пытаюсь скопировать вышеприведенную строку кода в pandas в pyspark.

Answer 1

Вы можете отложить задержку исполнения

max_date = df.select(F.max(df['date']).alias("max_date"))    
df.crossJoin(max_date).withColumn(
   'period_difference', F.datediff(F.col('max_date'), F.col('date')
)

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