使用Spark下的corr计算皮尔森相似度Pearson时，报错Can only zip RDDs with same number of elements in each partition....

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()
  }
}