Mit monotonically_increasing_id() für die Zuweisung der Zeilennummer an pyspark dataframe

Ich bin mit monotonically_increasing_id() zuweisen Zeilennummer pyspark dataframe mit syntax unter:

df1 = df1.withColumn("idx", monotonically_increasing_id())

Nun df1 hat 26,572,528 records. Also ich hatte erwartet, idx Wert von 0-26,572,527.

Aber wenn ich select max(idx), sein Wert ist komischerweise riesig: 335,008,054,165.

Was ist Los mit dieser Funktion?
ist es zuverlässig, um diese Funktion zu verwenden, die für die Verschmelzung mit einem anderen dataset mit einer ähnlichen Anzahl von Datensätzen?

Habe ich rund 300 dataframes, die ich will, zu vereinen in einer einzigen dataframe. So ein dataframe enthält IDs und andere enthalten unterschiedliche Aufzeichnungen, die Ihnen entsprechenden row-wise

InformationsquelleAutor muni | 2018-01-11

  1. 13

    Aus der Dokumentation

    Einer Spalte erzeugt monoton steigende 64-bit-Ganzzahlen.

    Die generierte ID ist garantiert monoton steigende und einzigartig, aber nicht in Folge. Die aktuelle Implementierung setzt die partition-ID in der oberen 31 bit, und die Datensatznummer innerhalb jeder partition in den unteren 33 bits. Die Annahme ist, dass die Daten-frame hat weniger als 1 Milliarde Partitionen, jede partition hat weniger als 8 Milliarden Datensätze.

    So, es ist nicht wie ein auto-increment id in RDBs und es ist nicht zuverlässig für die Zusammenführung aus.

    Wenn Sie brauchen ein auto-Inkrement-Verhalten wie in RDBs und Ihre Daten sind sortierbar, dann können Sie row_number

    df.createOrReplaceTempView('df')
spark.sql('select row_number() over (order by "some_column") as num, * from df')
+---+-----------+
|num|some_column|
+---+-----------+
|  1|   ....... |
|  2|   ....... |
|  3| ..........|
+---+-----------+

    Wenn Ihre Daten nicht sortiert werden können, und Sie haben nichts dagegen, mit rdds zu schaffen, die Indizes und dann wieder fallen zu dataframes, die Sie verwenden können rdd.zipWithIndex()

    Kann ein Beispiel gefunden hier

    Kurz:

    # since you have a dataframe, use the rdd interface to create indexes with zipWithIndex()
df = df.rdd.zipWithIndex()
# return back to dataframe
df = df.toDF()

df.show()

# your data           | indexes
+---------------------+---+
|         _1          | _2| 
+-----------=---------+---+
|[data col1,data col2]|  0|
|[data col1,data col2]|  1|
|[data col1,data col2]|  2|
+---------------------+---+

    Müssen Sie wahrscheinlich einige weitere Transformationen nach, die um Ihre dataframe, was Sie brauchen, es zu sein. Hinweis: nicht eine sehr performante Lösung.

    Hoffe, das hilft. Viel Glück!

    Edit:
    Kommen Sie zu denken, können Sie kombinieren die monotonically_increasing_id zu verwenden, die row_number:

    # create a monotonically increasing id 
df = df.withColumn("idx", monotonically_increasing_id())

# then since the id is increasing but not consecutive, it means you can sort by it, so you can use the `row_number`
df.createOrReplaceTempView('df')
new_df = spark.sql('select row_number() over (order by "idx") as num, * from df')

    Nicht sicher über Leistung obwohl.

    kann ich row_number ohne Sortierung. Wie ich nicht wollen, um Sie zu Sortieren, verwenden Sie lieber wie es ist. RDDs zu sein scheinen viel hin und her für rund 300 dataframes
    Leider row_number ist eine windowing-Funktion, und es kann nicht verwendet werden, ohne Sortierung. Werfen Sie einen Blick auf die aktualisierte Antwort, als es helfen könnte.
    wenn wir die Sortierung von "idx", dann brauchen wir nicht wirklich, die row_number für die Teilnahme Recht? Ich kann direkt "idx"
    ja, aber idx ist nicht konsekutiv, das heißt, Sie können nicht tun, ein Graf und vergleichen Sie verschiedene dataframes.
    Lassen Sie uns weiter, diese Diskussion im chat.

    InformationsquelleAutor mkaran

  2. 13

    Verwendung von api-Funktionen, die Sie tun können, einfach die folgenden

    from pyspark.sql.window import Window as W
from pyspark.sql import functions as F
df1 = df1.withColumn("idx", F.monotonically_increasing_id())
windowSpec = W.orderBy("idx")
df1.withColumn("idx", F.row_number().over(windowSpec)).show()

    Ich hoffe die Antwort ist hilfreich

    Bitte seien Sie besonders vorsichtig mit der Antwort, da es verschiebt alle Zeilen in einer einzigen partition (die kann dazu führen, OOM).

    InformationsquelleAutor Ramesh Maharjan

  3. 0

    Fand ich die Lösung von @mkaran nützlich, Aber für mich war es keine Bestellung, Spalte, während Sie mit dem Fenster-Funktion. Ich wollte pflegen Sie die Reihenfolge der Zeilen des dataframe als Ihre Indizes (was Sie sehen, in ein pandas dataframe). Daher die Lösung im edit-Bereich kamen von nutzen. Da ist es eine gute Lösung (wenn die Leistung kein Problem), möchte ich es als eine separate Antwort.

    # Add a increasing data column 
df_index = df.withColumn("idx", monotonically_increasing_id())

# Create the window specification
w = Window.orderBy("idx")

# Use row number with the window specification
df_index = df_index.withColumn("index", F.row_number().over(w))

# Drop the created increasing data column
df2_index = df2_index.drop("idx")

    df Ihre ursprünglichen dataframe und df_index ist neuer dataframe.

    InformationsquelleAutor Ankita Mehta

Schreibe einen Kommentar