package internet

import org.apache.spark.mllib.clustering.{KMeansModel, KMeans}
import org.apache.spark.mllib.linalg.{Vectors,Vector}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, 
 
SparkConf}

/**
  * Created by 汪本成 on 2016/7/24.
  */
object CheckAll {

  def main(args: Array[String]) {
    //建立入口物件
val conf = new SparkConf().setAppName("CheckAll").setMaster("local")
    val sc= new SparkContext(conf)
    val HDFS_DATA_PATH = "hdfs://node1:9000/user/spark/sparkLearning/cluster/kddcup.data"
val 
 
rawData = sc.textFile(HDFS_DATA_PATH)

    /** 分類統計樣本，降序排序 **/
//    clusteringTake1(rawData)
/** 評價k值 **/
//    clusteringTake2(rawData)
//    clusteringTake3(rawData)
//    clusteringTake4(rawData)
//    clusteringTake5(rawData)
/** R資料視覺化 **/
    /** 異常檢測 **/
var beg = System.currentTimeMillis()
    anomalies(rawData)
    var 
 
end = System.currentTimeMillis()
    println("用時：" + (end - beg) / 1000 + "s")
  }



  //Clustering，Task1
def clusteringTake1(rawData: RDD[String]) = {
    //分類統計樣本個數，降序排序
rawData.map(_.split(",").last).countByValue().toSeq.sortBy(_._2).reverse.foreach(println)


    val labelsAndData = rawData.map {
      line =>
        //將csv格式的行拆分成列，建立一個buffer，是一個可變列表
val buffer = line.split(",").toBuffer
        //刪除下標從1開始的三個類別型列
buffer.remove(1, 3)
        //刪除下標最後的標號列
val label = buffer.remove(buffer.length - 1)
        //保留其他值並將其轉換成一個數值型(Double型物件)陣列
val vector = Vectors.dense(buffer.map(_.toDouble).toArray)
        //將陣列和標號組成一個元祖
(label, vector)
    }

    /**
      * 為啥要進行labelsAndData => data轉化?
      * 1、k均值在執行過程中只用到特徵向量(即沒有用到資料集的目標標號列)
      * 2、使data這個RDD只包含元祖的只包含元組的第二個元素
* 3、實現2可以通過元組型別RDD的values屬性得到，在放入快取中，減少落地
*/
//提取出元組的特徵向量
val data = labelsAndData.values.cache()

    //例項化Kmeans類物件
val kmeans = new KMeans()
    //建立KMeansModel
val model = kmeans.run(data)
    //輸出每個簇的質心
model.clusterCenters.foreach(println)

    val clusterLabelCount = labelsAndData.map {
      case (label, datum) =>
        //預測樣本datum的分類cluster
val cluster = model.predict(datum)
        //返回類別-簇的元組
(cluster, label)
    }.countByValue()

    //對簇-類別對分別進行計數，並以可讀方式輸出
clusterLabelCount.toSeq.sorted.foreach {
      case ((cluster, label), count) =>
        println(f"$cluster%1s$label%18s$count%8s")
    }
    data.unpersist()
  }

  /**
    * 歐氏距離公式
* a.toArray.zip(b.toArray)對應 "兩個向量相應元素"
    * map(p => p._1 - p._2)對應 "差"
    * map(d => d*d).sum對應 "平方和"
    * math.sqrt()對應 "平方根"
    * @param a
* @param b
* @return
*/
def distance(a: Vector, b: Vector) =
    math.sqrt(a.toArray.zip(b.toArray).map(p => p._1 - p._2).map(d => d * d).sum)

  /**
    * 歐氏距離公式應用到model中
* KMeansModel.predict方法中呼叫了KMeans物件的findCloest方法
* @param datum
* @param model
* @return
*/
def distToCenter(datum: Vector, model: KMeansModel) = {
    //預測樣本datum的分類cluster
val cluster = model.predict(datum)
    //計算質心
val center = model.clusterCenters(cluster)
    //應用距離公式
distance(center, datum)
  }

  /**
    * 平均質心距離
* @param data
* @param k
* @return
*/
def clusteringScore(data: RDD[Vector], k: Int): Double = {
    val kmeans = new KMeans()
    //設定k值
kmeans.setK(k)
    //建立KMeansModel
val model = kmeans.run(data)
    //計算k值model平均質心距離，mean()是平均函式
data.map(datum => distToCenter(datum, model)).mean()
  }

  /**
    * 平均質心距離優化
* @param data
* @param k
* @param run   執行次數
* @param epsilon  閾值
* @return
*/
def clusteringScore2(data: RDD[Vector], k: Int, run: Int, epsilon: Double): Double = {
    val kmeans = new KMeans()
    kmeans.setK(k)
    //設定k的執行次數
kmeans.setRuns(run)
    //設定閾值
kmeans.setEpsilon(epsilon)
    val model = kmeans.run(data)
    data.map(datum => distToCenter(datum, model)).mean()
  }

