scala> spark.createDataFrame(df.repartition($"position").sortWithinPartitions($"id").rdd.mapPartitions(func, true), schema).show
+--------+---+--------+-----+------------+-------+
|position| id|    type|value|prev_o_value|count_o|
+--------+---+--------+-----+------------+-------+
|       1| 10|original|  100|        null|   null|
|       1| 20|original|  200|        null|   null|
|       1| 30|adjusted|  300|         200|      2|
|       1| 40|original|  400|        null|   null|
|       1| 50|adjusted|  500|         400|      3|
|       1| 60|original|  600|        null|   null|
|       1| 70|original|  700|        null|   null|
|       1| 80|adjusted|  800|         700|      5|
|       1| 90|adjusted|  900|         700|      5|
|       1|100|original| 1000|        null|   null|
|       1|110|adjusted| 1100|        1000|      2|
|       1|120|original| 1200|        null|   null|
|       1|130|adjusted| 1300|        1200|      1|
|       1|140|original| 1400|        null|   null|
|       1|150|adjusted| 1500|        1400|      1|
|      10| 10|original|  100|        null|   null|
|      10| 20|original|  200|        null|   null|
|      10| 30|adjusted|  300|         200|      2|
|      10| 40|original|  400|        null|   null|
|      10| 50|adjusted|  500|         400|      3|
+--------+---+--------+-----+------------+-------+
only showing top 20 rows

scala> val li = List((1,1),(1,2),(1,3),(1,5))
li: List[(Int, Int)] = List((1,1), (1,2), (1,3), (1,5))

scala> li.toDF.show
+---+---+
| _1| _2|
+---+---+
|  1|  1|
|  1|  2|
|  1|  3|
|  1|  5|
+---+---+


scala> val it = li.toIterator
it: Iterator[(Int, Int)] = non-empty iterator

scala> it.scanLeft((0,0))((x1,x2) => (x2._1, x1._2 + x2._2)).toArray
res1: Array[(Int, Int)] = Array((0,0), (1,1), (1,3), (1,6), (1,11))

scala> val it = li.toIterator
it: Iterator[(Int, Int)] = non-empty iterator

scala> it.scanLeft((0,0,0))((x1,x2) => (x2._1, x2._2, x1._3 + x2._2)).toArray
res2: Array[(Int, Int, Int)] = Array((0,0,0), (1,1,1), (1,2,3), (1,3,6), (1,5,11))

val spark = org.apache.spark.sql.SparkSession.builder().config("spark.sql.parquet.binaryAsString","true").getOrCreate()
val df = spark.read.load("hdfs://.../t")

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType, IntegerType}
import scala.collection.mutable.Queue

val schema = new StructType().add(
             StructField("position", IntegerType, true)).add(
             StructField("id", IntegerType, true)).add(
             StructField("type", StringType, true)).add(
             StructField("value", IntegerType, true)).add(
             StructField("prev_o_value", IntegerType, true)).add(
             StructField("count_o", IntegerType, true))

def func(iter: Iterator[Row]) : Iterator[Row] = {
  val limit = 2000;
  var q = new Queue[Int];
  var total = 0;
  iter.scanLeft(Row(0, 0, "_", 0, 0, 0))
  {
    case (x1, x2) =>
    val prev_o_value = if (x1.getString(2) == "original") (x1.getInt(3)) else (x1.getInt(4));
    
    if (x2.getString(2) == "original") {
      q += x2.getInt(3);
      total += x2.getInt(3); 
    } else (
      while (total > limit){total -= q.dequeue;}
    );

    Row(x2.getInt(0), x2.getInt(1), x2.getString(2), x2.getInt(3),
        if (x2.getString(2) == "adjusted") (prev_o_value) else (null.asInstanceOf[Integer]),
        if (x2.getString(2) == "adjusted") (q.size) else (null.asInstanceOf[Integer]));
  }.drop(1)
}

spark.createDataFrame(df.repartition($"position").sortWithinPartitions($"id").rdd.mapPartitions(func, true), schema).show