title: Spark window function on dataframe with large number of columns
body: <p>I have an ML dataframe which I read from csv files. It contains three types of columns:</p>
  
  <blockquote>
    <p>ID Timestamp Feature1 Feature2...Feature_n</p>
  </blockquote>
  
  <p>where n is ~ 500 (500 features in ML parlance). The total number of rows in the dataset is ~ 160 millions. </p>
  
  <p>As this is the result of a previous full join, there are many features which do not have values set.</p>
  
  <p>My aim is to run a "fill" function(fillna style form python pandas), where each empty feature value gets set with the previously available value for that column, per Id and Date.</p>
  
  <p>I am trying to achieve this with the following spark 2.2.1 code:</p>
  
  <pre><code> val rawDataset = sparkSession.read.option("header", "true").csv(inputLocation)
  
   val window = Window.partitionBy("ID").orderBy("DATE").rowsBetween(-50000, -1)
  
   val columns = Array(...) //first 30 columns initially, just to see it working
  
  val rawDataSetFilled = columns.foldLeft(rawDataset) { (originalDF, columnToFill) =&gt;
        originalDF.withColumn(columnToFill, coalesce(col(columnToFill), last(col(columnToFill), ignoreNulls = true).over(window)))
      }
  </code></pre>
  
  <p>I am running this job on a 4 m4.large instances on Amazon EMR, with spark 2.2.1. and dynamic allocation enabled. </p>
  
  <p>The job runs for over 2h without completing.</p>
  
  <p>Am I doing something wrong, at the code level? Given the size of the data, and the instances, I would assume it should finish in a reasonable amount of time? And I haven't even tried with the full 500 columns, just with about 30!</p>
  
  <p>Looking in the container logs, all I see are many logs like this:</p>
  
  <blockquote>
    <p>INFO codegen.CodeGenerator: Code generated in 166.677493 ms  </p>
    
    <p>INFO execution.ExternalAppendOnlyUnsafeRowArray: Reached spill
    threshold of
    4096 rows, switching to
    org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter</p>
  </blockquote>
  
  <p>I have tried setting parameter <strong>spark.sql.windowExec.buffer.spill.threshold</strong> to something larger, without any impact. Is theresome other setting I should know about? Those 2 lines are the only ones I see in any container log. </p>
  
  <p>In Ganglia, I see most of the CPU cores peaking around full usage, but the memory usage is lower than the maximum available. All executors are allocated and are doing work.</p>