Получить данные из ближайшего ряда вместо самого раннего

Question

У меня есть следующий фрейм данных:

ptid,blast_date,test_name,result_date,test_result,date_diff
PT085087309 2013-10-03  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-04  Influenza A.H1.respiratory.qualitative  2013-10-04  not detected    0
PT085087309 2013-10-07  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    2
PT085087309 2013-10-09  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    4
PT085087309 2013-10-14  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    9
PT085087309 2013-10-15  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    10
PT085087309 2013-10-18  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    13
PT085087309 2013-10-21  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-23  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-24  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-25  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-27  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-28  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-31  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-01  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-04  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-06  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-08  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-11  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-14  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-11-15  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    0
PT085087309 2013-11-18  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    3
PT085087309 2013-11-19  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    4
PT085087309 2013-11-21  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    6
PT085087309 2014-09-29  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    8
PT085087309 2014-09-30  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    9
PT085087309 2014-10-01  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    10
PT085087309 2014-10-02  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    11
PT085087309 2014-10-03  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    12
PT085087309 2014-10-04  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    13
PT085087309 2014-10-06  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2014-10-07  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2014-10-09  Influenza A.H1.respiratory.qualitative  NA  NA  0

Я хочу, чтобы вышеуказанный фрейм данных проверял предыдущие строки, где когда-либо столбец result и result_date равен NA, и получал данные из предыдущих строк, ближайших к дате. и заполните его вместо NA.

Полученный кадр данных:

ptid,blast_date,test_name,result_date,test_result,date_diff
PT085087309 2013-10-03  Influenza A.H1.respiratory.qualitative  NA  NA  0
PT085087309 2013-10-04  Influenza A.H1.respiratory.qualitative  2013-10-04  not detected    0
PT085087309 2013-10-07  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    2
PT085087309 2013-10-09  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    4
PT085087309 2013-10-14  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    9
PT085087309 2013-10-15  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    10
PT085087309 2013-10-18  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    13
PT085087309 2013-10-21  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-10-23  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-10-24  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-10-25  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-10-27  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-10-28  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-10-31  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-01  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-04  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-06  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-08  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-11  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-14  Influenza A.H1.respiratory.qualitative  2013-10-05  not detected    0
PT085087309 2013-11-15  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    0
PT085087309 2013-11-18  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    3
PT085087309 2013-11-19  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    4
PT085087309 2013-11-21  Influenza A.H1.respiratory.qualitative  2013-11-15  not detected    6
PT085087309 2014-09-29  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    8
PT085087309 2014-09-30  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    9
PT085087309 2014-10-01  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    10
PT085087309 2014-10-02  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    11
PT085087309 2014-10-03  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    12
PT085087309 2014-10-04  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    13
PT085087309 2014-10-06  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    0
PT085087309 2014-10-07  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    0
PT085087309 2014-10-09  Influenza A.H1.respiratory.qualitative  2014-09-21  not detected    0

У меня есть следующий код, но я получаю данные результата с самой ранней даты. Не могли бы вы, ребята, посоветовать мне, как изменить код ниже, чтобы получить данные результата из ближайшей строки даты.

w=Window().partitionBy("ptid","test_name").orderBy("blast_date")

df_tests_filled = df_all_tests.withColumn("collect", f.collect_list(f.array("result_date","test_result")).over(w))\
  .withColumn("collect", f.expr("""filter(collect,x-> array_contains(x,'NA')!=True)""")[0])\
  .withColumn("result_date", f.when((f.col("result_date")=='NA')&(f.col("collect").isNotNull()),f.col("collect")[0]).otherwise(f.col("result_date")))\
  .withColumn("test_result", f.when((f.col("test_result")=='NA')&(f.col("collect").isNotNull()),f.col("collect")[1]).otherwise(f.col("test_result"))).drop("timestamp","collect")

Answer 1

Код, который вы имеете выше, предназначен для другой цели (rolling window), и он не будет работать для this case ( Я точно знаю, потому что я написал этот код ). В этом случае вы должны использовать оконные функции, чтобы получить incremental sum из 1(when date is not null), 0(if it is null), таким образом, мы можем получить желаемые группы, и вы можете заменить, используя first function более new window groups.

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

w=Window().partitionBy("ptid","test_name").orderBy("blast_date")
w2=Window().partitionBy("ptid","test_name","inc_sum").orderBy("blast_date").rowsBetween(Window.unboundedPreceding,Window.unboundedFollowing)
df.withColumn("inc_sum",F.sum(F.when(F.col("result_date")!='NA', F.lit(1))\
                                   .otherwise(F.lit(0))).over(w))\
  .withColumn("result_date", F.when(F.col("result_date")=='NA',F.first("result_date").over(w2))\
                              .otherwise(F.col("result_date")))\
  .withColumn("test_result", F.when(F.col("test_result")=='NA',F.first("test_result").over(w2))\
                              .otherwise(F.col("test_result"))).drop("inc_sum").show()

#+-----------+----------+--------------------+-----------+------------+---------+
#|       ptid|blast_date|           test_name|result_date| test_result|date_diff|
#+-----------+----------+--------------------+-----------+------------+---------+
#|PT085087309|2013-10-03|Influenza A.H1.re...|         NA|          NA|        0|
#|PT085087309|2013-10-04|Influenza A.H1.re...| 2013-10-04|not detected|        0|
#|PT085087309|2013-10-07|Influenza A.H1.re...| 2013-10-05|not detected|        2|
#|PT085087309|2013-10-09|Influenza A.H1.re...| 2013-10-05|not detected|        4|
#|PT085087309|2013-10-14|Influenza A.H1.re...| 2013-10-05|not detected|        9|
#|PT085087309|2013-10-15|Influenza A.H1.re...| 2013-10-05|not detected|       10|
#|PT085087309|2013-10-18|Influenza A.H1.re...| 2013-10-05|not detected|       13|
#|PT085087309|2013-10-21|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-10-23|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-10-24|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-10-25|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-10-27|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-10-28|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-10-31|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-11-01|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-11-04|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-11-06|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-11-08|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-11-11|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#|PT085087309|2013-11-14|Influenza A.H1.re...| 2013-10-05|not detected|        0|
#+-----------+----------+--------------------+-----------+------------+---------+
only showing top 20 rows

Answer 2

Я думаю, это то, что вы ищете:

df['col'] = df['col'].fillna(method='ffill')

https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.fillna.html