групповой с когда в pyspark

Question

Я хочу манипулировать моим транзакционным фреймом в зависимости от некоторых условий. Мой фактический фрейм данных:

+---+------------+---+-------------------+
| id|install_date|age|           txn_date|
+---+------------+---+-------------------+
|  1|  2019-10-01| 33|2019-09-20 15:27:22|
|  1|  2019-10-01| 33|2019-10-02 14:17:20|
|  1|  2019-10-01| 33|2019-10-07 15:17:12|
|  2|  2019-08-10| 45|2019-06-01 14:07:19|
|  2|  2019-08-10| 45|2019-05-01 15:27:22|
|  3|  2019-09-01| 37|2019-09-20 16:17:20|
|  3|  2019-09-01| 37|2019-10-10 15:27:22|

После этого выполните некоторые манипуляции с ним для достижения этого состояния -

---+------------+---+----------+--------------------+------------------------------------+
| id|install_date|age|  txn_date|app_install_duration|first_txn_duration_after_app_install|
+---+------------+---+----------+--------------------+------------------------------------+
|  1|  2019-10-01| 33|2019-09-20|                  16|                                 -11|
|  1|  2019-10-01| 33|2019-10-02|                  16|                                   1|
|  1|  2019-10-01| 33|2019-10-07|                  16|                                   6|
|  2|  2019-08-10| 45|2019-06-01|                  68|                                 -70|
|  2|  2019-08-10| 45|2019-05-01|                  68|                                -101|
|  3|  2019-09-01| 37|2019-09-20|                  46|                                  19|
|  3|  2019-09-01| 37|2019-10-10|                  46|                                  39|
+---+------------+---+----------+--------------------+------------------------------------+

Теперь я хочу, чтобы мой фрейм данных выглядел следующим образом:

+---+------------+---+----------+--------------------+------------------------------------+-----------
| id|install_date|age|  txn_date|app_install_duration|first_txn_duration_after_app_install|is_active
+---+------------+---+----------+--------------------+------------------------------------+-----------
|  1|  2019-10-01| 33|2019-10-02|                  16|                                   1|    1
|  2|  2019-08-10| 45|2019-06-01|                  68|                                 -70|    0
|  3|  2019-09-01| 37|2019-09-20|                  46|                                  19|    1
+---+------------+---+----------+--------------------+------------------------------------+-----------

Что я сделал до сих пор:

df=spark.createDataFrame([(1,'2019-10-01',33,'2019-09-20 15:27:22'),
                           (1,'2019-10-01',33,'2019-10-02 14:17:20'),
                           (1,'2019-10-01',33,'2019-10-07 15:17:12'),
                         (2,'2019-08-10',45,'2019-06-01 14:07:19'),
                         (2,'2019-08-10',45,'2019-05-01 15:27:22'),
                         (3,'2019-09-01',37,'2019-09-20 16:17:20'),
                         (3,'2019-09-01',37,'2019-10-10 15:27:22')],
                          ['id','install_date','age','txn_date'])    
df = df.withColumn('install_date',to_date(unix_timestamp(F.col('install_date'),'yyyy-MM-dd').cast("timestamp")))
df= df.withColumn('app_install_duration', F.datediff(F.current_date(), df.install_date))

df = df.withColumn('txn_date',to_date(unix_timestamp(F.col('txn_date'),'yyyy-MM-dd HH:mm:ss').cast("timestamp")))

df= df.withColumn('first_txn_duration_after_app_install', F.datediff(df.txn_date, df.install_date))

df.show()