Как рассчитать количество дней перекрытия во фрейме данных в PySpark? - PullRequest
Новая
       0

Как рассчитать количество дней перекрытия во фрейме данных в PySpark?

1 голос
/ 28 мая 2020

Мне нужно рассчитать дни перекрытия по строкам во фрейме данных. Данные выглядят так: 

+-------+-------------------+-------------------+------------------+
|id|             begin|                end|              days|
+-------+-------------------+-------------------+------------------+
|1|2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778|
|1|2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|
|1|2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|
|1|2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889|
|1|2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|
|1|2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|
|1|2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|
|1|2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|
|1|2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444|
+-------+-------------------+-------------------+------------------+

Здесь первая запись охватывает 2019 год (365 дней). Все остальные записи совпадают с первой записью. Я хочу, чтобы функция подсчитывала общее количество дней, что составляет 365 дней в наборе данных после удаления перекрывающихся дней.

Я действительно решил эту проблему в R, но я не могу запускать циклы в PySpark.

Я ищу такой результат. 

+-------+-------------------+-------------------+------------------+------------------+
|     id|              begin|                end|              days|           overlap|
+-------+-------------------+-------------------+------------------+------------------+
|1      |2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|            0|
|1      |2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778| 7.090277777777778|
|1      |2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|           27.1375|
|1      |2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|29.584027777777777|
|1      |2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444| 47.96944444444444|
|1      |2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|10.430555555555555|
|1      |2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|14.472222222222221|
|1      |2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889| 19.24513888888889|
|1      |2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|209.07083333333333|
+-------+-------------------+-------------------+------------------+------------------+

Даты не в порядке, и есть сценарий ios, где нет перекрытия.

Сценарий 2: Нет перекрытия 

+-------+-------------------+-------------------+-----+-----+
|  id   |              begin|                end| days| over|
+-------+-------------------+-------------------+-----+-----+
|2      |2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|    0|
|2      |2019-12-25 00:00:00|2020-01-01 00:00:00|  7.0|    0|
+-------+-------------------+-------------------+-----+-----+

Сценарий 3: Частичное перекрытие 

+-------+-------------------+-------------------+-----+-----+
|     id|              begin|                end| days| over|
+-------+-------------------+-------------------+-----+-----+
|3      |2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|    0|
|3      |2019-12-20 00:00:00|2020-01-01 00:00:00| 12.0|    5|
+-------+-------------------+-------------------+-----+-----+

Сценарий 4: Более сложный Здесь первая запись охватывает первые 358 дней 2019 года. Вторая запись полностью перекрывается с первой. и поэтому все дни еще не закончились. Третья запись не перекрывается со второй записью, но перекрывает 5 дней с первой и, следовательно, 5 дней в столбце «больше». 

+-------+-------------------+-------------------+-----+-----+
|     id|              begin|                end| days| over|
+-------+-------------------+-------------------+-----+-----+
|4      |2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|    0|
|4      |2019-01-01 00:00:00|2019-11-25 00:00:00|328.0|328.0|
|4      |2019-12-20 00:00:00|2020-01-01 00:00:00| 12.0|    5|
+-------+-------------------+-------------------+-----+-----+

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

В R я создал еще один столбец под названием «перекрытие» и использовал функцию перекрытия для l oop чтобы сравнить все значения с другими.

Функция R, которая производит желаемый результат: 

abc<-data.frame()
for (i in id) {
  xyz<- dataset %>% filter(id==i) %>% arrange(begin)

  for(j in 1:(nrow(xyz)-1)){
    k=j
    while(k<nrow(xyz)){
      xyz$overlap[j]<- xyz$overlap[j] + Overlap(c(xyz$begin[j], xyz$end[j]), c(xyz$begin[k+1], xyz$end[k+1])) 
      k=k+1
    }

  }
  abc<- bind_rows(abc,xyz)
}

Я все еще изучаю pyspark, и мне нужна помощь с этим.

Ответ на фрагмент кода от @ murtiha sh

Привет, это похоже на ответ, но все же не тот результат, который я ищу. Вывод вашего кода 

+-------+-------------------+-------------------+-----------------+-------+
|     id|              begin|                end|             days|overlap|
+-------+-------------------+-------------------+-----------------+-------+
|7777777|2019-01-05 01:00:00|2019-04-04 00:00:00|88.95833333333333|      0|
|7777777|2019-04-04 00:00:00|2019-07-11 00:00:00|             98.0|      0|
|7777777|2019-07-11 00:00:00|2019-09-17 00:00:00|             68.0|      1|
|7777777|2019-09-17 00:00:00|2019-09-19 22:01:00|2.917361111111111|      0|
|7777777|2019-09-19 22:01:00|2020-01-01 00:00:00|103.0826388888889|     -1|
|7777777|2019-09-19 22:01:00|2020-01-01 00:00:00|103.0826388888889|     -1|
+-------+-------------------+-------------------+-----------------+-------+

