得到的数据框星火列中的最大值的最佳方法

问题描述 投票：41回答：10

我试图找出让在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").collect()[0].asDict()['A'])

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

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

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

上述每一个给出了正确的答案，但在没有火花的分析工具的我不能告诉这是最好的。

无论从直觉或经验上的上述方法是最有效的星火运行或资源使用方面的任何想法，或是否有比上面的那些更直接的方法是什么？

python apache-spark apache-spark-sql pyspark spark-dataframe
10个回答
47
投票
>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。但似乎在方法3的“asDict（）”可以被删除

0
投票
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).However，testDataFrame.agg(max($"A"), max($"B")).collect()(0)返回一个List，[3.0,6.0]

16
投票

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

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

12
投票

备注：星火打算在大数据工作 - 分布式计算。的示例数据帧的大小是非常小的，所以实际的例子的顺序可以相对于被改变以小〜例子。

最慢：Method_1，因为.describe（ “A”）计算最小值，最大值，平均值，STDDEV和count（5个计算在整个列）

适用介质：方法4，因为.rdd（DF到RDD变换）减慢的过程。

更快：Method_3〜Method_2〜method_5，因为逻辑非常相似，所以斯巴克的催化剂优化与如下操作的最小数量非常类似的逻辑（获取特定列的最大值，收集单值数据帧）; （.asDict（）增加了一些额外的时间比较3,2〜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 4: Convert to RDD
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}

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

他们都给出相同的答案

3
投票

（使用星火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
2
投票

下面的例子演示了如何获得一个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

同样分钟，平均等可被计算如下：

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|
+------+------+------+
1
投票

我认为最好的解决方案将使用head()

考虑到你的例子：

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

使用AGG和python的最大方法，我们可以得到价值如下： from pyspark.sql.functions import max df.agg(max(df.A)).head()[0]

这将返回：3.0

请确保您有正确的进口： from pyspark.sql.functions import max我们这里使用的最大功能是pySPark SQL库函数，而不是Python的默认最大功能。

1
投票

在pyspark你可以这样做：

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

这里是这样，通过只是在做计算统计的懒办法：

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;

你会看到在属性的统计数据 - 最小，最大，层次分明，空，等

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.