我有一個如下所示的資料框。我想計算 V 上每個 id 的累積總和，這樣當前一行的累積大于或等于 25 的閾值時，累積總和將重置為當前值，如下圖所示。我曾嘗試在 V 上撰寫一個用戶定義的運算子，但我收到一個錯誤，指出它不可迭代。我試過你滯后，但我也沒有成功。我需要幫助！
df = sqlContext.createDataFrame(
[('Mark', 0.0), ('Mark', 1), ('Mark', 1),
('Mark', 1), ('Mark', 25), ('Mark', 1),
('Mark', 1),('Mark', 1),('Mark', 20),
('Mark', 1),('Mark', 1),('Mark', 1),
('Mark', 1),('Mark', 1),('John', 0),
('John', 1),('John', 1),('John', 1),
('John', 1),('John', 1),('John', 1),
('John', 1),('John', 9),('John', 1),
('John', 1),('John', 1),('John', 1),
('John', 1),('John', 1),('John', 1),
('John', 1),('John', 1),('John', 1),
('John', 7),('John', 1)],
('id', "V"))
uj5u.com熱心網友回復：
也許有更漂亮更快的方法，這會起作用但效率不高。Windows 使用起來很昂貴（記憶體）。如果您打算在生產中使用它，請務必小心。如果需要速度，與視窗一起使用的自定義 udf 可能會比這快一點以避免雙視窗：
val df = Seq(("Mark", 0), ("Mark", 1), ("Mark", 1),
("Mark", 1), ("Mark", 25), ("Mark", 1),
("Mark", 1),("Mark", 1),("Mark", 20),
("Mark", 1),("Mark", 1),("Mark", 1),
("Mark", 1),("Mark", 1),("John", 0),
("John", 1),("John", 1),("John", 1),
("John", 1),("John", 1),("John", 1),
("John", 1),("John", 9),("John", 1),
("John", 1),("John", 1),("John", 1),
("John", 1),("John", 1),("John", 1),
("John", 1),("John", 1),("John", 1),
("John", 7),("John", 1)).toDF("id","V")
val windowSpecLag = Window.partitionBy("id").orderBy("id")
val windowSpec = Window.partitionBy("id").orderBy("id").rowsBetween(Window.unboundedPreceding,Window.currentRow )
// add a running sum to the window
val divis = df.withColumn("sum",sum("V").over(windowSpec)).withColumn("divis",floor(col("sum")/25))
// shift around the numbers so the math works as desired
val lagged = divis.withColumn("clag", (lag("divis", 1, 0) over windowSpecLag) )
//re-run running total on newly partitioned data
val windowSpecFixed = Window.partitionBy("id","clag").orderBy("id","clag").rowsBetween(Window.unboundedPreceding,Window.currentRow )
lagged.withColumn("runningTotalUnder25",sum("V").over(windowSpecDivis)).show(100)
如果你想有效地做到這一點，我可能會嘗試重新定義問題，以便我可以使用 group by。或者改變資料的定義方式。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qianduan/333880.html