Window 関数: 現在の行より前の行offset値を返し、現在の行より前にoffset行が存在する場合はdefaultします。 たとえば、1 の offset は、ウィンドウ パーティション内の特定のポイントで前の行を返します。
これは、SQL の LAG 関数と同じです。
構文
from pyspark.sql import functions as sf
sf.lag(col, offset=1, default=None)
パラメーター
| パラメーター | タイプ | Description |
|---|---|---|
col |
pyspark.sql.Column または列名 |
列または式の名前。 |
offset |
int、省略可能 | 拡張する行の数。 既定値の は 1 です。 |
default |
任意 | 既定値。 |
返品ポリシー
pyspark.sql.Column: offsetに基づく現在の行の前の値。
例示
例 1: ラグを使用して前の値を取得する
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|
+---+---+--------------+
例 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", 1, 0).over(w)).show()
+---+---+--------------+
| c1| c2|previous_value|
+---+---+--------------+
| a| 1| 0|
| a| 2| 1|
| a| 3| 2|
| b| 2| 0|
| b| 8| 2|
+---+---+--------------+
例 3: オフセットが 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|
+---+---+--------------+