title: running sum / cumulative sum with floor and ceiling Py Spark
body: <p>I'm new to spark, and I am trying to calculate a window running sum that is floored by 0 and ceiled by 8</p>
  
  <p>a toy example is given below (note that the actual data is closer to millions of rows):</p>
  
  <pre><code>import pyspark.sql.functions as F
  from pyspark.sql import Window
  import pandas as pd
  from pyspark.sql.functions import pandas_udf, PandasUDFType
  
  pdf = pd.DataFrame({'ids':    [1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3],
                      'day':    [1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4], 
                      'counts': [-3, 3, -6, 3, 3, 6, -3, -6, 3, 3, 3, -3]})
  sdf = spark.createDataFrame(pdf)
  sdf = sdf.orderBy(sdf.ids,sdf.day)
  </code></pre>
  
  <p>This creates the table</p>
  
  <pre><code>+----+---+-------+
  |aIds|day|eCounts|
  +----+---+-------+
  |   1|  1|     -3|
  |   1|  2|      3|
  |   1|  3|     -6|
  |   1|  4|      3|
  |   2|  1|      3|
  |   2|  2|      6|
  |   2|  3|     -3|
  |   2|  4|     -6|
  |   3|  1|      3|
  |   3|  2|      3|
  |   3|  3|      3|
  |   3|  4|     -3|
  +----+---+-------+
  </code></pre>
  
  <p>Below is an example of the result of doing a running sum, and <strong>the expected output runSumCap</strong></p>
  
  <pre><code>+----+---+-------+------+---------+
  |aIds|day|eCounts|runSum|runSumCap|
  +----+---+-------+------+---------+
  |   1|  1|     -3|    -3|        0| &lt;-- reset to 0
  |   1|  2|      3|     0|        3|
  |   1|  3|     -6|    -6|        0| &lt;-- reset to 0
  |   1|  4|      3|    -3|        3|
  |   2|  1|      3|     3|        3|
  |   2|  2|      6|     9|        8| &lt;-- reset to 8
  |   2|  3|     -3|     6|        5| 
  |   2|  4|     -6|     0|        0| &lt;-- reset to 0
  |   3|  1|      3|     3|        3|
  |   3|  2|      3|     6|        6|
  |   3|  3|      3|     9|        8| &lt;-- reset to 8
  |   3|  4|     -3|     6|        5|
  +----+---+-------+------+---------+
  </code></pre>
  
  <p>i know i can calculate the running sum as </p>
  
  <pre><code>partition = Window.partitionBy('aIds').orderBy('aIds','day').rowsBetween(Window.unboundedPreceding, Window.currentRow)`
  sdf1 = sdf.withColumn('runSum',F.sum(sdf.eCounts).over(partition))
  sdf1.orderBy('aIds','day').show()
  </code></pre>
  
  <p>To achieve the expected I have tried looking into @pandas_udf to modify the sum:</p>
  
  <pre><code>@pandas_udf('double', PandasUDFType.GROUPED_AGG)
  def runSumCap(counts):
      #counts columns is passed as a pandas series
      floor = 0
      cap = 8
      runSum = 0
      runSumList = []
      for count in counts.tolist():
        runSum = runSum + count
        if(runSum &gt; cap):
          runSum = 8
        elif(runSum &lt; floor ):
          runSum = 0
        runSumList += [runSum]
      return pd.Series(runSumList)
  
  
  partition = Window.partitionBy('aIds').orderBy('aIds','day').rowsBetween(Window.unboundedPreceding, Window.currentRow)
  sdf1 = sdf.withColumn('runSum',runSumCap(sdf['counts']).over(partition))
  </code></pre>
  
  <p>However this does not work, and it does not seem like the most efficient way to do this.
  How can i make this work? Is there a way to keep it parallel, or do i have to go to pandas dataframes</p>
  
  <p><strong>EDIT:</strong>
  Came with some clarifications about present columns to order the dataset by, and some more insights into what I am trying to achieve</p>
  
  <p><strong>EDIT2:</strong>
  The answer that was provided by @DrChess almost yields the correct result, but the series isn't matching the correct day for some reason:</p>
  
  <pre><code>+----+---+-------+------+
  |aIds|day|eCounts|runSum|
  +----+---+-------+------+
  |   1|  1|     -3|     0|
  |   1|  2|      3|     0|
  |   1|  3|     -6|     3|
  |   1|  4|      3|     3|
  |   2|  1|      3|     3|
  |   2|  2|      6|     8|
  |   2|  3|     -3|     0|
  |   2|  4|     -6|     5|
  |   3|  1|      3|     6|
  |   3|  2|      3|     3|
  |   3|  3|      3|     8|
  |   3|  4|     -3|     5|
  +----+---+-------+------+
  </code></pre>