Wie füge ich eine dauerhafte Spalte mit Zeilen-IDs zu Spark DataFrame hinzu?

Ich konnte das nicht reproduzieren. Ich verwende Spark 2.0, vielleicht hat sich das Verhalten geändert oder ich mache nicht dasselbe wie Sie.

val df = Seq(("one", 1,true),("two", 2,false),("three", 3,true),("four", 4,true))
.toDF("name", "value","flag")
.withColumn("rowd", monotonically_increasing_id())

df.show

val df2 = df.filter(col("flag")=== true)

df2.show

df: org.Apache.spark.sql.DataFrame = [name: string, value: int ... 2 more fields]
+-----+-----+-----+----+
| name|value| flag|rowd|
+-----+-----+-----+----+
|  one|    1| true|   0|
|  two|    2|false|   1|
|three|    3| true|   2|
| four|    4| true|   3|
+-----+-----+-----+----+
df2: org.Apache.spark.sql.Dataset[org.Apache.spark.sql.Row] = [name: string, value: int ... 2 more fields]
+-----+-----+----+----+
| name|value|flag|rowd|
+-----+-----+----+----+
|  one|    1|true|   0|
|three|    3|true|   2|
| four|    4|true|   3|
+-----+-----+----+----+

Ich habe kürzlich an einem ähnlichen Problem gearbeitet. Obwohl monotonically_increasing_id() sehr schnell ist, ist es nicht zuverlässig und gibt Ihnen keine fortlaufenden Zeilennummern, sondern erhöht nur eindeutige ganze Zahlen.

Das Erstellen einer Windows-Partition und die anschließende Verwendung von row_number().over(some_windows_partition) ist sehr zeitaufwändig. 

Die bisher beste Lösung ist die Verwendung von gezippt mit Index und die anschließende Konvertierung der gezippten Datei in den ursprünglichen Datenrahmen, wobei das neue Schema die Indexspalte enthält.

Versuche dies:

from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, LongType

new_schema = StructType(**original_dataframe**.schema.fields[:] + [StructField("index", LongType(), False)])
zipped_rdd = **original_dataframe**.rdd.zipWithIndex()
indexed = (zipped_rdd.map(lambda ri: row_with_index(*list(ri[0]) + [ri[1]])).toDF(new_schema))

Wo original_dataframe die dataframe ist, müssen Sie einen Index hinzufügen und row_with_index ist das neue Schema mit dem Spaltenindex, den Sie schreiben können 

row_with_index = Row(
"calendar_date"
,"year_week_number"
,"year_period_number"
,"realization"
,"index"
)

Hier waren calendar_date, year_week_number, year_period_number und realization die Spalten meiner ursprünglichen dataframe. Sie können die Namen durch die Namen Ihrer Spalten ersetzen. Der Index ist der neue Spaltenname, den Sie für die Zeilennummern hinzufügen mussten.

Dieser Prozess ist im Vergleich zur row_number().over(some_windows_partition)-Methode wesentlich effizienter und reibungsloser.

Hoffe das hilft.

Um die Verschiebungsauswertung von monotonically_increasing_id () zu umgehen, können Sie versuchen, den Datenrahmen auf die Festplatte zu schreiben und erneut zu lesen. Dann ist die ID-Spalte jetzt einfach ein Datenfeld, das gerade gelesen wird, und nicht irgendwann in der Pipeline dynamisch berechnet. Obwohl es eine ziemlich hässliche Lösung ist, funktionierte es bei einem schnellen Test.

Das hat bei mir funktioniert. Erstellt eine weitere Identitätsspalte und verwendet die Fensterfunktion row_number

import org.Apache.spark.sql.functions.{row_number}
import org.Apache.spark.sql.expressions.Window

val df1: DataFrame = df.withColumn("Id",lit(1))

df1
.select(
...,
row_number()
.over(Window
.partitionBy("Id"
.orderBy(col("...").desc))
)
.alias("Row_Nbr")
)

Um eine bessere Leistung mit der Chris T-Lösung zu erzielen, können Sie versuchen, in einen gemeinsam genutzten Datenrahmen von Apache Ignite zu schreiben, anstatt auf die Festplatte . https://ignite.Apache.org/use-cases/spark/ shared-memory-layer.html