Spark 分布式實(shí)現(xiàn)距離判別分析

距離判別

設(shè)有兩個(gè)總體G1檬果，G2朝扼，從第一個(gè)總體抽取n個(gè)樣本，從第二個(gè)總體中抽取m個(gè)樣本量愧，每個(gè)樣本都有p個(gè)測量指標(biāo)淫半。取任一樣本實(shí)測指標(biāo)為X=(x1,x2,……xp)'溃槐，分別計(jì)算X到兩個(gè)總體的距離D1，D2,按距離最近準(zhǔn)則判別歸類科吭。
距離計(jì)算公式（馬氏距離）：

在這里插入圖片描述

判別分析

數(shù)據(jù)展示與說明

在這里插入圖片描述

某商場從市場隨機(jī)抽取20中品牌的電視機(jī)進(jìn)行調(diào)查昏滴，其中13中暢銷，7種滯銷对人。按照電視機(jī)的質(zhì)量評分影涉、功能評分、價(jià)格手機(jī)資料规伐。其中“1”表示暢銷蟹倾，“2”表示滯銷，根據(jù)該樣本建立判別函數(shù)，對以后的新樣本進(jìn)行評測鲜棠。

實(shí)現(xiàn)過程

首先也是要自定義一個(gè)計(jì)算樣本均值向量的自定義聚合函數(shù)肌厨，同上一篇

然后按照公式進(jìn)行計(jì)算

def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName(s"${this.getClass.getSimpleName}")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    val sc = spark.sparkContext

    val irisData = spark.read
      .option("header", true)
      .option("inferSchema", true)
      .csv("F:\\DataSource\\dda.txt")

    val schema = irisData.schema
    val fts = schema.filterNot(_.name == "class").map(_.name).toArray

    val amountVectorAssembler: VectorAssembler = new VectorAssembler()
      .setInputCols(fts)
      .setOutputCol("features")

    val vec2Array = udf((vec: DenseVector) => vec.toArray)

    val irisFeatrus = amountVectorAssembler
      .transform(irisData)
      .select($"class", vec2Array($"features") as "features")

    val ui = spark.udf.register("udafMedian", new meanVector(fts.length))

    // 計(jì)算樣本均值向量
    val uiGroup = irisFeatrus
      .groupBy($"class")
      .agg(ui($"features") as "ui", count($"class") as "len")

    // 類別、協(xié)方差矩陣豁陆、均值向量
    val covMatrix = irisFeatrus
      .join(uiGroup, "class")
      .rdd
      .map(row => {
        val lable = row.getAs[String]("class")
        val len = row.getAs[Long]("len")
        val u = densevec(row.getAs[Seq[Double]]("ui").toArray)
        val x = densevec(row.getAs[Seq[Double]]("features").toArray)
        val denseMatrix = (x - u).toDenseMatrix
        lable -> (denseMatrix, u, len)
      })
      .reduceByKey((d1, d2) => {
        (DenseMatrix.vertcat(d1._1, d2._1), d1._2, d1._3)
      })
      .map(tp => {
        val len = tp._2._3 - 1
        val t: DenseMatrix[Double] = (tp._2._1.t * tp._2._1).map(x => x / len)
        (tp._1, t, tp._2._2)
      })

    val covmBroad = sc.broadcast(covMatrix.collect())

    // 定義判別函數(shù)
    def dfunction(vec: Seq[Double]) = {
      covmBroad.value
        .map(tp => {
          val xui = (densevec(vec.toArray) - tp._3).toDenseMatrix
          val d = (xui * inv(tp._2) * xui.t).data.head
          (d, tp._1)
        })
        .minBy(_._1)
        ._2
    }
    val nGudf = udf((vec: Seq[Double]) => dfunction(vec))


  val predictions =   irisFeatrus
      .withColumn("nG", nGudf($"features"))

    predictions.show(truncate = false)

    spark.stop()
  }

結(jié)果查看：從結(jié)果可以看到柑爸，僅有一列判別錯(cuò)誤

在這里插入圖片描述

參考資料：

《多元統(tǒng)計(jì)分析及R語言建模》 – 王斌會