  //Clustering，Take2
def clusteringTake2(rawData: RDD[String]): Unit ={
    val data = rawData.map {
      line =>
        val buffer = line.split(",").toBuffer
        buffer.remove(1, 3)
        buffer.remove(buffer.length - 1)
        Vectors.dense(buffer.map(_.toDouble).toArray)
    }.cache()

    val run = 10
val epsilon = 1.0e-4
//在(5,30)區間內以5為等差數列數值不同k值對其評分
(5 to 30 by 5).map(k => (k, clusteringScore(data, k))).foreach(println)
    //在(20,120)區間內以10為等差數列數值不同k值對其評分
(30 to 100 by 10).par.map(k => (k, clusteringScore2(data, k, run, epsilon))).foreach(println)

    data.unpersist()
  }


  /**
    * 加工出R視覺化資料存入HDFS中
* @param rawData
* @param k
* @param run
* @param epsilon
*/
def visualizationInR(rawData: RDD[String], k: Int, run: Int, epsilon: Double): Unit ={
    val data = rawData.map {
      line =>
        val buffer = line.split(",").toBuffer
        buffer.remove(1, 3)
        buffer.remove(buffer.length - 1)
        Vectors.dense(buffer.map(_.toDouble).toArray)
    }.cache()

    val kmeans = new KMeans()
    kmeans.setK(k)
    kmeans.setRuns(run)
    kmeans.setEpsilon(epsilon)
    val model = kmeans.run(data)

    val sample = data.map(
      datum =>
        model.predict(datum) + "," + datum.toArray.mkString(",")
    ).sample(false, 0.05)   //選擇了5%行
sample.saveAsTextFile("hdfs://nodel:9000/user/spark/R/sample")
    data.unpersist()
  }

  /**
    *
    * @param data
* @return
*/
def buildNormalizationFunction(data: RDD[Vector]): (Vector => Vector) = {
    //將陣列緩衝為Array
val dataAsArray = data.map(_.toArray)
    //資料集第一個元素的長度
val numCols = dataAsArray.first().length
    //返回資料集的元素個數
val n = dataAsArray.count()
    //兩個陣列對應元素相加求和
val sums = dataAsArray.reduce((a, b) => a.zip(b).map(t => t._1 + t._2))
    //將RDD聚合後進行求平方和操作
val sumSquares = dataAsArray.aggregate(new Array[Double](numCols))(
      (a, b) => a.zip(b).map(t => t._1 + t._2 * t._2),
(a, b) => a.zip(b).map(t => t._1 + t._2)
    )

    /** zip函式將傳進來的兩個引數中相應位置上的元素組成一個pair陣列。
* 如果其中一個引數元素比較長，那麼多餘的引數會被刪掉。
* 個人理解就是讓兩個數組裡面的元素一一對應進行某些操作
*/
val stdevs = sumSquares.zip(sums).map {
      case (sumSq, sum) => math.sqrt(n * sumSq - sum * sum) / n
    }
    val means = sums.map(_ / n)

    (datum : Vector) => {
      val normalizedArray = (datum.toArray, means, stdevs).zipped.map(
        (value, mean, stdev) =>
          if(stdev <= 0) (value- mean) else (value - mean) /stdev
      )
      Vectors.dense(normalizedArray)
    }
  }


  //clustering，Task3
def clusteringTake3(rawData: RDD[String]): Unit ={
    val data = rawData.map { line =>
      val buffer = line.split(',').toBuffer
      buffer.remove(1, 3)
      buffer.remove(buffer.length - 1)
      Vectors.dense(buffer.map(_.toDouble).toArray)
    }

    val run = 10
val epsilon = 1.0e-4
val normalizedData = data.map(buildNormalizationFunction(data)).cache()

    (60 to 120 by 10).par.map(
      k => (k, clusteringScore2(normalizedData, k, run, epsilon))
    ).toList.foreach(println)

    normalizedData.unpersist()
  }

  /**
    * 基於one-hot編碼實現類別型變數替換邏輯
* @param rawData
* @return
*/
def buildCategoricalAndLabelFunction(rawData: RDD[String]): (String => (String, Vector))  = {
    val splitData = rawData.map(_.split(","))
    //建立三個特徵
val protocols = splitData.map(_(1)).distinct().collect().zipWithIndex.toMap   //特徵值是1，0，0
val services = splitData.map(_(2)).distinct().collect().zipWithIndex.toMap    //特徵值是0，1，0
val tcpStates = splitData.map(_(3)).distinct().collect().zipWithIndex.toMap   //特徵值是0，0，1
    //
(line: String) => {
      val buffer = line.split(",").toBuffer
      val protocol = buffer.remove(1)
      val service = buffer.remove(1)
      val tcpState = buffer.remove(1)
      val label = buffer.remove(buffer.length - 1)
      val vector = buffer.map(_.toDouble)

      val newProtocolFeatures = new Array[Double](protocols.size)
      newProtocolFeatures(protocols(protocol)) = 1.0
val newServiceFeatures = new Array[Double](services.size)
      newServiceFeatures(services(service)) = 1.0
val newTcpStateFeatures = new Array[Double](tcpStates.size)
      newTcpStateFeatures(tcpStates(tcpState)) = 1.0
vector.insertAll(1, newTcpStateFeatures)
      vector.insertAll(1, newServiceFeatures)
      vector.insertAll(1, newProtocolFeatures)

      (label, Vectors.dense(vector.toArray))
    }
  }


