spark经常使用功能：使用Spark计算数列统计值

参考 ：

--  https://cloud.tencent.com/developer/article/1475487

先来回顾一下数据和对应的统计结果：

本文使用的是iris分类数据集，数据下载地址为：

http://archive.ics.uci.edu/ml/datasets/Iris

下载后转换为xlsx格式的文件，数据以下：

对应的统计结果以下：

在介绍以前，我仍是想先说明一点，这一篇只是想先带你们体验一把Spark SQL，相关更多关于原理相关的知识，我们会在后面的文章中详细介绍。

一、数据导入
这里我们经过读取Excel的方式读取出相应的数据，并获得一个DataFrame：

def createDFByCSV(spark:SparkSession) = {
    val df = spark.sqlContext.read.format("com.databricks.spark.csv")
      .option("header","true") //这里若是在csv第一行有属性的话，没有就是"false"
      .option("inferSchema",true.toString)//这是自动推断属性列的数据类型。
      .load("resources/iris.csv")

    df.show()
  }

结果以下：

二、使用Spark SQL计算统计值
2.1 最大值、最小值
使用Spark SQL统计最大值或者最小值，首先使用agg函数对数据进行聚合，这个函数通常配合group by使用，不使用group by的话就至关于对全部的数据进行聚合。

随后，直接使用max和min函数就能够，想要输出多个结果的话，中间用逗号分开，而使用as给聚合后的结果赋予一个列名，至关于sql中的as：

import spark.implicits._

    df.agg(max($"feature1") as "max_feature1",
        min($"feature2") as "min_feature2")
      .show()

结果输出以下：

上面的$表明一列的意思，至关于col函数：

df.agg(max(col("feature1")) as "max_feature1",
        min(col("feature2")) as "min_feature2")
      .show()

1.2 平均值
平均值的计算使用mean函数：

df.agg(mean($"feature1") as "mean_feature1",
      mean($"feature2") as "mean_feature2").show()

输出为：

1.3 样本标准差&整体标准差
样本标准差的计算有两个函数可使用，分别是stddev函数和stddev_samp函数，而整体标准差使用stddev_pop方法。须要注意的一点是，这里和hive sql是有区别的，在hive sql中，stddev函数表明的是整体标准差，而在spark sql中，stddev函数表明的是样本标准差，能够查看一下源代码：

经过代码验证一下：

df.agg(stddev($"feature1") as "stddev_feature1",
      stddev_pop($"feature1") as "stddev_pop_feature1",
      stddev_samp($"feature1") as "stddev_samp_feature1").show()

输出结果为：

1.4 中位数
SparkSQL中也没有直接计算中位数的方法，因此咱们仍是借鉴上一篇中的思路，再来回顾一下：

计算中位数也好，计算四分位数也好，无非就是要取得两个位置嘛，假设咱们的数据从小到大排，按照一、二、三、.. 、n进行编号，当数量n为奇数时，取编号（n + 1）／2位置的数便可，当n为偶数时，取(int)（n + 1）／2位置和(int)（n + 1）／2 + 1位置的数取平均便可。但两者其实能够统一到一个公式中：

1）假设n = 149 ，(n+1)/2 = 75 ，小数部分为0，那么中位数=75位置的数 （1 - 0）+ 76位置的数 （0 - 0）
2）假设n = 150，(n+1)/2 = 75，小数部分为0.5,那么中位数=75位置的数 （1 - 0.5）+ 76位置的数 （0.5 - 0）

因此，能够把这个过程分解为三个步骤，第一步是给数字进行一个编号，spark中一样使用row_number()函数（该函数的具体用法后续再展开，这里只提供一个简单的例子），第二步是计算(n+1)/2的整数部分和小数部分，第三步就是根据公式计算中位数。

首先使用row_number()给数据进行编号：

val windowFun = Window.orderBy(col("feature3").asc)
df.withColumn("rank",row_number().over(windowFun)).show(false)

输出以下：

