我想使用每個 ID 鍵的聚合來選擇具有 max(day) 的行。
| ID | col1 | col2 | 月 | 天 |
|---|---|---|---|---|
| 人工智能1 | 5 | 2 | 簡夫 | 15 |
| 人工智能2 | 6 | 0 | 十二月 | 16 |
| 人工智能1 | 1 | 7 | 行進 | 16 |
| 人工智能3 | 9 | 4 | 十一月 | 18 |
| 人工智能2 | 3 | 20 | 費夫 | 20 |
| 人工智能3 | 10 | 8 | 六月 | 06 |
期望的結果：
| ID | col1 | col2 | 月 | 天 |
|---|---|---|---|---|
| 人工智能1 | 1 | 7 | 行進 | 16 |
| 人工智能2 | 3 | 20 | 費夫 | 20 |
| 人工智能3 | 9 | 4 | 十一月 | 18 |
uj5u.com熱心網友回復：
我想到的唯一解決方案是：
- 獲取每個 ID 的最高日（使用groupBy）
- 使用join將最高天的值附加到每一行（具有匹配的 ID）
- 然后是一個簡單的過濾器，其中兩行的值匹配
# select the max value for each of the ID
maxDayForIDs = df.groupBy("ID").max("day").withColumnRenamed("max(day)", "maxDay")
# now add the max value of the day for each line (with matching ID)
df = df.join(maxDayForIDs, "ID")
# keep only the lines where it matches "day" equals "maxDay"
df = df.filter(df.day == df.maxDay)
uj5u.com熱心網友回復：
通常這種操作是使用視窗函式來完成的，比如
rank,
dense_rank
或row_number。
from pyspark.sql import functions as F, Window as W
df = spark.createDataFrame(
[('AI1', 5, 2, 'janv', '15'),
('AI2', 6, 0, 'Dec', '16'),
('AI1', 1, 7, 'March', '16'),
('AI3', 9, 4, 'Nov', '18'),
('AI2', 3, 20, 'Fev', '20'),
('AI3', 10, 8, 'June', '06')],
['ID', 'col1', 'col2', 'month', 'Day']
)
w = W.partitionBy('ID').orderBy(F.desc('Day'))
df = df.withColumn('_rn', F.row_number().over(w))
df = df.filter('_rn=1').drop('_rn')
df.show()
# --- ---- ---- ----- ---
# | ID|col1|col2|month|Day|
# --- ---- ---- ----- ---
# |AI1| 1| 7|March| 16|
# |AI2| 3| 20| Fev| 20|
# |AI3| 9| 4| Nov| 18|
# --- ---- ---- ----- ---
uj5u.com熱心網友回復：
讓它變得簡單
new= (df.withColumn('max',first('Day').over(w))#Order by day descending and keep first value in a group in max
.where(col('Day')==col('max'))#filter where max=Day
.drop('max')#drop max
).show()
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/481494.html
標籤：数据框 阿帕奇火花 pyspark apache-spark-sql 最大限度
下一篇：Pyspark中的比較資料框