Wie repliziere ich die row_number()-Funktionalität von SQL mit Spark-RDDs für partitionierte Daten?

Spark RDD-Äquivalent von SQL Row_Number für partitionierte Daten

In SQL generiert row_number() eine fortlaufende Nummer für Zeilen innerhalb partitionierter Datensätze. Diese Funktion ist in Spark-RDDs nicht direkt verfügbar. Es gibt jedoch Problemumgehungen, um eine ähnliche Funktionalität zu erreichen.

Partitionierung des RDD

Partitionierung ist entscheidend für die Generierung von Zeilennummern innerhalb von Gruppen. In Ihrem Fall müssen Sie das RDD vor dem Sortieren nach dem Schlüsselwert (K) partitionieren. Betrachten Sie den aktualisierten Code:

val temp2 = temp1
  .map(x => (x._1, (x._2, x._3, x._4)))
  .sortBy(a => (a._1, -a._2._2, -a._2._3))
  .zipWithIndex
  .map(a => (a._1._1, a._1._2._1, a._1._2._2, a._1._2._3, a._2 + 1))

Durch die Anwendung von sortBy auf (a._1, -a._2._2, -a._2._3) sortieren Sie nach Schlüsselwert und dann absteigend nach Spalte 2. und schließlich absteigend col3, was das SQL-Verhalten row_number() nachahmt.

Zeile hinzufügen Zahlen

Nach dem Partitionieren und Sortieren können Sie die Zeilennummern mit zipWithIndex hinzufügen:

val rowNums = temp2.map(a => (a._1, a._2, a._3, a._4, a._5)).cache()

Hinweis: Die in der Antwort bereitgestellte DataFrame-Implementierung ist a Lösung für DataFrames, aber nicht für RDDs.

Das obige ist der detaillierte Inhalt vonWie repliziere ich die row_number()-Funktionalität von SQL mit Spark-RDDs für partitionierte Daten?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!