【问题标题】：Best way to get the max value in a Spark dataframe column在 Spark 数据框列中获取最大值的最佳方法
【发布时间】：2016-01-18 09:49:42
【问题描述】：

我正在尝试找出在 Spark 数据框列中获取最大值的最佳方法。

考虑以下示例：

df = spark.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
df.show()

创建：

+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

我的目标是在 A 列中找到最大值（通过检查，这是 3.0）。使用 PySpark，我可以想到以下四种方法：

# Method 1: Use describe()
float(df.describe("A").filter("summary = 'max'").select("A").first().asDict()['A'])

# Method 2: Use SQL
df.registerTempTable("df_table")
spark.sql("SELECT MAX(A) as maxval FROM df_table").first().asDict()['maxval']

# Method 3: Use groupby()
df.groupby().max('A').first().asDict()['max(A)']

# Method 4: Convert to RDD
df.select("A").rdd.max()[0]

上面的每一个都给出了正确的答案，但是在没有 Spark 分析工具的情况下，我无法判断哪个是最好的。

任何来自直觉或经验的想法，关于上述哪种方法在 Spark 运行时或资源使用方面最有效，或者是否有比上述方法更直接的方法？

【问题讨论】：

方法 2 和 3 是等效的，并且使用相同的物理和优化的逻辑计划。方法 4 在 rdd 上应用 reduce with max。它可能比直接在 DataFrame 上操作要慢。方法 1 或多或少等同于 2 和 3。
@zero323 df.select(max("A")).collect()[0].asDict()['max(A)'] 怎么样？看起来等同于方法 2，但更紧凑，也比方法 3 更直观。
- 最慢的是方法4，因为你将整个列进行DF到RDD的转换，然后提取最大值；

标签： python apache-spark pyspark apache-spark-sql

【解决方案1】：

备注：Spark 旨在用于大数据 - 分布式计算。示例 DataFrame 的大小非常小，因此可以相对于小示例更改实际示例的顺序。

最慢：Method_1，因为.describe("A") 计算最小值、最大值、平均值、标准差和计数（整列计算 5 次）。

中：Method_4，因为.rdd（DF 到 RDD 的转换）会减慢进程。

Faster：Method_3 ~ Method_2 ~ Method_5，因为逻辑非常相似，所以 Spark 的催化剂优化器遵循非常相似的逻辑，操作次数最少（获取特定列的最大值，收集单值数据帧；.asDict() 补充加时赛比较 2、3 和 5）

import pandas as pd
import time

time_dict = {}

dfff = self.spark.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
#--  For bigger/realistic dataframe just uncomment the following 3 lines
#lst = list(np.random.normal(0.0, 100.0, 100000))
#pdf = pd.DataFrame({'A': lst, 'B': lst, 'C': lst, 'D': lst})
#dfff = self.sqlContext.createDataFrame(pdf)

tic1 = int(round(time.time() * 1000))
# Method 1: Use describe()
max_val = float(dfff.describe("A").filter("summary = 'max'").select("A").collect()[0].asDict()['A'])
tac1 = int(round(time.time() * 1000))
time_dict['m1']= tac1 - tic1
print (max_val)

tic2 = int(round(time.time() * 1000))
# Method 2: Use SQL
dfff.registerTempTable("df_table")
max_val = self.sqlContext.sql("SELECT MAX(A) as maxval FROM df_table").collect()[0].asDict()['maxval']
tac2 = int(round(time.time() * 1000))
time_dict['m2']= tac2 - tic2
print (max_val)

tic3 = int(round(time.time() * 1000))
# Method 3: Use groupby()
max_val = dfff.groupby().max('A').collect()[0].asDict()['max(A)']
tac3 = int(round(time.time() * 1000))
time_dict['m3']= tac3 - tic3
print (max_val)

tic4 = int(round(time.time() * 1000))
# Method 4: Convert to RDD
max_val = dfff.select("A").rdd.max()[0]
tac4 = int(round(time.time() * 1000))
time_dict['m4']= tac4 - tic4
print (max_val)

