查询另一个 DataFrame 查询 Spark 的结果答案

【问题标题】：Query a result of another DataFrame query Spark查询另一个 DataFrame 查询 Spark 的结果
【发布时间】：2018-05-30 00:08:51
【问题描述】：

为 Python 使用 spark 和 Pandas 库。我基本上有以下几点：

df.groupby('sIP').agg({'duration': 'sum'}).show()

这给了我以下输出：

+---------------+------------------+                                            
|            sIP|     sum(duration)|
+---------------+------------------+
| 203.13.173.243|               0.0|
|  204.62.16.153|         17522.927|
|  203.30.177.95|             1.081|
|   56.23.191.99|           113.186|

所有不同源 IP 地址持续时间的总和。

我现在要做的是从该总和中获取最大值，并获取关联的源 IP。所以我必须在修改后查询 DataFrame（上面的查询）。

所以我尝试：

df.groupby('sIP').agg({'duration': 'max'}).show()

但还是查询到原来的DataFrame，如下：

+---------------+-------------+                                                 
|            sIP|max(duration)|
+---------------+-------------+
| 203.13.173.243|          0.0|
|  204.62.16.153|     1790.305|
|  203.30.177.95|        1.081|
|   56.23.191.99|       22.563|

是否有覆盖数据框状态的方法？我看过转换，但没有成功，collect() 是我研究过的另一件事，但仍然一无所获......

我错过了什么？

提前致谢。

【问题讨论】：

标签： python sql pandas apache-spark dataframe

【解决方案1】：

您需要将第一个转换保存到新的数据帧，否则结果将被丢弃。

df1 = df.groupby('sIP').agg({'duration': 'max'}).toDF('sIP', 'max_duration')

然后你就可以对新的dataframe随意操作了，即：

df1.sort('max_duration', ascending=False).show(1)

无论如何，您都可以在没有中间数据帧的情况下在一个转换中完成这两项工作：

df.groupby('sIP').agg({'duration': 'max'}).sort('max(duration)', ascending=False).show(1)

【讨论】：

我无法将它们应用在一起。但是第一个解决方案（单独的）效果很好。谢谢！

【解决方案2】：

您可以使用sort_values + tail

df.groupby('sIP').agg({'duration': 'sum'}).sort_values('sum(duration)').tail(1)

【讨论】：

【解决方案3】：

斯卡拉

df.groupBy("ip").agg(sum("duration").as("sum_duration")).agg(max("sum_duration")).show()

PySpark

import pyspark.sql.functions as F
df.groupBy("ip").agg(F.sum("duration").alias("sum_duration")).agg(F.max("sum_duration")).show()

【讨论】：