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的列，使得減號左右兩邊類型匹配。