title: Pyspark - Ranking columns keeping ties
body: <p>I'm looking for a way to rank columns of a dataframe preserving ties.  Specifically for this example, I have a pyspark dataframe as follows where I want to generate ranks for colA &amp; colB (though I want to support being able to rank N number of columns)    </p>
  
  <pre><code> +--------+----------+-----+----+
   |  Entity|        id| colA|colB|
   +-------------------+-----+----+
   |       a|8589934652|   21|  50|
   |       b|       112|    9|  23|
   |       c|8589934629|    9|  23|
   |       d|8589934702|    8|  21|         
   |       e|        20|    2|  21|        
   |       f|8589934657|    2|   5|          
   |       g|8589934601|    1|   5|         
   |       h|8589934653|    1|   4|          
   |       i|8589934620|    0|   4|          
   |       j|8589934643|    0|   3|         
   |       k|8589934618|    0|   3|         
   |       l|8589934602|    0|   2|         
   |       m|8589934664|    0|   2|         
   |       n|        25|    0|   1|         
   |       o|        67|    0|   1|         
   |       p|8589934642|    0|   1|         
   |       q|8589934709|    0|   1|         
   |       r|8589934660|    0|   1|         
   |       s|        30|    0|   1|         
   |       t|        55|    0|   1|         
   +--------+----------+-----+----+
  </code></pre>
  
  <p>What I'd like is a way to rank this dataframe where tied values receive the same rank such as:</p>
  
  <pre><code> +--------+----------+-----+----+---------+---------+
   |  Entity|        id| colA|colB|colA_rank|colB_rank|
   +-------------------+-----+----+---------+---------+
   |       a|8589934652|   21|  50|        1|        1|
   |       b|       112|    9|  23|        2|        2|
   |       c|8589934629|    9|  21|        2|        3|
   |       d|8589934702|    8|  21|        3|        3|        
   |       e|        20|    2|  21|        4|        3|      
   |       f|8589934657|    2|   5|        4|        4|       
   |       g|8589934601|    1|   5|        5|        4|     
   |       h|8589934653|    1|   4|        5|        5|     
   |       i|8589934620|    0|   4|        6|        5|    
   |       j|8589934643|    0|   3|        6|        6|  
   |       k|8589934618|    0|   3|        6|        6| 
   |       l|8589934602|    0|   2|        6|        7|
   |       m|8589934664|    0|   2|        6|        7|
   |       n|        25|    0|   1|        6|        8|
   |       o|        67|    0|   1|        6|        8|
   |       p|8589934642|    0|   1|        6|        8|
   |       q|8589934709|    0|   1|        6|        8|
   |       r|8589934660|    0|   1|        6|        8|
   |       s|        30|    0|   1|        6|        8|
   |       t|        55|    0|   1|        6|        8|
   +--------+----------+-----+----+---------+---------+
  </code></pre>
  
  <p>My current implementation with the first dataframe looks like:</p>
  
  <pre><code> def getRanks(mydf, cols=None, ascending=False):
       from pyspark import Row
       # This takes a dataframe and a list of columns to rank
       # If no list is provided, it ranks *all* columns
       # returns a new dataframe
  
       def addRank(ranked_rdd, col, ascending):
           # This assumes an RDD of the form (Row(...), list[...])
           # it orders the rdd by col, finds the order, then adds that to the 
           # list
           myrdd = ranked_rdd.sortBy(lambda (row, ranks):  row[col], 
                   ascending=ascending).zipWithIndex()
           return myrdd.map(lambda ((row, ranks), index): (row, ranks + 
                  [index+1]))
  
       myrdd = mydf.rdd
       fields = myrdd.first().__fields__
       ranked_rdd = myrdd.map(lambda x: (x, []))
  
       if (cols is None):
           cols = fields
       for col in cols:
           ranked_rdd = addRank(ranked_rdd, col, ascending)
       rank_names = [x + "_rank" for x in cols]
  
       # Hack to make sure columns come back in the right order
       ranked_rdd = ranked_rdd.map(lambda (row, ranks): Row(*row.__fields__ + 
                    rank_names)(*row + tuple(ranks)))
       return ranked_rdd.toDF()
  </code></pre>
  
  <p>which produces:</p>
  
  <pre><code> +--------+----------+-----+----+---------+---------+
   |  Entity|        id| colA|colB|colA_rank|colB_rank|
   +-------------------+-----+----+---------+---------+
   |       a|8589934652|   21|  50|        1|        1|
   |       b|       112|    9|  23|        2|        2|
   |       c|8589934629|    9|  23|        3|        3|
   |       d|8589934702|    8|  21|        4|        4|        
   |       e|        20|    2|  21|        5|        5|      
   |       f|8589934657|    2|   5|        6|        6|       
   |       g|8589934601|    1|   5|        7|        7|     
   |       h|8589934653|    1|   4|        8|        8|     
   |       i|8589934620|    0|   4|        9|        9|    
   |       j|8589934643|    0|   3|       10|       10|  
   |       k|8589934618|    0|   3|       11|       11|
   |       l|8589934602|    0|   2|       12|       12|
   |       m|8589934664|    0|   2|       13|       13|
   |       n|        25|    0|   1|       14|       14|
   |       o|        67|    0|   1|       15|       15|
   |       p|8589934642|    0|   1|       16|       16|
   |       q|8589934709|    0|   1|       17|       17|
   |       r|8589934660|    0|   1|       18|       18|
   |       s|        30|    0|   1|       19|       19|
   |       t|        55|    0|   1|       20|       20|
   +--------+----------+-----+----+---------+---------+
  </code></pre>
  
  <p>As you can see, the function getRanks() takes a dataframe, specifies the columns to be ranked, sorts them, and uses zipWithIndex() to generate an ordering or rank.  However, I can't figure out a way to preserve ties.  </p>
  
  <p>This stackoverflow post is the closest solution I've found:
  <a href="https://stackoverflow.com/questions/36605062/pyspark-add-a-new-column-with-a-rank-by-user">rank-users-by-column</a> But it appears to only handle 1 column (I think).  </p>
  
  <p>Thanks so much for the help in advance!</p>
  
  <p>EDIT:  column 'id' is generated from calling monotonically_increasing_id() and in my implementation is cast to a string.</p>