title: spark sql conditional maximum
body: <p>I have a tall table which contains up to 10 values per group. How can I transform this table into a wide format i.e. add 2 columns where these resemble the value smaller or equal to a threshold?</p>
  
  <p>I want to find the maximum per group, but it needs to be smaller than a specified value like:</p>
  
  <pre><code>min(max('value1), lit(5)).over(Window.partitionBy('grouping))
  </code></pre>
  
  <p>However <code>min()</code>will only work for a column and not for the Scala value which is returned from the inner function?</p>
  
  <p>The problem can be described as:</p>
  
  <pre><code>Seq(Seq(1,2,3,4).max,5).min
  </code></pre>
  
  <p>Where <code>Seq(1,2,3,4)</code> is returned by the window.</p>
  
  <p>How can I formulate this in spark sql?</p>
  
  <h1>edit</h1>
  
  <p>E.g.</p>
  
  <pre><code>+--------+-----+---------+
  |grouping|value|something|
  +--------+-----+---------+
  |       1|    1|    first|
  |       1|    2|   second|
  |       1|    3|    third|
  |       1|    4|   fourth|
  |       1|    7|        7|
  |       1|   10|       10|
  |      21|    1|    first|
  |      21|    2|   second|
  |      21|    3|    third|
  +--------+-----+---------+
  </code></pre>
  
  <p>created by</p>
  
  <pre><code>case class MyThing(grouping: Int, value:Int, something:String)
  val df = Seq(MyThing(1,1, "first"), MyThing(1,2, "second"), MyThing(1,3, "third"),MyThing(1,4, "fourth"),MyThing(1,7, "7"), MyThing(1,10, "10"),
  MyThing(21,1, "first"), MyThing(21,2, "second"), MyThing(21,3, "third")).toDS
  </code></pre>
  
  <p>Where</p>
  
  <pre><code>df
  .withColumn("somethingAtLeast5AndMaximum5", max('value).over(Window.partitionBy('grouping)))
  .withColumn("somethingAtLeast6OupToThereshold2", max('value).over(Window.partitionBy('grouping)))
  .show
  </code></pre>
  
  <p>returns</p>
  
  <pre><code>+--------+-----+---------+----------------------------+-------------------------+
  |grouping|value|something|somethingAtLeast5AndMaximum5| somethingAtLeast6OupToThereshold2 |
  +--------+-----+---------+----------------------------+-------------------------+
  |       1|    1|    first|                          10|                       10|
  |       1|    2|   second|                          10|                       10|
  |       1|    3|    third|                          10|                       10|
  |       1|    4|   fourth|                          10|                       10|
  |       1|    7|        7|                          10|                       10|
  |       1|   10|       10|                          10|                       10|
  |      21|    1|    first|                           3|                        3|
  |      21|    2|   second|                           3|                        3|
  |      21|    3|    third|                           3|                        3|
  +--------+-----+---------+----------------------------+-------------------------+
  </code></pre>
  
  <p>Instead, I rather would want to formulate:</p>
  
  <pre><code>lit(Seq(max('value).asInstanceOf[java.lang.Integer], new java.lang.Integer(2)).min).over(Window.partitionBy('grouping))
  </code></pre>
  
  <p>But that does not work as <code>max('value)</code> is not a scalar value.</p>
  
  <p>Expected output should look like</p>
  
  <pre><code>+--------+-----+---------+----------------------------+-------------------------+
  |grouping|value|something|somethingAtLeast5AndMaximum5|somethingAtLeast6OupToThereshold2|
  +--------+-----+---------+----------------------------+-------------------------+
  |       1|    4|   fourth|                           4|                        7|
  |      21|    1|    first|                           3|                     NULL|
  +--------+-----+---------+----------------------------+-------------------------+
  </code></pre>
  
  <h2>edit2</h2>
  
  <p>When trying a pivot</p>
  
  <pre><code>df.groupBy("grouping").pivot("value").agg(first('something)).show
  +--------+-----+------+-----+------+----+----+
  |grouping|    1|     2|    3|     4|   7|  10|
  +--------+-----+------+-----+------+----+----+
  |       1|first|second|third|fourth|   7|  10|
  |      21|first|second|third|  null|null|null|
  +--------+-----+------+-----+------+----+----+
  </code></pre>
  
  <p>The second part of the problem remains that some columns might not exist or be null.</p>
  
  <p>When aggregating to arrays:</p>
  
  <pre><code>df.groupBy("grouping").agg(collect_list('value).alias("value"), collect_list('something).alias("something"))
  +--------+-------------------+--------------------+
  |grouping|              value|           something|
  +--------+-------------------+--------------------+
  |       1|[1, 2, 3, 4, 7, 10]|[first, second, t...|
  |      21|          [1, 2, 3]|[first, second, t...|
  +--------+-------------------+--------------------+
  </code></pre>
  
  <p>The values are already next to each other, but the right values need to be selected. This is probably still more efficient than a join or window function.</p>