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

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

scala> val windowSpec1 = Window.partitionBy($"position").orderBy($"id")
windowSpec1: org.apache.spark.sql.expressions.WindowSpec = org.apache.spark.sql.expressions.WindowSpec@790b51f1

scala> val df1 = df.withColumn("grp", sum(when($"type" === "original", 1).otherwise(0)).over(windowSpec1))
df1: org.apache.spark.sql.DataFrame = [position: int, id: int ... 3 more fields]

scala> val windowSpec2 = Window.partitionBy($"position", $"grp")
windowSpec2: org.apache.spark.sql.expressions.WindowSpec = org.apache.spark.sql.expressions.WindowSpec@72092d90

scala> val df2 = df1.withColumn("prev_o_value", when($"type" === "adjusted", max(when($"type" === "original", $"value").otherwise(null)).over(windowSpec2)).otherwise(null))
df2: org.apache.spark.sql.DataFrame = [position: int, id: int ... 4 more fields]

scala> df2.show
+--------+---+--------+-----+---+------------+
|position| id|    type|value|grp|prev_o_value|
+--------+---+--------+-----+---+------------+
|      10| 10|original|  100|  1|        null|
|      10| 20|original|  200|  2|        null|
|      10| 30|adjusted|  300|  2|         200|
|      10| 40|original|  400|  3|        null|
|      10| 50|adjusted|  500|  3|         400|
|      10| 60|original|  600|  4|        null|
|      10| 70|original|  700|  5|        null|
|      10| 80|adjusted|  800|  5|         700|
|      10| 90|adjusted|  900|  5|         700|
|      10|100|original| 1000|  6|        null|
|      10|110|adjusted| 1100|  6|        1000|
|      10|120|original| 1200|  7|        null|
|      10|130|adjusted| 1300|  7|        1200|
|      10|140|original| 1400|  8|        null|
|      10|150|adjusted| 1500|  8|        1400|
|       2| 10|original|  100|  1|        null|
|       2| 20|original|  200|  2|        null|
|       2| 30|adjusted|  300|  2|         200|
|       2| 40|original|  400|  3|        null|
|       2| 50|adjusted|  500|  3|         400|
+--------+---+--------+-----+---+------------+
only showing top 20 rows

import org.apache.spark.sql.Row
df.repartition($"position").sortWithinPartitions($"id").map{
  var curr_o_value = null.asInstanceOf[Integer];
  var prev_o_value = null.asInstanceOf[Integer];
  {case Row(position: Int, id: Int, xtype: String, value: Int) => {
      prev_o_value = curr_o_value;
      curr_o_value = (if (xtype == "original") (value) else (prev_o_value));
      (position, id, xtype, value, (if (xtype == "adjusted") (prev_o_value) else (null.asInstanceOf[Integer])))
    }
  }
}.show

import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType, IntegerType}

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))

def func(iter: Iterator[Row]) : Iterator[Row] = {
  var res = List[Row]();
  var prev_o_value = null.asInstanceOf[Integer];
  while (iter.hasNext)
  {
    iter.next match { case Row(position: Int, id: Int, xtype: String, value: Int) => 
    res = res ::: List(Row(position, id, xtype, value, if (xtype == "adjusted") (prev_o_value) else (null.asInstanceOf[Integer])));
    prev_o_value = if (xtype == "original") (value) else (prev_o_value);
    }
  }
  res.iterator
}

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