Желаемый результат должен быть: 

+-------+-------------------+-------------------+-----------------+-------+
|     id|              begin|                end|             days|overlap|
+-------+-------------------+-------------------+-----------------+-------+
|7777777|2019-01-05 01:00:00|2019-04-04 00:00:00|88.95833333333333|      0|
|7777777|2019-04-04 00:00:00|2019-07-11 00:00:00|             98.0|      0|
|7777777|2019-07-11 00:00:00|2019-09-17 00:00:00|             68.0|      0|
|7777777|2019-09-17 00:00:00|2019-09-19 22:01:00|2.917361111111111|      0|
|7777777|2019-09-19 22:01:00|2020-01-01 00:00:00|103.0826388888889|103.082|
|7777777|2019-09-19 22:01:00|2020-01-01 00:00:00|103.0826388888889|      0|
+-------+-------------------+-------------------+-----------------+-------+

Объяснение: Первые четыре строки не перекрываются. Пятая и шестая строки имеют точно такой же период (и не перекрываются с другими строками), поэтому для одной из 5-й или 6-й строки перекрытие должно составлять 103,08 дня.

Обновление: не работает с этим конкретным сценарий. Вывод фрагмента кода @ murtiha sh

+-------+-------------------+-------------------+------------------+-------+
|  imono|              begin|                end|              days|overlap|
+-------+-------------------+-------------------+------------------+-------+
|9347774|2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778|    0.0|
|9347774|2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|    0.0|
|9347774|2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|    0.0|
|9347774|2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444|    0.0|
|9347774|2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|    0.0|
|9347774|2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|    0.0|
|9347774|2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889|    0.0|
|9347774|2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|    7.0|
|9347774|2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|    0.0|
+-------+-------------------+-------------------+------------------+-------+

Желаемый результат: Это 

 +-------+-------------------+-------------------+------------------+-------+
    |  imono|              begin|                end|              days|overlap|
    +-------+-------------------+-------------------+------------------+-------+
    |9347774|2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778|    0.0|
    |9347774|2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|    0.0|
    |9347774|2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|    0.0|
    |9347774|2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444|    0.0|
    |9347774|2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|    0.0|
    |9347774|2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|    0.0|
    |9347774|2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889|    0.0|
    |9347774|2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|    365|
    |9347774|2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|    0.0|
    +-------+-------------------+-------------------+------------------+-------+

или 

 +-------+-------------------+-------------------+------------------+-------+
|  imono|              begin|                end|              days|overlap|

+-------+-------------------+-------------------+------------------+-------+
|9347774|2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778|    7.1|
|9347774|2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|   27.1|
|9347774|2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|   29.5|
|9347774|2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444|   48.0|
|9347774|2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|   10.4|
|9347774|2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|   14.5|
|9347774|2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889|   19.2|
|9347774|2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|    0.0|
|9347774|2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|  209.1|
+-------+-------------------+-------------------+------------------+-------+

Пояснение: вторая последняя запись охватывает весь год и все другие записи совпадают с этим. Таким образом, либо получается, что вторая последняя запись перекрывается = 365, либо все другие записи имеют свои дни как перекрытие, а вторая последняя запись имеет перекрытие 0 дней.

Update2: не работает в этом конкретном сценарии. Вывод из фрагмента кода @ murtiha sh (Update2) 

+-------+-------------------+-------------------+------------------+-------+
|  imono|              begin|                end|              days|overlap|
+-------+-------------------+-------------------+------------------+-------+
|9395123|2019-01-19 05:01:00|2019-02-06 00:00:00|17.790972222222223|   17.0|
|9395123|2019-02-06 00:00:00|2019-06-17 00:00:00|             131.0|    0.0|
|9395123|2019-01-19 05:01:00|2020-01-01 00:00:00| 346.7909722222222|    0.0|
|9395123|2019-06-17 00:00:00|2020-01-01 00:00:00|             198.0|    0.0|
+-------+-------------------+-------------------+------------------+-------+

Желаемый результат: 

+-------+-------------------+-------------------+------------------+-------+
|  id   |              begin|                end|              days|overlap|
+-------+-------------------+-------------------+------------------+-------+
|8888888|2019-01-19 05:01:00|2019-02-06 00:00:00|17.790972222222223|   17.8|
|8888888|2019-02-06 00:00:00|2019-06-17 00:00:00|             131.0|    0.0|
|8888888|2019-01-19 05:01:00|2020-01-01 00:00:00| 346.7909722222222|    329|
|8888888|2019-06-17 00:00:00|2020-01-01 00:00:00|             198.0|    0.0|
+-------+-------------------+-------------------+------------------+-------+