tic5 = int(round(time.time() * 1000))
# Method 5: Use agg()
max_val = dfff.agg({"A": "max"}).collect()[0][0]
tac5 = int(round(time.time() * 1000))
time_dict['m5']= tac5 - tic5
print (max_val)

print time_dict

集群边缘节点上的结果，以毫秒 (ms) 为单位：

小DF（毫秒）：{'m1': 7096, 'm2': 205, 'm3': 165, 'm4': 211, 'm5': 180}

更大的 DF（毫秒）：{'m1': 10260, 'm2': 452, 'm3': 465, 'm4': 916, 'm5': 373}

【讨论】：

【解决方案2】：

要获得价值，请使用这些中的任何一个

df1.agg({"x": "max"}).collect()[0][0]
df1.agg({"x": "max"}).head()[0]
df1.agg({"x": "max"}).first()[0]

或者，我们可以为 'min' 做这些

from pyspark.sql.functions import min, max
df1.agg(min("id")).collect()[0][0]
df1.agg(min("id")).head()[0]
df1.agg(min("id")).first()[0]

【讨论】：

【解决方案3】：

我使用了这个链中已经存在的另一种解决方案（@satprem Rath）。

要找到数据框中年龄的最小值：

df.agg(min("age")).show()

+--------+
|min(age)|
+--------+
|      29|
+--------+

编辑：添加更多上下文。

虽然上述方法打印了 result，但在将结果分配给变量以供以后重用时遇到了问题。

因此，仅获取分配给变量的 int 值：

from pyspark.sql.functions import max, min  

maxValueA = df.agg(max("A")).collect()[0][0]
maxValueB = df.agg(max("B")).collect()[0][0]

【讨论】：

请围绕您的解决方案添加一些上下文和解释。

【解决方案4】：

首先添加导入行：

from pyspark.sql.functions import min, max

查找数据框中年龄的最小值：

df.agg(min("age")).show()

+--------+
|min(age)|
+--------+
|      29|
+--------+

查找数据框中年龄的最大值：

df.agg(max("age")).show()

+--------+
|max(age)|
+--------+
|      77|
+--------+

【讨论】：

【解决方案5】：

以下示例显示了如何获取 Spark 数据框列中的最大值。

from pyspark.sql.functions import max

df = sql_context.createDataFrame([(1., 4.), (2., 5.), (3., 6.)], ["A", "B"])
df.show()
+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

result = df.select([max("A")]).show()
result.show()
+------+
|max(A)|
+------+
|   3.0|
+------+

print result.collect()[0]['max(A)']
3.0

类似min、mean等可以计算如下：

from pyspark.sql.functions import mean, min, max

result = df.select([mean("A"), min("A"), max("A")])
result.show()
+------+------+------+
|avg(A)|min(A)|max(A)|
+------+------+------+
|   2.0|   1.0|   3.0|
+------+------+------+

【讨论】：

同意。我是 pyspark 的新手（对 Python 来说是旧的），这更直观。
扩展这个答案 - 如果你有 NaN，以下将起作用：df.select('A').dropna().select([max('A')])

【解决方案6】：

在 pyspark 中你可以这样做：

max(df.select('ColumnName').rdd.flatMap(lambda x: x).collect())

【讨论】：

【解决方案7】：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val testDataFrame = Seq(
  (1.0, 4.0), (2.0, 5.0), (3.0, 6.0)
).toDF("A", "B")

val (maxA, maxB) = testDataFrame.select(max("A"), max("B"))
  .as[(Double, Double)]
  .first()
println(maxA, maxB)

结果是(3.0,6.0)，和testDataFrame.agg(max($"A"), max($"B")).collect()(0)一样。但是testDataFrame.agg(max($"A"), max($"B")).collect()(0)返回一个List，[3.0,6.0]

【讨论】：

【解决方案8】：

我相信最好的解决方案是使用head()

考虑你的例子：

+---+---+
|  A|  B|
+---+---+
|1.0|4.0|
|2.0|5.0|
|3.0|6.0|
+---+---+

