package com.huawei.bigdata.spark.examples

import org.apache.spark.mllib.stat.Statistics

import org.apache.spark.sql.types.DoubleType

import org.apache.spark.{SparkConf, SparkContext}

/**

  * Created by wulei on 2017/8/3.

  */

object PointCorrPredict {

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

    val sparkConf = new SparkConf().setAppName("PointCorrPredict")

    val sc = new SparkContext(sparkConf)

    val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)

    sqlContext.sql("use vio_offsite")

    //360111010002,360102029001

    val dataFrame = sqlContext.sql("select kk_id,direct,day,hour,cnt,speed from kk_hour_scale").orderBy("day","hour")

    val newDataFrame = dataFrame.filter("kk_id = '3601110100'and direct = '02'")

                      .orderBy(dataFrame("day").desc,dataFrame("hour").desc).select(dataFrame.col("cnt").cast(DoubleType)).limit(100)

      .rdd.map(row=>row.getAs[Double]("cnt"))

    /*val dd =  newDataFrame.collect().take(3)

   dd.foreach(println)*/

    val destinationDataFrame = sqlContext.sql("select origin_kakou,destination_kakou from kk_relation ")

    val newDestinationDataFrame = destinationDataFrame.filter("origin_kakou = '360111010002'").select("destination_kakou").collect()

    for (i <- 0 until newDestinationDataFrame.length){

      println(newDestinationDataFrame(i))

      println(newDestinationDataFrame(i).toString().substring(1,11))

      println(newDestinationDataFrame(i).toString().substring(11,13))

      val tmpDataFrame = dataFrame.filter("kk_id = '"+ newDestinationDataFrame(i).toString().substring(1,11)

                         +"' and direct = '"+newDestinationDataFrame(i).toString().substring(11,13)+"'")

                        .orderBy(dataFrame("day").desc,dataFrame("hour").desc).select(dataFrame.col("cnt").cast(DoubleType)).limit(100)

        .rdd.map(row=>row.getAs[Double]("cnt"))

      //tmpDataFrame.foreach(row => println(row))

      var correlationPearson: Double = Statistics.corr(newDataFrame,tmpDataFrame)//计算不同数据之间的相关系数:皮尔逊

      println("\ncorrelationPearson：" + correlationPearson) //打印结果

    }

    println("11111")

  sc.stop()

  }

}