Я действительно не понимаю, что делает ваш фрагмент кода, и поэтому я не могу чтобы настроить его для моей цели. Спасибо за помощь!

Ответы [ 2 ]

1 голос
/ 29 мая 2020

Для Spark2.4+ вы можете использовать sequence ( сгенерировать диапазон дат ), collect_list, и использовать сочетание функций массива и функций более высокого порядка для получения желаемых перекрытий. 

df.show() #sample dataframe
#+---+-------------------+-------------------+-----+
#| id|              begin|                end| days|
#+---+-------------------+-------------------+-----+
#|  2|2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|
#|  2|2019-12-25 00:00:00|2020-01-01 00:00:00|  7.0|
#|  3|2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|
#|  3|2019-12-20 00:00:00|2020-01-01 00:00:00| 12.0|
#|  4|2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|
#|  4|2019-01-01 00:00:00|2019-11-25 00:00:00|328.0|
#|  4|2019-12-20 00:00:00|2020-01-01 00:00:00| 12.0|
#+---+-------------------+-------------------+-----+

from pyspark.sql import functions as F
from pyspark.sql.window import Window

w1=Window().partitionBy("id").orderBy("begin")
df.withColumn("seq", F.expr("""sequence(to_timestamp(begin), to_timestamp(end),interval 1 day)"""))\
  .withColumn("seq1", F.expr("""flatten(filter(collect_list(seq) over\
                                (partition by id),x-> arrays_overlap(x,seq)==True and seq!=x))"""))\
  .withColumn("overlap", F.when(F.row_number().over(w1)==1, F.lit(0))\
              .otherwise(F.size(F.array_intersect("seq","seq1"))-1)).orderBy("id","end").drop("seq","seq1").show()

#+---+-------------------+-------------------+-----+-------+
#| id|              begin|                end| days|overlap|
#+---+-------------------+-------------------+-----+-------+
#|  2|2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|      0|
#|  2|2019-12-25 00:00:00|2020-01-01 00:00:00|  7.0|      0|
#|  3|2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|      0|
#|  3|2019-12-20 00:00:00|2020-01-01 00:00:00| 12.0|      5|
#|  4|2019-01-01 00:00:00|2019-11-25 00:00:00|328.0|    328|
#|  4|2019-01-01 00:00:00|2019-12-25 00:00:00|358.0|      0|
#|  4|2019-12-20 00:00:00|2020-01-01 00:00:00| 12.0|      5|
#+---+-------------------+-------------------+-----+-------+

UPDATE:

Это должно охватывать все случаи: 

from pyspark.sql import functions as F
from pyspark.sql.window import Window

w1=Window().partitionBy("id").orderBy("begin")
w2=Window().partitionBy("id","begin","end").orderBy("begin")
w3=Window().partitionBy("id","begin","end")
w4=Window().partitionBy("id","begin","end","maxrownum").orderBy("begin")
df.withColumn("seq", F.expr("""sequence(to_timestamp(begin), to_timestamp(end),interval 1 day)"""))\
  .withColumn('maxrownum', F.max(F.row_number().over(w2)).over(w3))\
  .withColumn('rowNum', F.row_number().over(w4))\
  .withColumn("seq1", F.expr("""flatten(filter(collect_list(seq) over\
                                (partition by id order by begin),x-> arrays_overlap(x,seq)==True and seq!=x))"""))\
  .withColumn("overlap", F.when(F.row_number().over(w1)==1, F.lit(0))\
              .when(F.size(F.array_intersect("seq","seq1"))!=0,F.size(F.array_intersect("seq","seq1"))-1)
              .when((F.col("maxrownum")!=1)&(F.col("rowNum")<F.col("maxrownum")),F.col("days"))\
              .otherwise(F.lit(0)))\
         .orderBy("id","end").drop("seq","seq1","maxrownum","rowNum").show()

#+-------+-------------------+-------------------+-----------------+-----------------+
#|     id|              begin|                end|             days|          overlap|
#+-------+-------------------+-------------------+-----------------+-----------------+
#|7777777|2019-01-05 01:00:00|2019-04-04 00:00:00|88.95833333333333|              0.0|
#|7777777|2019-04-04 00:00:00|2019-07-11 00:00:00|             98.0|              0.0|
#|7777777|2019-07-11 00:00:00|2019-09-17 00:00:00|             68.0|              0.0|
#|7777777|2019-09-17 00:00:00|2019-09-19 22:01:00|2.917361111111111|              0.0|
#|7777777|2019-09-19 22:01:00|2020-01-01 00:00:00|103.0826388888889|103.0826388888889|
#|7777777|2019-09-19 22:01:00|2020-01-01 00:00:00|103.0826388888889|              0.0|
#+-------+-------------------+-------------------+-----------------+-----------------+

