Запросить результат другого запроса DataFrame Spark

Question

Использование Spark и библиотеки Pandas для Python.В основном у меня есть следующее:

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|

Есть ли способ перезаписать состояние кадра данных?Я смотрел на преобразования, но безуспешно, коллекционирование () было другой вещью, которую я изучал, но все еще ничего ...

Чего мне не хватает?

Заранее спасибо.

Answer 1

Scala

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()

Answer 2

Вам необходимо сохранить ваше первое преобразование в новом фрейме данных, в противном случае результат просто отбрасывается.

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

Затем вы можете работать с новым фреймом данных, как вам нравится, то есть:

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

Вы все равно можете сделать оба в одном преобразовании без промежуточного кадра данных:

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

Answer 3

Вы можете сделать с sort_values + tail

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