Найти среднее и общее время между датами в строке строки в pyspark?

Question

В столбце spark df есть список упорядоченных дат , удаленный с помощью '|' , Это выглядит так:

+------------------------------------------------------------+
|      date                                                  |
+------------------------------------------------------------+
|2018-08-01|2017-06-01|2015-11-01|2012-08-21|2010-04-02      |
|2016-08-11|2013-08-25|2013-04-01|2012-01-01                 |
+------------------------------------------------------------+

Я хочу рассчитать среднее время для каждой строки и общую разницу (последняя дата - первая дата) в днях, например:

+------------------------------------------------------------+---------------+---------------+
|      date                                                  | average time  | total time    |
+------------------------------------------------------------+---------------+---------------+
|2018-08-01|2017-06-01|2015-11-01|2012-08-21|2010-08-01      |   584         | 2920          |
|2016-08-11|2013-08-25|2013-04-01|2012-08-11                 |   365         | 1460          |
+------------------------------------------------------------+---------------+---------------+

Answer 1

Мы можем использовать встроенный array functions кадра данных для этого случая ..

Example:

val df= Seq(("2018-08-01|2017-06-01|2015-11-01|2012-08-21|2010-04-02"),("2016-08-11|2013-08-25|2013-04-01|2012-01-01")).toDF("id") //sample data

df.withColumn("arr_id", array_sort(split(col("id"),"\\|"))). //split to create array and sort the array
withColumn("first_ele", col("arr_id")(0)). // to get first element of array
withColumn("last_ele",reverse(col("arr_id"))(0)). //to get last element of array
withColumn("total_diff",datediff(col("last_ele").cast("date"), col("first_ele").cast("date"))). //get total diff of first and last dates
withColumn("avg_time", col("total_diff")/size(col("arr_id"))). //caluculate diff of date and divide with size of array column
select("id", "total_diff","avg_time"). 
show(false)

Result:

+------------------------------------------------------+----------+--------+
|id                                                    |total_diff|avg_time|
+------------------------------------------------------+----------+--------+
|2018-08-01|2017-06-01|2015-11-01|2012-08-21|2010-04-02|3043      |608.6   |
|2016-08-11|2013-08-25|2013-04-01|2012-01-01           |1684      |421.0   |
+------------------------------------------------------+----------+--------+

Я думаю, что вывод и ответ отключены, потому что start and end dates в массиве немного отличаются в input и output в вопросе!

Answer 2

@ Шу дал потрясающий ответ в Scala.

Это просто версия pyspark. Не стесняйтесь дать ему принятый ответ.

from pyspark.sql import functions as F
df.withColumn("date", F.split(F.regexp_replace("date","\|",","),','))\
.withColumn("start", F.reverse(F.col("date"))[0]).withColumn("end",F.col("date")[0])\
.withColumn("total_time", F.datediff("end","start"))\
.withColumn("Average", F.col("total_time")/(F.size(F.col("date"))))\
.drop("start","end").show()

+--------------------+----------+-------+
|                date|total_time|Average|
+--------------------+----------+-------+
|[2018-08-01, 2017...|      3043|  608.6|
|[2016-08-11, 2013...|      1684|  421.0|
+--------------------+----------+-------+

Answer 3

Вот упрощенная версия других ответов, использующих встроенные функции массива в Spark 2.4 +.

Сначала разбейте строку дат, чтобы получить массив. Теперь, поскольку даты уже отсортированы, используйте функцию element_at, чтобы получить первые и последние даты, поскольку, когда индекс отрицателен, он обращается к элементам массива от последнего к первому. Остальные просто вычисляют datediff и делят его на размер массива, чтобы получить среднее значение:

df.withColumn("array_dates", split(col("date"), "[|]"))\
  .withColumn("total_time", datediff(element_at("array_dates", -1), element_at("array_dates", 1))) \
  .withColumn("average_time", col("total_time") / size(col("array_dates"))) \
  .drop("array_dates")\
  .show(2, False)


#+------------------------------------------------------+----------+------------+
#|date                                                  |total_time|average_time|
#+------------------------------------------------------+----------+------------+
#|2018-08-01|2017-06-01|2015-11-01|2012-08-21|2010-04-02|3043      |608.6       |
#|2016-08-11|2013-08-25|2013-04-01|2012-01-01           |1684      |421.0       |
#+------------------------------------------------------+----------+------------+

Если даты в строке не упорядочены, как в вашем Например, вы можете отсортировать массив после разделения, используя array_sort