我有一個 PySpark 資料框，如下所示：
Id timestamp col1 col2
abc 789 0 1
def 456 1 0
abc 123 1 0
def 321 0 1
我想按 ID 列分組或磁區，然后應該根據時間戳的順序創建 col1 和 col2 的串列。
Id timestamp col1 col2
abc [123,789] [1,0] [0,1]
def [321,456] [0,1] [1,0]
我的做法：
from pyspark.sql import functions as F
from pyspark.sql import Window as W
window_spec = W.partitionBy("id").orderBy('timestamp')
ranged_spec = window_spec.rowsBetween(W.unboundedPreceding, W.unboundedFollowing)
df1 = df.withColumn("col1", F.collect_list("reco").over(window_spec))\
.withColumn("col2", F.collect_list("score").over(window_spec))\
df1.show()
但這不是回傳 col1 和 col2 的串列。
uj5u.com熱心網友回復：
我不認為使用groupBy聚合可以可靠地保留訂單。所以視窗函式似乎是要走的路。
設定：
from pyspark.sql import functions as F, Window as W
df = spark.createDataFrame(
[('abc', 789, 0, 1),
('def', 456, 1, 0),
('abc', 123, 1, 0),
('def', 321, 0, 1)],
['Id', 'timestamp', 'col1', 'col2'])
腳本：
w1 = W.partitionBy('Id').orderBy('timestamp')
w2 = W.partitionBy('Id').orderBy(F.desc('timestamp'))
df = df.select(
'Id',
*[F.collect_list(c).over(w1).alias(c) for c in df.columns if c != 'Id']
)
df = (df
.withColumn('_rn', F.row_number().over(w2))
.filter('_rn=1')
.drop('_rn')
)
結果：
df.show()
# --- ---------- ------ ------
# | Id| timestamp| col1| col2|
# --- ---------- ------ ------
# |abc|[123, 789]|[1, 0]|[0, 1]|
# |def|[321, 456]|[0, 1]|[1, 0]|
# --- ---------- ------ ------
你也非常接近你需要的東西。我玩過，這似乎也有效：
window_spec = W.partitionBy("Id").orderBy('timestamp')
ranged_spec = window_spec.rowsBetween(W.unboundedPreceding, W.unboundedFollowing)
df1 = (df
.withColumn("timestamp", F.collect_list("timestamp").over(ranged_spec))
.withColumn("col1", F.collect_list("col1").over(ranged_spec))
.withColumn("col2", F.collect_list("col2").over(ranged_spec))
).drop_duplicates()
df1.show()
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/497516.html