使用python的agg和max方法我们可以得到如下值：

from pyspark.sql.functions import max df.agg(max(df.A)).head()[0]

这将返回： 3.0

确保您有正确的导入：
from pyspark.sql.functions import max 我们这里使用的max函数是pySPark sql库函数，不是python默认的max函数。

【讨论】：

确保您有正确的导入，您需要导入以下内容：from pyspark.sql.functions import max 我们在这里使用的最大值是 pySpark sql 函数而不是 python 最大值如果您使用别名会更好from pyspark.sql.functions import max as mx

【解决方案9】：

这是一种懒惰的方法，只需进行计算统计：

df.write.mode("overwrite").saveAsTable("sampleStats")
Query = "ANALYZE TABLE sampleStats COMPUTE STATISTICS FOR COLUMNS " + ','.join(df.columns)
spark.sql(Query)

df.describe('ColName')

或

spark.sql("Select * from sampleStats").describe('ColName')

或者你可以打开一个蜂巢壳然后

describe formatted table sampleStats;

您将在属性中看到统计信息 - 最小值、最大值、不同值、空值等。

【讨论】：

【解决方案10】：

另一种方法：

df.select(f.max(f.col("A")).alias("MAX")).limit(1).collect()[0].MAX

根据我的数据，我得到了以下基准：

df.select(f.max(f.col("A")).alias("MAX")).limit(1).collect()[0].MAX
CPU times: user 2.31 ms, sys: 3.31 ms, total: 5.62 ms
Wall time: 3.7 s

df.select("A").rdd.max()[0]
CPU times: user 23.2 ms, sys: 13.9 ms, total: 37.1 ms
Wall time: 10.3 s

df.agg({"A": "max"}).collect()[0][0]
CPU times: user 0 ns, sys: 4.77 ms, total: 4.77 ms
Wall time: 3.75 s

他们都给出了相同的答案

【讨论】：

“df.limit(1).collect()[0]”可以替换为“df.first()”

【解决方案11】：

数据框的特定列的最大值可以通过使用 -

来实现

your_max_value = df.agg({"your-column": "max"}).collect()[0][0]

【讨论】：

我更喜欢您的解决方案而不是公认的解决方案。添加两个“[0]”只给出结果

【解决方案12】：

如果有人想知道如何使用 Scala（使用 Spark 2.0.+）来做到这一点，请看这里：

scala> df.createOrReplaceTempView("TEMP_DF")
scala> val myMax = spark.sql("SELECT MAX(x) as maxval FROM TEMP_DF").
    collect()(0).getInt(0)
scala> print(myMax)
117

【讨论】：

【解决方案13】：

>df1.show()
+-----+--------------------+--------+----------+-----------+
|floor|           timestamp|     uid|         x|          y|
+-----+--------------------+--------+----------+-----------+
|    1|2014-07-19T16:00:...|600dfbe2| 103.79211|71.50419418|
|    1|2014-07-19T16:00:...|5e7b40e1| 110.33613|100.6828393|
|    1|2014-07-19T16:00:...|285d22e4|110.066315|86.48873585|
|    1|2014-07-19T16:00:...|74d917a1| 103.78499|71.45633073|

>row1 = df1.agg({"x": "max"}).collect()[0]
>print row1
Row(max(x)=110.33613)
>print row1["max(x)"]
110.33613

答案与方法3几乎相同。但似乎method3中的“asDict（）”可以删除

【讨论】：

谁能解释一下为什么需要collect()[0]？
@jibiel collect() 返回一个列表（在这种情况下只有一个项目），因此您需要访问列表中的第一个（唯一）项目
@Burt head() 如果collect()[0] 可以使用。
@Aliaxander 这有点长。不再安装代码和 Spark。
虽然.collect()[0] 有效，但使用.first()[0] 可能更安全。根据定义，collect() 将“在驱动程序中将数据集的所有元素作为数组返回。”，这是一台机器。如果语法错误，最终可能会使用过多的内存。