title: Spark job returns a different result on each run
body: <p>I am working on a scala code which performs Linear Regression on certain datasets. Right now I am using 20 cores and 25 executors and everytime I run a Spark job I get a different result.</p>
  
  <p>The input size of the files are 2GB and 400 MB.However, when I run the job with 20 cores and 1 executor, I get consistent results.</p>
  
  <p>Has anyone experienced such a thing so far?</p>
  
  <p>Please find the code below:</p>
  
  <pre><code>import org.apache.spark.SparkContext
  import org.apache.spark.SparkContext._
  import org.apache.spark.SparkConf
  import org.apache.spark.sql.SQLContext
  import org.apache.spark.rdd.RDD
  import org.apache.spark.sql.SchemaRDD
  import org.apache.spark.Partitioner
  import org.apache.spark.storage.StorageLevel
  
  object TextProcess{
    def main(args: Array[String]){
              val conf = new SparkConf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
              val sc = new SparkContext(conf)
              val sqlContext = new org.apache.spark.sql.SQLContext(sc)
              val numExecutors=(conf.get("spark.executor.instances").toInt)
              // Read the 2 input files
              // First file is either cases / controls
              val input1 = sc.textFile(args(0))
              // Second file is Gene Expression
              val input2 = sc.textFile(args(1))
  
                //collecting header information
              val header1=sc.parallelize(input1.take(1))
              val header2=sc.parallelize(input2.take(1))
  
              //mapping data without the header information
              val map1 = input1.subtract(header1).map(x =&gt; (x.split(" ")(0)+x.split(" ")(1), x))
              val map2 = input2.subtract(header2).map(x =&gt; (x.split(" ")(0)+x.split(" ")(1), x))
  
  
              //joining data. here is where the order was getting affected. 
              val joinedMap = map1.join(map2)
  
              //adding the header back to the top of RDD
              val x = header1.union(joinedMap.map{case(x,(y,z))=&gt;y})
  
              val y = header2.union(joinedMap.map{case(x,(y,z))=&gt;z})
  
              //removing irrelevant columns
              val rddX = x.map(x=&gt;x.split(" ").drop(3)).zipWithIndex.map{case(a,b)=&gt; a.map(x=&gt;b.toString+" "+x.toString)}
              val rddY = y.map(x=&gt;x.split(" ").drop(2)).zipWithIndex.map{case(a,b)=&gt; a.map(x=&gt;b.toString+" "+x.toString)}
  
  
              //transposing and cross joining data. This keeps the identifier at the start
              val transposedX = rddX.flatMap(x =&gt; x.zipWithIndex.map(x=&gt;x.swap)).reduceByKey((a,b)=&gt; a+":"+b).map{case(a,b)=&gt;b.split(":").sorted}
              val transposedY = rddY.flatMap(x =&gt; x.zipWithIndex.map(x=&gt;x.swap)).reduceByKey((a,b)=&gt; a+":"+b).map{case(a,b)=&gt;b.split(":").sorted}.persist(StorageLevel.apply(false, true, false, false, numExecutors))
  
              val cleanedX = transposedX.map(x=&gt;x.map(x=&gt;x.slice(x.indexOfSlice(" ")+1,x.length)))
              val cleanedY = transposedY.map(x=&gt;x.map(x=&gt;x.slice(x.indexOfSlice(" ")+1,x.length))).persist(StorageLevel.apply(false, true, false, false, numExecutors))
  
  
              val cartXY = cleanedX.cartesian(cleanedY)
              val finalDataSet= cartXY.map{case(a,b)=&gt;a zip b} 
              //convert to key value pair
              val regressiondataset = finalDataSet.map(x=&gt;(x(0),x.drop(1).filter{case(a,b)=&gt; a!="NA" &amp;&amp; b!="NA" &amp;&amp; a!="null" &amp;&amp; b!="null"}.map{case(a,b)=&gt; (a.toDouble, b.toDouble)}))
  
  
              val linearOutput = regressiondataset.map(s =&gt; new LinearRegression(s._1 ,s._2).outputVal)
  
              linearOutput.saveAsTextFile(args(2))
              cleanedY.unpersist()
              transposedY.unpersist()
  
    }
  }
  
  
  class LinearRegression (val keys: (String, String),val pairs: Array[(Double,Double)]) {
  
  
    val size = pairs.size
  
    // first pass: read in data, compute xbar and ybar
    val sums = pairs.aggregate(new X_X2_Y(0D,0D,0D))(_ + new X_X2_Y(_),_+_)
    val bars = (sums.x / size, sums.y / size)
  
    // second pass: compute summary statistics
    val sumstats = pairs.foldLeft(new X2_Y2_XY(0D,0D,0D))(_ + new X2_Y2_XY(_, bars))
  
    val beta1 = sumstats.xy / sumstats.x2
    val beta0 = bars._2 - (beta1 * bars._1)
    val betas = (beta0, beta1)
  
    //println("y = " + ("%4.3f" format beta1) + " * x + " + ("%4.3f" format beta0))
  
  
    // analyze results
    val correlation = pairs.aggregate(new RSS_SSR(0D,0D))(_ + RSS_SSR.build(_, bars, betas),_+_)
    val R2 = correlation.ssr / sumstats.y2
    val svar = correlation.rss / (size - 2)
    val svar1 = svar / sumstats.x2
    val svar0 = ( svar / size ) + ( bars._1 * bars._1 * svar1)
    val svar0bis = svar * sums.x2 / (size * sumstats.x2)
  
    /* println("R^2                 = " + R2)
     println("std error of beta_1 = " + Math.sqrt(svar1))
     println("std error of beta_0 = " + Math.sqrt(svar0))
     println("std error of beta_0 = " + Math.sqrt(svar0bis))
     println("SSTO = " + sumstats.y2)
     println("SSE  = " + correlation.rss)
     println("SSR  = " + correlation.ssr)*/
  
  
    def outputVal() = keys._1
                      +"\t"+keys._2
                      +"\t"+beta1
                      +"\t"+beta0
                      +"\t"+R2
                      +"\t"+Math.sqrt(svar1)
                      +"\t"+Math.sqrt(svar0)
                      +"\t"+sumstats.y2
                      +"\t"+correlation.rss
                      +"\t"+correlation.ssr+"\t;
  
  }
  
  
  object RSS_SSR {
    def build(p: (Double,Double), bars: (Double,Double), betas: (Double,Double)): RSS_SSR = {
      val fit = (betas._2 * p._1) + betas._1
      val rss = (fit-p._2) * (fit-p._2)
      val ssr = (fit-bars._2) * (fit-bars._2)
      new RSS_SSR(rss, ssr)
    }
  }
  
  class RSS_SSR(val rss: Double, val ssr: Double) {
    def +(p: RSS_SSR): RSS_SSR = new RSS_SSR(rss+p.rss, ssr+p.ssr)
  }
  
  class X_X2_Y(val x: Double, val x2: Double, val y: Double) {
    def this(p: (Double,Double)) = this(p._1, p._1*p._1, p._2)
    def +(p: X_X2_Y): X_X2_Y = new X_X2_Y(x+p.x,x2+p.x2,y+p.y)
  }
  
  class X2_Y2_XY(val x2: Double, val y2: Double, val xy: Double) {
    def this(p: (Double,Double), bars: (Double,Double)) = this((p._1-bars._1)*(p._1-bars._1), (p._2-bars._2)*(p._2-bars._2),(p._1-bars._1)*(p._2-bars._2))
    def +(p: X2_Y2_XY): X2_Y2_XY = new X2_Y2_XY(x2+p.x2,y2+p.y2,xy+p.xy)
  }
  </code></pre>