根据火花数据框中另一列的值查找列的最大值？

Posted 2023-03-31

【中文标题】根据火花数据框中另一列的值查找列的最大值？

【英文标题】：Find the Max value of a column based on the value of the other column in spark dataframe?

【发布时间】：2016-11-14 12:17:51

【问题描述】：

我有以下数据框，其中第三列是“clickedAirbnb”，第四列是“rank”

|    Tom|             Paris|             |                 1|
|    Tom|            Mumbai|             |                 1|
|    Tom|          Stockolm|             |                 1|
|    Tom|              Oslo|      airbnb1|                 2|
|    Tom|             Tokyo|      airbnb1|                 2|
|    Tom|         Bangalore|      airbnb1|                 2|
|    Sam|             Seoul|     airbnb11|                 1|
|    Sam|             Tokyo|     airbnb11|                 1|
|    Sam|            Berlin|     airbnb12|                 2|
|    Sam|         Bangalore|     airbnb12|                 2|
|    Sam|         Singapore|     airbnb12|                 2|
|    Sam|              Oslo|      airbnb2|                 3|
|    Sam|         Amsterdam|      airbnb2|                 3|
|    Sam|         Bangalore|      airbnb2|                 3|

我想返回“clickedAirbnb”列包含空值且“rank”列的最大值小于 2 的那些。 这是我尝试过但不起作用的方法（抱怨列上的操作无效）

val result1and2 = result.where(col("clickedAirbnb").contains("")
                           && max(col("rank")) <= 2)

有没有办法计算列的最大值？

更新1： 稍微备份一下，结果 df 是这样计算的

          val window = Window.partitionBy(df1("User")).orderBy(df1("clickedAirbnb"))
          val result = df1.withColumn("clickedDestHotRank", dense_rank().over(window))

现在返回clickedAirbnb(第三栏)为空且rank(第四栏)最大值不超过2的用户

val result2 = result.where(col("clickedAirbnb").contains("")
                               && (max(col("rank")) <=2))

【问题讨论】：

你的意思是rank小于2？

@mtoto - 请查看以上更新。

【参考方案1】：

看来你想要：

只对那些在clickedAirbnb中没有价值的人进行分组 当且仅当其 max(rank) 时返回组

类似的东西，也许：

//those that have no value in clickedAirbnb
val resultTmp = result.where(col("clickedAirbnb")==="")

//is its max("rank")<=2 ?  
val b = resultTmp.select(max("rank")<=2).first().getBoolean(0)  

if(b)
  resultTmp.show()

希望我理解得很好。

【参考方案2】：

我希望是这样的：

>>> from pyspark.sql.functions import *
>>> sc = spark.sparkContext
>>> rdd = sc.parallelize([
    ['Tom','Paris','',1],
    ['Tom','Mumbai','',1],
    ['Tom','Stockolm','',1],
    ['Tom','Oslo','airbnb1',2],
    ['Tom','Tokyo','airbnb1',2],
    ['Tom','Bangalore','airbnb1',2],
    ['Sam','Seoul','airbnb11',1],
    ['Sam','Tokyo','airbnb11',1],
    ['Sam','Berlin','airbnb12',2],
    ['Sam','Bangalore','airbnb12',2],
    ['Sam','Singapore','airbnb12',2],
    ['Sam','Oslo','airbnb2',3],
    ['Sam','Amsterdam','airbnb2',3],
    ['Sam','Bangalore','airbnb2',3]
])

>>> df = rdd.toDF(['name','city','clickedAirbnb', 'rank'])
>>> df.show()
+----+---------+-------------+----+
|name|     city|clickedAirbnb|rank|
+----+---------+-------------+----+
| Tom|    Paris|             |   1|
| Tom|   Mumbai|             |   1|
| Tom| Stockolm|             |   1|
| Tom|     Oslo|      airbnb1|   2|
| Tom|    Tokyo|      airbnb1|   2|
| Tom|Bangalore|      airbnb1|   2|
| Sam|    Seoul|     airbnb11|   1|
| Sam|    Tokyo|     airbnb11|   1|
| Sam|   Berlin|     airbnb12|   2|
| Sam|Bangalore|     airbnb12|   2|
| Sam|Singapore|     airbnb12|   2|
| Sam|     Oslo|      airbnb2|   3|
| Sam|Amsterdam|      airbnb2|   3|
| Sam|Bangalore|      airbnb2|   3|
+----+---------+-------------+----+

>>> df.where(col("clickedAirbnb") == "").where(col("rank") <= 2).show()
+----+--------+-------------+----+
|name|    city|clickedAirbnb|rank|
+----+--------+-------------+----+
| Tom|   Paris|             |   1|
| Tom|  Mumbai|             |   1|
| Tom|Stockolm|             |   1|
+----+--------+-------------+----+

【参考方案3】：

注册一个临时表，然后编写所需的查询

your_data_frame.registerTempTable("table1");

res = sqlCtx.sql("select * where clickedAirbnb = "" and max(rank)<=2 from table1) ;