UPDATE2:

from pyspark.sql import functions as F
from pyspark.sql.window import Window

w1=Window().partitionBy("id").orderBy("begin")
w2=Window().partitionBy("id","begin","end").orderBy("begin")
w3=Window().partitionBy("id","begin","end")
w4=Window().partitionBy("id","begin","end","maxrownum").orderBy("begin")
df.withColumn("seq", F.expr("""sequence(to_timestamp(begin), to_timestamp(end),interval 1 day)"""))\
  .withColumn('maxrownum', F.max(F.row_number().over(w2)).over(w3))\
  .withColumn('rowNum', F.row_number().over(w4))\
  .withColumn("seq1", F.expr("""flatten(filter(collect_list(seq) over\
                                (partition by id),x-> arrays_overlap(x,seq)==True and seq!=x))"""))\
  .withColumn("overlap", F.when(F.row_number().over(w1)==1, F.lit(0))\
              .when(F.size(F.array_intersect("seq","seq1"))!=0,F.size(F.array_intersect("seq","seq1"))-1)
              .when((F.col("maxrownum")!=1)&(F.col("rowNum")<F.col("maxrownum")),F.col("days"))\
              .when(F.col("maxrownum")==1,F.col("days"))\
              .otherwise(F.lit(0)))\
              .replace(1,0)\
         .orderBy("id","end").drop("seq","seq1","rowNum","maxrownum").show()

#+-------+-------------------+-------------------+------------------+------------------+
#|     id|              begin|                end|              days|           overlap|
#+-------+-------------------+-------------------+------------------+------------------+
#|9347774|2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778|               7.0|
#|9347774|2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|           27.1375|
#|9347774|2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|29.584027777777777|
#|9347774|2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444| 47.96944444444444|
#|9347774|2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|10.430555555555555|
#|9347774|2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|14.472222222222221|
#|9347774|2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889| 19.24513888888889|
#|9347774|2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|               0.0|
#|9347774|2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|209.07083333333333|
#+-------+-------------------+-------------------+------------------+------------------+
0 голосов
/ 28 мая 2020

Проверьте решение ниже, посмотрите, работает ли оно для вас. 

import pyspark.sql.functions as F

df = sc.parallelize([["1","2019-01-01 00:00:00","2019-01-08 02:10:00","7.090277777777778"],
["1","2019-02-04 05:28:00","2019-03-05 19:29:00","29.584027777777777"],
["1","2019-06-05 22:18:00","2020-01-01 00:00:00","209.07083333333333"],
["1","2019-05-17 16:25:00","2019-06-05 22:18:00","19.24513888888889"],
["1","2019-05-03 05:05:00","2019-05-17 16:25:00","14.472222222222221"],
["1","2019-01-08 02:10:00","2019-02-04 05:28:00","27.1375"],
["1","2019-01-01 00:00:00","2020-01-01 00:00:00","365.0"],
["1","2019-04-22 18:45:00","2019-05-03 05:05:00","10.430555555555555"],
["1","2019-03-05 19:29:00","2019-04-22 18:45:00","47.96944444444444"]]).toDF(("id","begin","end","days"))

df.withColumn("overlap", ((F.unix_timestamp(col("end")).cast("long") - F.unix_timestamp(col("begin")).cast("long"))/(24*3600))).show()

+---+-------------------+-------------------+------------------+------------------+
| id|              begin|                end|              days|           overlap|
+---+-------------------+-------------------+------------------+------------------+
|  1|2019-01-01 00:00:00|2019-01-08 02:10:00| 7.090277777777778| 7.090277777777778|
|  1|2019-02-04 05:28:00|2019-03-05 19:29:00|29.584027777777777|29.584027777777777|
|  1|2019-06-05 22:18:00|2020-01-01 00:00:00|209.07083333333333|          209.1125|
|  1|2019-05-17 16:25:00|2019-06-05 22:18:00| 19.24513888888889| 19.24513888888889|
|  1|2019-05-03 05:05:00|2019-05-17 16:25:00|14.472222222222221|14.472222222222221|
|  1|2019-01-08 02:10:00|2019-02-04 05:28:00|           27.1375|           27.1375|
|  1|2019-01-01 00:00:00|2020-01-01 00:00:00|             365.0|             365.0|
|  1|2019-04-22 18:45:00|2019-05-03 05:05:00|10.430555555555555|10.430555555555555|
|  1|2019-03-05 19:29:00|2019-04-22 18:45:00| 47.96944444444444| 47.92777777777778|
+---+-------------------+-------------------+------------------+------------------+
...