спарк дел расчет на столбце на основе старых значений предыдущего ряда

Question

У меня есть данный фрейм данных, как показано ниже

+----------------+---------------+----------+------------------+-------------+
|Transaction_date|    Added  date|coupon_id |cart_value        | coupon_value|
+----------------+---------------+----------+------------------+-------------+
|2018-01-16      |2018-02-01     |2390324796|12.5              |1.8          |
|2018-01-16      |2018-01-04     |1100111212|1.0               |2.0          |
|2018-01-19      |2018-01-04     |1100111212|2.5               |2.0          |
+----------------+---------------+----------+------------------+-------------+

Мне нужно применить значение купона к значению корзины и обновить остаток купона и значение автоматического погашения, что я хочу делать только тогда, когда «Transaction_date» больше, чем «Added Date» значения купона

Логика

updatedBalance = (coupon_value - cart_value), если cart_value больше, активировать только доступное значение купона.

Погашено = Сколько выкуплено в данной транзакции

Я хочу что-то вроде этого

+----------------+---------------+----------+------------------+-------------+-------------+-------------+
|Transaction_date|    Added  date|coupon_id |cart_value        | coupon_value|UpdatedBalance|Redeemed      |
+----------------+---------------+----------+------------------+-------------+-------------+-------------+
|2018-01-16      |2018-02-01     |2390324796|12.5              |1.8          |0            |0            |
|2018-01-16      |2018-01-04     |1100111212|1.0               |2.0          |1            |1            |
|2018-01-19      |2018-01-04     |1100111212|2.5               |2.0          |0            |1            |
+----------------+---------------+----------+------------------+-------------+-------------+-------------+

Я пытаюсь сделать это в spark-scala

Answer 1

При условии, что раздел находится на всей таблице и упорядочен по убыванию add_date, ниже будет работать

scala> val df =Seq(("2018-01-16","2018-02-01",2390324796L,12.5,1.8),("2018-01-16","2018-01-04",1100111212L,1.0,2.0),("2018-01-19","2018-01-04",1100111212L,2.5,2.0)).toDF("Transaction_date","Added_date","coupon_id","cart_value","coupon_value")
df: org.apache.spark.sql.DataFrame = [Transaction_date: string, Added_date: string ... 3 more fields]

scala> df.show(false)
+----------------+----------+----------+----------+------------+
|Transaction_date|Added_date|coupon_id |cart_value|coupon_value|
+----------------+----------+----------+----------+------------+
|2018-01-16      |2018-02-01|2390324796|12.5      |1.8         |
|2018-01-16      |2018-01-04|1100111212|1.0       |2.0         |
|2018-01-19      |2018-01-04|1100111212|2.5       |2.0         |
+----------------+----------+----------+----------+------------+


scala> val df2 = df.withColumn("UpdatedBalance",when('coupon_value>'cart_value,'coupon_value-'cart_value).otherwise(0))
df2: org.apache.spark.sql.DataFrame = [Transaction_date: string, Added_date: string ... 4 more fields]

scala> df2.show(false)
+----------------+----------+----------+----------+------------+--------------+
|Transaction_date|Added_date|coupon_id |cart_value|coupon_value|UpdatedBalance|
+----------------+----------+----------+----------+------------+--------------+
|2018-01-16      |2018-02-01|2390324796|12.5      |1.8         |0.0           |
|2018-01-16      |2018-01-04|1100111212|1.0       |2.0         |1.0           |
|2018-01-19      |2018-01-04|1100111212|2.5       |2.0         |0.0           |
+----------------+----------+----------+----------+------------+--------------+

scala> import org.apache.spark.sql.expressions._
import org.apache.spark.sql.expressions._


scala> df2.withColumn("Redeemed",sum('UpdatedBalance).over(Window.orderBy('Added_date.desc))).show(false)
19/01/03 10:31:50 WARN window.WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
+----------------+----------+----------+----------+------------+--------------+--------+
|Transaction_date|Added_date|coupon_id |cart_value|coupon_value|UpdatedBalance|Redeemed|
+----------------+----------+----------+----------+------------+--------------+--------+
|2018-01-16      |2018-02-01|2390324796|12.5      |1.8         |0.0           |0.0     |
|2018-01-16      |2018-01-04|1100111212|1.0       |2.0         |1.0           |1.0     |
|2018-01-19      |2018-01-04|1100111212|2.5       |2.0         |0.0           |1.0     |
+----------------+----------+----------+----------+------------+--------------+--------+


scala>