Donnerstag, Juni 4, 2020

Computing Rang einer Zeile

Ich möchte Ranges Benutzer-id basiert auf einem Feld. Für den gleichen Wert des Feldes, Rang sollte gleich sein. Dass Daten in Hive-Tabelle.

z.B.

user value
a       5
b       10
c       5
d       6

Rank
a - 1
c - 1
d - 3
b - 4

Wie kann ich das tun?

1 Kommentar

  1. 22

    Ist es möglich, zu verwenden rank Fenster-Funktion entweder mit einem DataFrame API:

    import org.apache.spark.sql.functions.rank
    import org.apache.spark.sql.expressions.Window
    
    val w = Window.orderBy($"value")
    
    val df = sc.parallelize(Seq(
      ("a", 5), ("b", 10), ("c", 5), ("d", 6)
    )).toDF("user", "value")
    
    df.select($"user", rank.over(w).alias("rank")).show
    
    //+----+----+
    //|user|rank|
    //+----+----+
    //|   a|   1|
    //|   c|   1|
    //|   d|   3|
    //|   b|   4|
    //+----+----+

    – oder raw-SQL:

    df.registerTempTable("df")
    sqlContext.sql("SELECT user, RANK() OVER (ORDER BY value) AS rank FROM df").show
    
    //+----+----+
    //|user|rank|
    //+----+----+
    //|   a|   1|
    //|   c|   1|
    //|   d|   3|
    //|   b|   4|
    //+----+----+

    aber es ist äußerst ineffizient.

    Können Sie auch versuchen, RDD-API, aber es ist nicht ganz einfach. Zunächst können konvertieren DataFrame zu RDD:

    import org.apache.spark.sql.Row
    import org.apache.spark.rdd.RDD
    import org.apache.spark.RangePartitioner
    
    val rdd: RDD[(Int, String)] = df.select($"value", $"user")
      .map{ case Row(value: Int, user: String) => (value, user) }
    
    val partitioner = new RangePartitioner(rdd.partitions.size,  rdd)
    val sorted =  rdd.repartitionAndSortWithinPartitions(partitioner)

    Als Nächstes müssen wir berechnen Reihen pro partition:

    def rank(iter: Iterator[(Int,String)]) =  {
      val zero = List((-1L, Integer.MIN_VALUE, "", 1L))
    
      def f(acc: List[(Long,Int,String,Long)], x: (Int, String)) = 
        (acc.head, x) match {
          case (
            (prevRank: Long, prevValue: Int, _, offset: Long),
            (currValue: Int, label: String)) => {
          val newRank = if (prevValue == currValue) prevRank else prevRank + offset
          val newOffset = if (prevValue == currValue) offset + 1L else 1L
          (newRank, currValue, label, newOffset) :: acc
        }
      }
    
      iter.foldLeft(zero)(f).reverse.drop(1).map{case (rank, _, label, _) =>
        (rank, label)}.toIterator
    }
    
    
    val partRanks = sorted.mapPartitions(rank)

    offset für jede partition

    def getOffsets(sorted: RDD[(Int, String)]) = sorted
      .mapPartitionsWithIndex((i: Int, iter: Iterator[(Int, String)]) => 
        Iterator((i, iter.size)))
      .collect
      .foldLeft(List((-1, 0)))((acc: List[(Int, Int)], x: (Int, Int)) => 
        (x._1, x._2 + acc.head._2) :: acc)
      .toMap
    
    val offsets = sc.broadcast(getOffsets(sorted))

    und die letzten Reihen:

    def adjust(i: Int, iter: Iterator[(Long, String)]) = 
      iter.map{case (rank, label) => (rank + offsets.value(i - 1).toLong, label)}
    
    val ranks = partRanks
      .mapPartitionsWithIndex(adjust)
      .map{case (i, label) => (1 + i , label)}
    • Ich denke, das ist eine tolle Antwort, jedoch konnten wir uns mehr Ausarbeitung, warum der dataframe API ist ineffizient hier?
    • Da Window definition ohne partitionBy mischt alles zu einer einzigen partition. Mit der heutigen Daset API -, Sie umschreiben könnte RDD version, obwohl.
    • Hinzufügen @zero323 Antwort, es kann ineffizient sein, auch mit der partitionBy – zum Beispiel, in einigen Arten von Transaktions-Daten, kann es üblich sein, für eine kleine Anzahl von Kunden zu halten, die die überwiegende Mehrheit der Transaktionen; ich bin gekommen, über banking-Daten, bei denen ein Kunde gültig hatten 45% aller Transaktionen über die bank, die wegen der bank war ein market maker ist und war (in den Daten), ein Kunde von sich selbst.

Kostenlose Online-Tests