3

I am using spark with Scala to transform a Dataframe , where I would like to compute a new variable which calculates the rank of one variable per row within many variables.

Example -

Input DF-

+---+---+---+
|c_0|c_1|c_2|
+---+---+---+
| 11| 11| 35|
| 22| 12| 66|
| 44| 22| 12|
+---+---+---+

Expected DF-

+---+---+---+--------+--------+--------+
|c_0|c_1|c_2|c_0_rank|c_1_rank|c_2_rank|
+---+---+---+--------+--------+--------+
| 11| 11| 35|        2|        3|        1|
| 22| 12| 66|       2|       3|       1|
| 44| 22| 12|       1|       2|       3|
+---+---+---+--------+--------+--------+



This has aleady been answered using R - Rank per row over multiple columns in R,

but I need to do the same in spark-sql using scala. Thanks for the Help!

Edit- 4/1 . Encountered one scenario where if the values are same the ranks should be different. Editing first row for replicating the situation.

3
  • 1
    What have you tried so far?
    – Andronicus
    Mar 29, 2019 at 16:39
  • I tried to create a new column of type array with all the elements in it and then try to map it and use zipwithindex after sorting the array to get the index. But after using a map on a df I am stuck and unable to use withcolumn to generate the three rank columns.
    – Amit
    Mar 29, 2019 at 17:00
  • Could you please add more details? What is the rank you need to calculate? Mar 29, 2019 at 17:14

3 Answers 3

1

If I understand correctly, you want to have the rank of each column, within each row.

Let's first define the data, and the columns to "rank".

val df = Seq((11,  21,  35),(22,  12, 66),(44, 22 , 12))
    .toDF("c_0", "c_1", "c_2")
val cols = df.columns

Then we define a UDF that finds the index of an element in an array.

val pos = udf((a : Seq[Int], elt : Int) => a.indexOf(elt)+1)

We finally create a sorted array (in descending order) and use the UDF to find the rank of each column.

val ranks = cols.map(c => pos(col("array"), col(c)).as(c+"_rank"))
df.withColumn("array", sort_array(array(cols.map(col) : _*), false))
  .select((cols.map(col)++ranks) :_*).show 
+---+---+---+--------+--------+--------+
|c_0|c_1|c_2|c_0_rank|c_1_rank|c_2_rank|
+---+---+---+--------+--------+--------+
| 11| 12| 35|       3|       2|       1|
| 22| 12| 66|       2|       3|       1|
| 44| 22| 12|       1|       2|       3|
+---+---+---+--------+--------+--------+

EDIT: As of Spark 2.4, the pos UDF that I defined can be replaced by the built in function array_position(column: Column, value: Any) that works exactly the same way (the first index is 1). This avoids using UDFs that can be slightly less efficient.

EDIT2: The code above will generate duplicated indices in case you have duplidated keys. If you want to avoid it, you can create the array, zip it to remember which column is which, sort it and zip it again to get the final rank. It would look like this:

val colMap = df.columns.zipWithIndex.map(_.swap).toMap
val zip = udf((s: Seq[Int]) => s
    .zipWithIndex
    .sortBy(-_._1)
    .map(_._2)
    .zipWithIndex
    .toMap
    .mapValues(_+1))
val ranks = (0 until cols.size)
    .map(i => 'zip.getItem(i) as colMap(i) + "_rank")
val result = df
    .withColumn("zip", zip(array(cols.map(col) : _*)))
    .select(cols.map(col) ++ ranks :_*)
7
  • This works, have accepted it, just wondering if using udf would impact the performance if used over a large dataset due to its nature of ser-deser , Also what are your thoughts on the parallelism , would this not be prone to OOM.
    – Amit
    Mar 30, 2019 at 8:20
  • This code is not prone to OOM at all. It is a simple row -wise calculation that will be perfectly distributed. No need to group rows in any way, and nothing on the driver so you're safe. Also because you asked about the UDF, I checked and as of spark 2.4, a built in function can replace my UDF (I edited my answer to mention it). Yet even if using the UDF, I don't think that the performance would suffer that much. If you try both, let us know ;-)
    – Oli
    Mar 30, 2019 at 22:36
  • Thanks Oli , will check on this and get back with performance results.
    – Amit
    Mar 31, 2019 at 3:23
  • I guess this would return me same index(rank in this case) if element values are same. I want them to increment.(11,11,35 should return 2.3,1)Should i add logic in udf or u believe there is an easier way.
    – Amit
    Apr 1, 2019 at 10:57
  • With this method, there cannot be any duplicated indices. Indeed, an element can only have one index in an array. BTW, if you're using Spark 2.4, you don't even need a UDF ;)
    – Oli
    Apr 1, 2019 at 11:07
