我有兩個 Spark 資料框：
df1
--- ----
| id| var|
--- ----
|323| [a]|
--- ----
df2
---- ---------- ----------
| src| str_value| num_value|
---- ---------- ----------
| [a]| ghn12| 0.0 |
---- ---------- ----------
| [a]| 54fdg| 1.2 |
---- ---------- ----------
| [a]| 90okl| 0.7 |
---- ---------- ----------
| [b]| jh456| 0.5 |
---- ---------- ----------
| [a]| ghn12| 0.2 |
---- ---------- ----------
| [c]| ghn12| 0.7 |
---- ---------- ----------
我需要從資料框中回傳前 3 行，其中df2值最小。因此，所需的輸出是（按 排序）：df1.var == df2.srcdf2.num_valuenum_value
---- ---------- ----------
| src| str_value| num_value|
---- ---------- ----------
| [a]| ghn12| 0.0 |
---- ---------- ----------
| [a]| ghn12| 0.2 |
---- ---------- ----------
| [a]| 90okl| 0.7 |
---- ---------- ----------
我知道如何使用 SQL 來實作這一點，但我對 PySpark/Spark SQL 有一些困難。
uj5u.com熱心網友回復：
我會使用dense_rank視窗函式來做到這一點。
from pyspark.sql import functions as F, Window as W
w = W.partitionBy('src').orderBy('num_value')
df3 = (
df2
.join(df1, df2.src == df1.var, 'semi')
.withColumn('_rank', F.dense_rank().over(w))
.filter('_rank <= 3')
.drop('_rank')
)
uj5u.com熱心網友回復：
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col
windowSpec = Window.partitionBy("src").orderBy("num_value")
df_joined = df1.join(df2,df1.var==df2.src).drop("var", "id")
df_joined.withColumn("row_number",row_number().over(windowSpec)).filter(col("row_number")<4).drop("row_number").show()
# --- --------- ---------
# |src|str_value|num_value|
# --- --------- ---------
# |[a]| ghn12| 0.0|
# |[a]| ghn12| 0.2|
# |[a]| 90okl| 0.7|
# --- --------- ---------
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/483624.html