接下来是肯定中位数的位置，这里咱们分别拿到（n + 1）／2的整数部分和小数部分：

val median_index = df.agg(
  ((count($"feature3") + 1) / 2).cast("int") as "rank",
  ((count($"feature3") + 1) / 2 %  1) as "float_part"
)

median_index.show()

输出以下：

这里小数部分不为0，意味着咱们不只要拿到rank=75的数，还要拿到rank=76的数，咱们最好把其放到一行上，这里使用一样lead函数，lead函数的做用就是拿到分组排序后，下一个位置或下n个位置的数，我们在后面的博客中还会细讲，这里也只是抛砖引玉：

val windowFun = Window.orderBy(col("feature3").asc)
df.withColumn("next_feature3",lead(col("feature3"),1).over(windowFun)).show(false)

输出以下：

接下来，join两个表，按公式计算中位数就能够啦，完整的代码以下：

val median_index = df.agg(
  ((count($"feature3") + 1) / 2).cast("int") as "rank",
  ((count($"feature3") + 1) / 2 %  1) as "float_part"
)

val windowFun = Window.orderBy(col("feature3").asc)

df.withColumn("rank",row_number().over(windowFun))
  .withColumn("next_feature3",lead(col("feature3"),1).over(windowFun))
  .join(median_index,Seq("rank"),"inner")
  .withColumn("median" ,($"float_part" - lit(0)) * $"next_feature3" + (lit(1) - $"float_part") * $"feature3")
  .show()

输出以下：

1.5 四分位数
先来复习下四分位数的两种解法，n+1方法和n-1方法：

对于n+1方法，若是数据量为n，则四分位数的位置为：

Q1的位置= (n+1) × 0.25
Q2的位置= (n+1) × 0.5
Q3的位置= (n+1) × 0.75

对于n-1方法，若是数据量为n，则四分位数的位置为：

Q1的位置=1+（n-1）x 0.25
Q2的位置=1+（n-1）x 0.5
Q3的位置=1+（n-1）x 0.75

这里的思路和求解中位数是同样的，咱们分别实现一下两种方法，首先是n+1方法：

val q1_index = df.agg(
  ((count($"feature3") + 1) * 0.25).cast("int") as "rank",
  ((count($"feature3") + 1) * 0.25 %  1) as "float_part"
)

val windowFun = Window.orderBy(col("feature3").asc)

df.withColumn("rank",row_number().over(windowFun))
  .withColumn("next_feature3",lead(col("feature3"),1).over(windowFun))
  .join(q1_index,Seq("rank"),"inner")
  .withColumn("q1" ,($"float_part" - lit(0)) * $"next_feature3" + (lit(1) - $"float_part") * $"feature3")
  .show()

输出为：

接下来是n-1方法：

val q1_index = df.agg(
  ((count($"feature3") - 1) * 0.25).cast("int") + 1 as "rank",
  ((count($"feature3") - 1) * 0.25 %  1) as "float_part"
)

val windowFun = Window.orderBy(col("feature3").asc)

df.withColumn("rank",row_number().over(windowFun))
  .withColumn("next_feature3",lead(col("feature3"),1).over(windowFun))
  .join(q1_index,Seq("rank"),"inner")
  .withColumn("q1" ,($"float_part" - lit(0)) * $"next_feature3" + (lit(1) - $"float_part") * $"feature3")
  .show()

输出为：

参考。

三、踩坑总结
在计算中位数或者四分位数时，我一开始的写法以下：

很奇怪的一点是，$"float_part" - 0没有报错，1 - $"float_part"却报错了，报的错误是：

看这里你们应该明白了，$"float_part" - 0中，减号左右两边的数据都应该是列名，与$"float_part" 类型相同，可是1 - $"float_part"两边都应该是个数字，与1的类型相同，因此后面一个报错了。

所以修改的方法是：

使用lit方法建立了一个全为0或者全为1的列，使得减号左右两边类型匹配。