Отрегулировать следующую дату

Question

У меня есть набор данных (df) со следующими столбцами:

ID	Date		Next Date	Datediff in days
1	2017-10-25	null		null
1	2016-12-13	2017-10-25	316
1	2016-11-23	2016-12-13	20
1	2016-11-14	2016-11-23	9
1	2015-07-07	2016-11-14	496
2	2015-05-15	null		null
2	2013-11-29	2015-05-15	532
2	2013-11-16	2013-11-29	13
2	2009-02-06	2013-11-16	1744
2	2006-06-21	2009-02-06	961
2	2002-06-25	2006-06-21	1457

В этом случае «Следующая дата» действительна только в том случае, если разница дат превышает 30 дней. Теперь я хочу заменить записи следующей даты действительной следующей датой и создать столбец «Новая следующая дата» следующим образом:

ID	Date		Next Date	Datediff in days	New Next Date
1	2017-10-25	null		null			null
1	2016-12-13	2017-10-25	316			2017-10-25
1	2016-11-23	2016-12-13	20			2017-10-25
1	2016-11-14	2016-11-23	9			2017-10-25
1	2015-07-07	2016-11-14	496			2016-11-14
2	2015-05-15	null		null			null
2	2013-11-29	2015-05-15	532			2015-05-15
2	2013-11-16	2013-11-29	13			2015-05-15
2	2009-02-06	2013-11-16	1744			2013-11-16
2	2006-06-21	2009-02-06	961			2009-02-06
2	2002-06-25	2006-06-21	1457			2006-06-21

Я думал об использовании функции запаздывания для создания вспомогательного столбца для замены дат, которые меньше 30 дней. Но для идентификатора клиента 1 может случиться так, что две недопустимые даты будут последовательными. Тогда это не работает.

Answer 1

Вы можете использовать функцию отставания, чтобы добавить столбец lagDate, используйте оператор if else, используя datediff , чтобы сравнить разницу:

 import org.apache.spark.sql.functions.{when, col, lag, datediff, lag, date_add}
 val windowId = Window.partitionBy(col("ID")).orderBy("Date")
 val newDf =  df.withColumn("dateLimit", date_add(col("Date"), 30))
  .withColumn("lagDate1", lag(col("Date"), -1).over(windowId))
  .withColumn("lagDate2", lag(col("Date"), -2).over(windowId))
  .withColumn("lagDate", 
    when(datediff(col("lagDate1"), col("dateLimit")) >=0, col("lagDate1")).otherwise(col("lagDate2"))
  )
  .withColumn("correctNextDate",
    when(datediff(col("NextDate"), col("Date")) > 30 || col("NextDate").isNull,
      col("NextDate")).
      otherwise(col("lagDate"))
  )