Convert GroupBy Object to Ordered List in Pyspark
<p>I'm using Spark 2.0.0 and dataframe.
Here is my input dataframe as</p>

<pre><code>| id | year        | qty    |
|----|-------------|--------|
| a  | 2012        | 10     |
| b  | 2012        | 12     |
| c  | 2013        | 5      |
| b  | 2014        | 7      |
| c  | 2012        | 3      |
</code></pre>

<p>What I want is </p>

<pre><code>| id | year_2012 | year_2013 | year_2014 |
|----|-----------|-----------|-----------|
| a  | 10        | 0         | 0         |
| b  | 12        | 0         | 7         |
| c  | 3         | 5         | 0         |
</code></pre>

<p>or</p>

<pre><code>| id | yearly_qty  |
|----|---------------|
| a  | [10, 0, 0]    |
| b  | [12, 0, 7]    |
| c  | [3, 5, 0]     |
</code></pre>

<p>The closest solution I found is <code>collect_list()</code> but this function doesn't provide order for the list. In my mind the solution should be like:</p>

<pre><code>data.groupBy('id').agg(collect_function)
</code></pre>

<p>Is there a way to generate this without filtering every id out using a loop?</p>

<p>The first one can be easily achieved using <code>pivot</code>:</p>

<pre><code>from itertools import chain

years = sorted(chain(*df.select("year").distinct().collect()))
df.groupBy("id").pivot("year", years).sum("qty")
</code></pre>

<p>which can be further converted to array form:</p>

<pre><code>from pyspark.sql.functions import array, col

(...
    .na.fill(0)
    .select("id",  array(*[col(str(x)) for x in years]).alias("yearly_qty")))
</code></pre>

<p>Obtaining the second one directly is probably not worth all the fuss since you'd have to fill the blanks first. Nevertheless you could try:</p>

<pre><code>from pyspark.sql.functions import collect_list, struct, sort_array, broadcast
years_df = sc.parallelize([(x, ) for x in years], 1).toDF(["year"])

(broadcast(years_df)
    .join(df.select("id").distinct())
    .join(df, ["year", "id"], "leftouter")
    .na.fill(0)
    .groupBy("id")
    .agg(sort_array(collect_list(struct("year", "qty"))).qty.alias("qty")))
</code></pre>

<p>It also requires Spark 2.0+ to get a support for <code>struct</code> collecting.</p>

<p>Both methods are quite expensive so you should be careful when using these. As a rule of thumb long is better than wide.</p>

python apache-spark pyspark apache-spark-sql spark-dataframe