0

One way to go about this would be to use windows.

val df = Seq((11,  21,  35),(22,  12, 66),(44, 22 , 12))
    .toDF("c_0", "c_1", "c_2")
(0 to 2)
    .map("c_"+_)
    .foldLeft(df)((d, column) => 
          d.withColumn(column+"_rank", rank() over Window.orderBy(desc(column))))
    .show
+---+---+---+--------+--------+--------+                                        
|c_0|c_1|c_2|c_0_rank|c_1_rank|c_2_rank|
+---+---+---+--------+--------+--------+
| 22| 12| 66|       2|       3|       1|
| 11| 21| 35|       3|       2|       2|
| 44| 22| 12|       1|       1|       3|
+---+---+---+--------+--------+--------+

But this is not a good idea. All the data will end up in one partition which will cause an OOM error if all the data does not fit inside one executor.

Another way would require to sort the dataframe three times, but at least that would scale to any size of data.

Let's define a function that zips a dataframe with consecutive indices (it exists for RDDs but not for dataframes)

def zipWithIndex(df : DataFrame, name : String) : DataFrame = {
    val rdd = df.rdd.zipWithIndex
      .map{ case (row, i) => Row.fromSeq(row.toSeq :+ (i+1)) }
    val newSchema = df.schema.add(StructField(name, LongType, false))
    df.sparkSession.createDataFrame(rdd, newSchema)
}

And let's use it on the same dataframe df:

(0 to 2)
    .map("c_"+_)
    .foldLeft(df)((d, column) => 
        zipWithIndex(d.orderBy(desc(column)), column+"_rank"))
    .show

which provides the exact same result as above.

5
  • I have another suggestion to use case class. This will help to not convert it to RDD and use direct Spark Dataset. Mar 29, 2019 at 17:26
  • thanks for the prompt reply, but the output does not match with the expected output. i expect lowest rank for highest value ..eg - c_0_rank should be 1 for c_0 having value 44 in first row not 3.
    – Amit
    Mar 29, 2019 at 18:20
  • Right, I missed the fact that it was sorted in descending order. I edited my answer. It's fixed.
    – Oli
    Mar 29, 2019 at 18:33
  • first row is correct, for second and third the output is still incorrect. please check the expected DF in the question i have formatted it for better clarity.
    – Amit
    Mar 29, 2019 at 18:41
  • Since I had not understood your question, I posted a new answer.
    – Oli
    Mar 29, 2019 at 20:00
0

You could probably create a window function. Do note that this is susceptible to OOM if you have too much data. But, I just wanted to introduce to the concept of window functions here.

inputDF.createOrReplaceTempView("my_df")
val expectedDF =  spark.sql("""
    select 
        c_0
        , c_1
        , c_2
        , rank(c_0) over (order by c_0 desc) c_0_rank
        , rank(c_1) over (order by c_1 desc) c_1_rank
        , rank(c_2) over (order by c_2 desc) c_2_rank 
    from my_df""")
expectedDF.show()

+---+---+---+--------+--------+--------+
|c_0|c_1|c_2|c_0_rank|c_1_rank|c_2_rank|
+---+---+---+--------+--------+--------+
| 44| 22| 12|       3|       3|       1|
| 11| 21| 35|       1|       2|       2|
| 22| 12| 66|       2|       1|       3|
+---+---+---+--------+--------+--------+
3
  • desired output should be- c_0,c_1,c_2,c_0_rank,c_1_rank,c_2_rank 44,22,12,1,2,3 11,21,35,3,2,1 22,12,66,2,3,1 highest number having lowest rank where c_n_rank column specifies corresponding rank for c_n column value.
    – Amit
    Mar 29, 2019 at 18:20
  • Edited for ordering by rank desc
    – rmathews7
    Mar 29, 2019 at 18:36
  • please check the expected DF in question, have edited it for better clarity
    – Amit
    Mar 29, 2019 at 18:41

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.