获取 Spark 数据帧列中的最大值的最佳方法

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

Consider the following example:

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 运行时或资源使用方面，或者是否有比上述方法更直接的方法？

340545 次浏览
小开
最佳答案 
>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

答案与 method3几乎相同，但似乎可以删除 method3中的“ asDect ()”

小开

如果有人想知道如何使用 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
小开

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

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

备注: 星火计划用于大数据分布式计算。示例 DataFrame 的大小非常小，因此可以根据小示例更改实际示例的顺序。

Slowest: Method_1, because .describe("A") calculates min, max, mean, stddev, and count (5 calculations over the whole column).

媒介: Method _ 4，因为 .rdd(DF 到 RDD 转换)减慢了这个过程。

更快: Method _ 3 ~ Method _ 2 ~ Method _ 5，因为逻辑非常相似，所以 Spark 的催化剂优化器遵循非常相似的逻辑，操作次数最少(得到一个特定列的 max，收集一个单值数据框; .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

集群边缘节点上的结果(毫秒) :

small DF (ms): {'m1': 7096, 'm2': 205, 'm3': 165, 'm4': 211, 'm5': 180}

较大的 DF (ms) : {'m1': 10260, 'm2': 452, 'm3': 465, 'm4': 916, 'm5': 373}

小开

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

Considering your example:

+---+---+
|  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

Make sure you have the correct import:
from pyspark.sql.functions import max The max function we use here is the pySPark sql library function, not the default max function of python.

小开

Another way of doing it:

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.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;

您将在属性中看到统计信息—— min、 max、 different、 null 等。

小开 
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)

And the result is (3.0,6.0), which is the same to the testDataFrame.agg(max($"A"), max($"B")).collect()(0).However, testDataFrame.agg(max($"A"), max($"B")).collect()(0) returns a List, [3.0,6.0]

小开

在火花中你可以这样做:

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

下面的示例演示如何获取 Spark 数据框列中的 max 值。

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

同样，最小值、平均值等可以计算如下:

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|
+------+------+------+
小开

首先添加导入行:

from pyspark.sql.functions import min, max

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

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


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

要在数据框中找到年龄的最大值:

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


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

To just get the value use any of these

  1. df1.agg({"x": "max"}).collect()[0][0]
  2. df1.agg({"x": "max"}).head()[0]
  3. 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]
小开

我使用了这个链中已经存在的另一个解决方案(通过@satprem rath)。

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

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


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

添加更多上下文。

While the above method printed the 结果, I faced issues when assigning the result to a variable to reuse later.

因此，为了只获得分配给变量的 int值:

from pyspark.sql.functions import max, min


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

提交答案