Getting all columns of Spark DataFrame after aggregation

Question

Suppose I've got a data frame :

+----+----+---+
|  c1|name|qty|
+----+----+---+
|abc1|   a|  1|
|abc2|   a|  0|
|abc3|   b|  3|
|abc4|   b|  2|
+----+----+---+

I would like to get only rows with minimal qty for every name:

+----+----+---+
|  c1|name|qty|
+----+----+---+
|abc2|   a|  0|
|abc4|   b|  2|
+----+----+---+

I am doing it like that:

df1 = df.groupBy('name').agg(sf.min('qty')).select("min(qty)")
df2 = df1.join(df, df1["min(qty)"] == df["qty"]).drop("min(qty)") // df2 is the result

It's working. I am wondering if it can be improved. How could you improve the solution above ?

Answer 1

You can use reduceBuKey here on the dataframe. Reduce based on the name and then select the lower key. I assumed the df is associated with a dataset with

case class (c1:String, name:String, qty:Integer)

val mappedPairRdd =
    df.map(row ⇒ (row.name, (row.c1, row.name, row.qty))).rdd;
  val reduceByKeyRDD = mappedPairRdd
    .reduceByKey((row1, row2) ⇒ {
      if (row1._3 > row2._3) {
        row2
      } else {
        row1
      }
    })
    .values;