我有一個看起來像這樣的 PySpark 資料框：
data = [(2010, 3, 12, 0, 'p1', 'state1'),
(2010, 3, 12, 0, 'p2', 'state2'),
(2010, 3, 12, 0, 'p3', 'state1'),
(2010, 3, 12, 0, 'p4', 'state2'),
(2010, 3, 12, 2, 'p1', 'state3'),
(2010, 3, 12, 2, 'p2', 'state1'),
(2010, 3, 12, 2, 'p3', 'state3'),
(2010, 3, 12, 4, 'p1', 'state1'),
(2010, 3, 12, 6, 'p1', 'state1')]
columns = ['year', 'month', 'day', 'hour', 'process_id','state']
df = spark.createDataFrame(data=data, schema=columns)
df.show()
---- ----- --- ---- ---------- ------
|year|month|day|hour|process_id| state|
---- ----- --- ---- ---------- ------
|2010| 3| 12| 0| p1|state1|
|2010| 3| 12| 0| p2|state2|
|2010| 3| 12| 0| p3|state1|
|2010| 3| 12| 0| p4|state2|
|2010| 3| 12| 2| p1|state3|
|2010| 3| 12| 2| p2|state1|
|2010| 3| 12| 2| p3|state3|
|2010| 3| 12| 4| p1|state1|
|2010| 3| 12| 6| p1|state1|
---- ----- --- ---- ---------- ------
資料框已按四列升序排序：、year和month如上所述。增量以 2 小時為間隔。dayhour
我想知道，對于每一個process_id，它的狀態在每一個內改變了多少次day。為此，我打算使用groupby類似這樣的東西：
chg_count_df = df.groupby('process_id', 'year', 'month', 'day').
agg(.....)
對于此示例，預期輸出為：
---- ----- --- ---------- ----------
|year|month|day|process_id| chg_count|
---- ----- --- ---------- ----------
|2010| 3| 12| p1| 2|
|2010| 3| 12| p2| 1|
|2010| 3| 12| p3| 1|
|2010| 3| 12| p4| 0|
---- ----- --- ---------- ----------
函式中應該包含什么agg(...)？還是有更好的方法來做到這一點？
uj5u.com熱心網友回復：
您可以使用lag視窗函式來檢查狀態是否已更改。然后groupBy使用sum.
from pyspark.sql import functions as F, Window as W
w = W.partitionBy('year', 'month', 'day', 'process_id').orderBy(F.desc('hour'))
df = df.withColumn('change', F.coalesce((F.lag('state').over(w) != F.col('state')).cast('int'), F.lit(0)))
df = df.groupBy('year', 'month', 'day', 'process_id').agg(F.sum('change').alias('chg_count'))
df.show()
# ---- ----- --- ---------- ---------
# |year|month|day|process_id|chg_count|
# ---- ----- --- ---------- ---------
# |2010| 3| 12| p1| 2|
# |2010| 3| 12| p2| 1|
# |2010| 3| 12| p3| 1|
# |2010| 3| 12| p4| 0|
# ---- ----- --- ---------- ---------
uj5u.com熱心網友回復：
chg_count_df = df.groupby('process_id', 'year', 'month', 'day').count()
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/497522.html