  //Clustering，Task4
def clusteringTake4(rawData: RDD[String]): Unit ={
    val paraseFunction = buildCategoricalAndLabelFunction(rawData)
    val data = rawData.map(paraseFunction).values
    val normalizedData = data.map(buildNormalizationFunction(data)).cache()

    val run = 10
val epsilon = 1.0e-4
(80 to 160 by 10).map(
      k=> (k, clusteringScore2(normalizedData, k, run, epsilon))
    ).toList.foreach(println)

    normalizedData.unpersist()
  }


  //Clustering, Task5
/**
    * 對各個簇的熵加權平均，將結果作為聚類得分
* @param counts
* @return
*/
def entropy(counts: Iterable[Int]) = {
    val values = counts.filter(_ > 0)
    val n: Double = values.sum
    values.map {
      v =>
        val p = v / n
        -p * math.log(p)
    }.sum
  }

  /**
    * 計算熵的加權平均
* @param normalizedLabelsAndData
* @param k
* @param run
* @param epsilon
* @return
*/
def clusteringScore3(normalizedLabelsAndData: RDD[(String, Vector)], k: Int, run: Int, epsilon: Double) = {
    val kmeans = new KMeans()
    kmeans.setK(k)
    kmeans.setRuns(run)
    kmeans.setEpsilon(epsilon)

    //建立KMeansModel
val model = kmeans.run(normalizedLabelsAndData.values)
    //對每個資料集預測簇類別
val labelAndClusters = normalizedLabelsAndData.mapValues(model.predict)
    //將RDD[(String, Vector)]  => RDD[(String, Vector)],即swap Keys / Values，對換鍵和值
val clustersAndLabels = labelAndClusters.map(_.swap)
    //按簇提取標號集合
val labelsInCluster = clustersAndLabels.groupByKey().values
    //計算所有集合中有多少標籤(label)，即標號的出現次數
val labelCounts = labelsInCluster.map(_.groupBy(l => l).map(_._2.size))
    //通過類別大小來反映平均資訊量，即熵
val n = normalizedLabelsAndData.count()
    //根據簇大小計算熵的加權平均
labelCounts.map(m => m.sum * entropy(m)).sum() / n
  }


  def clusteringTake5(rawData: RDD[String]): Unit ={
    val parseFunction = buildCategoricalAndLabelFunction(rawData)
    val labelAndData = rawData.map(parseFunction)
    val normalizedLabelsAndData = labelAndData.mapValues(buildNormalizationFunction(labelAndData.values)).cache()

    val run = 10
val epsilon = 1.0e-4
(80 to 160 by 10).map(
      k => (k, clusteringScore3(normalizedLabelsAndData, k, run, epsilon))
    ).toList.foreach(println)

    normalizedLabelsAndData.unpersist()
  }


  //Detect anomalies(發現異常)
def bulidAnomalyDetector(data: RDD[Vector], normalizeFunction: (Vector => Vector)): (Vector => Boolean) = {
    val normalizedData = data.map(normalizeFunction)
    normalizedData.cache()

    val kmeans = new KMeans()
    kmeans.setK(150)
    kmeans.setRuns(10)
    kmeans.setEpsilon(1.0e-6)
    val model = kmeans.run(normalizedData)

    normalizedData.unpersist()

    //度量新資料點到最近簇質心的距離
val distances = normalizedData.map(datum => distToCenter(datum, model))
    //設定閥值為已知資料中離中心點最遠的第100個點到中心的距離
val threshold = distances.top(100).last

    //檢測，若超過該閥值就為異常點
(datum: Vector) => distToCenter(normalizeFunction(datum), model) > threshold
  }

  /**
    * 異常檢測
* @param rawData
*/
def anomalies(rawData: RDD[String]) = {
    val parseFunction = buildCategoricalAndLabelFunction(rawData)
    val originalAndData = rawData.map(line => (line, parseFunction(line)._2))
    val data = originalAndData.values
    val normalizeFunction = buildNormalizationFunction(data)
    val anomalyDetector = bulidAnomalyDetector(data, normalizeFunction)
    val anomalies = originalAndData.filter {
      case (original, datum) => anomalyDetector(datum)
    }.keys
    //取10個異常點打印出來
anomalies.take(10).foreach(println)
  }
}