Verzögerung

Fensterfunktion: Gibt den Wert zurück, der Zeilen vor der aktuellen Zeile ist offset , und default wenn es weniger als offset Zeilen vor der aktuellen Zeile gibt. Beispielsweise gibt eine offset von 1 die vorherige Zeile an einem beliebigen Punkt in der Fensterpartition zurück.

Dies entspricht der LAG-Funktion in SQL.

Syntax

from pyspark.sql import functions as sf

sf.lag(col, offset=1, default=None)

Die Parameter

Parameter	Typ	Description
col	pyspark.sql.Column oder Spaltenname	Name der Spalte oder des Ausdrucks.
offset	int, optional	Die Anzahl der zu erweiternden Zeilen. Der Standardwert ist 1.
default	optional	Standardwert.

Rückkehr

pyspark.sql.Column: Wert vor der aktuellen Zeile basierend auf offset.

Examples

Beispiel 1: Verwenden von Verzögerung zum Abrufen eines vorherigen Werts

from pyspark.sql import functions as sf
from pyspark.sql import Window
df = spark.createDataFrame(
    [("a", 1), ("a", 2), ("a", 3), ("b", 8), ("b", 2)], ["c1", "c2"])
df.show()

+---+---+
| c1| c2|
+---+---+
|  a|  1|
|  a|  2|
|  a|  3|
|  b|  8|
|  b|  2|
+---+---+

w = Window.partitionBy("c1").orderBy("c2")
df.withColumn("previous_value", sf.lag("c2").over(w)).show()

+---+---+--------------+
| c1| c2|previous_value|
+---+---+--------------+
|  a|  1|          NULL|
|  a|  2|             1|
|  a|  3|             2|
|  b|  2|          NULL|
|  b|  8|             2|
+---+---+--------------+

Beispiel 2: Verwenden von Verzögerung mit einem Standardwert

from pyspark.sql import functions as sf
from pyspark.sql import Window
df = spark.createDataFrame(
    [("a", 1), ("a", 2), ("a", 3), ("b", 8), ("b", 2)], ["c1", "c2"])
w = Window.partitionBy("c1").orderBy("c2")
df.withColumn("previous_value", sf.lag("c2", 1, 0).over(w)).show()

+---+---+--------------+
| c1| c2|previous_value|
+---+---+--------------+
|  a|  1|             0|
|  a|  2|             1|
|  a|  3|             2|
|  b|  2|             0|
|  b|  8|             2|
+---+---+--------------+

Beispiel 3: Verwenden von Verzögerung mit einem Offset von 2

from pyspark.sql import functions as sf
from pyspark.sql import Window
df = spark.createDataFrame(
    [("a", 1), ("a", 2), ("a", 3), ("b", 8), ("b", 2)], ["c1", "c2"])
w = Window.partitionBy("c1").orderBy("c2")
df.withColumn("previous_value", sf.lag("c2", 2, -1).over(w)).show()

+---+---+--------------+
| c1| c2|previous_value|
+---+---+--------------+
|  a|  1|            -1|
|  a|  2|            -1|
|  a|  3|             1|
|  b|  2|            -1|
|  b|  8|            -1|
+---+---+--------------+