我需要为多个双数据类型列找到中值。请求建议以找到正确的方法。
下面是我的一列示例数据集。我希望我的样本的中值返回为1。
scala> sqlContext.sql("select num from test").show();
+---+
|num|
+---+
|0.0|
|0.0|
|1.0|
|1.0|
|1.0|
|1.0|
+---+
我尝试了以下选项
1) Hive UDAF百分位数，它只适用于BigInt。
2) Hive UDAT percentile_approx，但它没有按预期工作(返回0.25 vs 1)。
sqlContext.sql("从测试中选择percentile_approx(num，0.5)").show()；
+----+
| _c0|
+----+
|0.25|
+----+
3) Spark窗口函数percent_rank-要找到中值，我认为的方法是查找所有高于0.5的percent_rnk，并选择最大percent_rak的相应num值。但它并不是在所有情况下都有效，尤其是当我有记录计数时，在这种情况下，中位数是排序分布中中间值的平均值。
同样在percent_rank中，由于我必须找到多列的中值，所以我必须在不同的数据帧中计算它，这对我来说是一种不太复杂的方法。如果我的理解不正确，请纠正我。
+---+-------------+
|num|percent_rank |
+---+-------------+
|0.0|0.0|
|0.0|0.0|
|1.0|0.4|
|1.0|0.4|
|1.0|0.4|
|1.0|0.4|
+---+---+
出于好奇，您正在使用哪个版本的Apache Spark？Apache Spark 2.0+中有一些修复，其中包括对approxQuantile的更改。
如果我运行下面的pySpark代码片段：
rdd = sc.parallelize([[1, 0.0], [1, 0.0], [1, 1.0], [1, 1.0], [1, 1.0], [1, 1.0]])
df = rdd.toDF(['id', 'num'])
df.createOrReplaceTempView("df")
其中CCD_ 2计算使用CCD_ 3作为：
df.approxQuantile("num", [0.5], 0.25)
或
spark.sql("select percentile_approx(num, 0.5) from df").show()
结果是：
- Spark 2.0.0:0.25
- Spark 2.0.1：1.0
- 火花2.1.0：1.0
注意，由于这些是近似数字(通过approxQuantile)，但通常情况下，这应该很好。如果您需要确切的中位数，一种方法是使用numpy.median。下面的代码片段是根据gench对How to find The median in Apache Spark with Python Dataframe API？的SO响应为这个df示例更新的：
from pyspark.sql.types import *
import pyspark.sql.functions as F
import numpy as np
def find_median(values):
try:
median = np.median(values) #get the median of values in a list in each row
return round(float(median),2)
except Exception:
return None #if there is anything wrong with the given values
median_finder = F.udf(find_median,FloatType())
df2 = df.groupBy("id").agg(F.collect_list("num").alias("nums"))
df2 = df2.withColumn("median", median_finder("nums"))
# print out
df2.show()
输出为：
+---+--------------------+------+
| id| nums|median|
+---+--------------------+------+
| 1|[0.0, 0.0, 1.0, 1...| 1.0|
+---+--------------------+------+
更新：使用RDD的Spark 1.6 Scala版本
如果您使用的是Spark 1.6，您可以通过Eugene Zhulenev的响应使用Scala代码计算median。我如何使用Apache Spark计算准确的中值。下面是与我们的示例一起使用的修改后的代码。
import org.apache.spark.SparkContext._
val rdd: RDD[Double] = sc.parallelize(Seq((0.0), (0.0), (1.0), (1.0), (1.0), (1.0)))
val sorted = rdd.sortBy(identity).zipWithIndex().map {
case (v, idx) => (idx, v)
}
val count = sorted.count()
val median: Double = if (count % 2 == 0) {
val l = count / 2 - 1
val r = l + 1
(sorted.lookup(l).head + sorted.lookup(r).head).toDouble / 2
} else sorted.lookup(count / 2).head.toDouble
输出为：
// output
import org.apache.spark.SparkContext._
rdd: org.apache.spark.rdd.RDD[Double] = ParallelCollectionRDD[227] at parallelize at <console>:34
sorted: org.apache.spark.rdd.RDD[(Long, Double)] = MapPartitionsRDD[234] at map at <console>:36
count: Long = 6
median: Double = 1.0
注意，这是使用RDDs计算准确的中值，即您需要将DataFrame列转换为RDD来执行此计算。