<commit_before>package water.sparkling.itest.yarn

import hex.kmeans.KMeansModel.KMeansParameters
import org.apache.spark.h2o._
import org.apache.spark.mllib.clustering.KMeans
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.DataFrame
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import water.sparkling.itest.{SparkITest, YarnTest}
import water.util.Timer

/**
 * Created by michal on 2/9/15.
 */
@RunWith(classOf[JUnitRunner])
class KMeansITestSuite extends FunSuite with SparkITest {

  test("MLlib KMeans on airlines_all data", YarnTest) {
    launch( "water.sparkling.itest.yarn.KMeansITest",
      env {
        sparkMaster("yarn-client")
        // Configure YARN environment
        conf("spark.yarn.max.executor.failures", 1) // In fail of executor, fail the test
        conf("spark.executor.instances", 6) // 10 executor instances
        conf("spark.executor.memory", "8g") // 20g per executor
        conf("spark.ext.h2o.port.base", 63331) //Start at baseport 63331
        conf("spark.driver.memory", "8g")
        conf("spark.executor.cores", 32) //Use up all the cores on the machines
      }
    )
  }
}

/**
 * Test runner loading large airlines data from YARN HDFS via H2O API
 * transforming them into RDD and launching MLlib K-means.
 */
object KMeansITest {

  def main(args: Array[String]): Unit = {
    try {
      test(args)
    } catch {
      case t:Throwable => {
        System.err.println(t.toString)
        System.err.println(t.getStackTrace.toString)
        water.H2O.exit(-1)
      }
    }
  }

  def test(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("KMeansITest")
    val sc = new SparkContext(conf)
    val h2oContext = H2OContext.getOrCreate(sc)

    import h2oContext._
    // Import all year airlines into H2O
    val path = "hdfs://mr-0xd6.0xdata.loc/datasets/airlines_all.csv"
    val timer1 = new water.util.Timer
    val d = new java.net.URI(path)
    val airlinesData = new H2OFrame(d)
    val timeToParse = timer1.time

    // Run Kmeans in H2O
    val ignore_columns = airlinesData.names diff Array("Month", "DayofMonth", "DayOfWeek")
    val H2OKMTimer = new Timer
    val kmeansParams = new KMeansParameters()
    kmeansParams._k = 5
    kmeansParams._max_iterations = 10
    kmeansParams._train = airlinesData
    kmeansParams._ignored_columns = ignore_columns
    kmeansParams._standardize = false
    val KmeansModel = new hex.kmeans.KMeans(kmeansParams).trainModel().get()
    val H2OKMBuildTime = H2OKMTimer.time

    // Score in H2O
    import org.apache.spark.sql.SQLContext
    implicit val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
    val pred = KmeansModel.score(airlinesData)
    val predDF = asDataFrame(pred)
    val clusterCounts = predDF.rdd.countByValue()

    // Run Kmeans in Spark
    val sqlQueryTimer = new water.util.Timer
    val airlinesDF = asDataFrame(airlinesData)(sqlContext)
    airlinesDF.registerTempTable("airlinesRDD")
    val airlinesTable = sqlContext.sql(
      """SELECT Month, DayofMonth, DayOfWeek FROM airlinesRDD"""
    )
    val sqlQueryTime = sqlQueryTimer.time

    assert (airlinesData.numRows == airlinesTable.count, "Transfer of H2ORDD to SparkRDD completed!")

    // Run Kmeans in Spark  on indices 10,19,26 (FlightNo, Distance, WeatherDelay)

    val airlinesVectorRDD = airlinesTable.map(row => Vectors.dense(row.getByte(0) * 1.0, row.getByte(1) * 1.0, row.getByte(2) * 1.0))
    val SparkKMTimer = new water.util.Timer
    val clusters = KMeans.train(airlinesVectorRDD, 5, 10)
    val SparkKMBuildTime = SparkKMTimer.time

    // Predict on Spark's Kmeans
    val sparkPredRDD = clusters.predict(airlinesVectorRDD)
    val srdd : DataFrame = sparkPredRDD.map(v => IntHolder(Option(v))).toDF
    val df : H2OFrame = srdd
    val sparkClusterCounts = sparkPredRDD.countByValue()

    // Get Within Set Sum of Squared Errors
    val sparkWSSSE = clusters.computeCost(airlinesVectorRDD)
    val h2oWSSSE = KmeansModel._output._withinss.fold(0.0)(_+_)

    println("Spark: Within Set Sum of Squared Errors = " + sparkWSSSE)
    println("Spark: Time to Build (s) = " + SparkKMBuildTime)
    println("H2O: Within Set Sum of Squared Errors = " + h2oWSSSE)
    println("H2O: Time to Build (s) = " + H2OKMBuildTime)

    val relativeMeanDiff = (sparkWSSSE - h2oWSSSE)/sparkWSSSE
    assert (relativeMeanDiff < 0.01, "Within Set Sum of Squared Errors matches!")

    // Shutdown Spark
    sc.stop()
    // Shutdown H2O explicitly
    water.H2O.exit(0)
  }
}<commit_msg>Increase relativeMeanDiff in KMeansITestSuite<commit_after>package water.sparkling.itest.yarn

import hex.kmeans.KMeansModel.KMeansParameters
import org.apache.spark.h2o._
import org.apache.spark.mllib.clustering.KMeans
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.DataFrame
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import water.sparkling.itest.{SparkITest, YarnTest}
import water.util.Timer

/**
 * Created by michal on 2/9/15.
 */
@RunWith(classOf[JUnitRunner])
class KMeansITestSuite extends FunSuite with SparkITest {

  test("MLlib KMeans on airlines_all data", YarnTest) {
    launch( "water.sparkling.itest.yarn.KMeansITest",
      env {
        sparkMaster("yarn-client")
        // Configure YARN environment
        conf("spark.yarn.max.executor.failures", 1) // In fail of executor, fail the test
        conf("spark.executor.instances", 6) // 10 executor instances
        conf("spark.executor.memory", "8g") // 20g per executor
        conf("spark.ext.h2o.port.base", 63331) //Start at baseport 63331
        conf("spark.driver.memory", "8g")
        conf("spark.executor.cores", 32) //Use up all the cores on the machines
      }
    )
  }
}

/**
 * Test runner loading large airlines data from YARN HDFS via H2O API
 * transforming them into RDD and launching MLlib K-means.
 */
object KMeansITest {

  def main(args: Array[String]): Unit = {
    try {
      test(args)
    } catch {
      case t:Throwable => {
        System.err.println(t.toString)
        System.err.println(t.getStackTrace.toString)
        water.H2O.exit(-1)
      }
    }
  }

  def test(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("KMeansITest")
    val sc = new SparkContext(conf)
    val h2oContext = H2OContext.getOrCreate(sc)

    import h2oContext._
    // Import all year airlines into H2O
    val path = "hdfs://mr-0xd6.0xdata.loc/datasets/airlines_all.csv"
    val timer1 = new water.util.Timer
    val d = new java.net.URI(path)
    val airlinesData = new H2OFrame(d)
    val timeToParse = timer1.time

    // Run Kmeans in H2O
    val ignore_columns = airlinesData.names diff Array("Month", "DayofMonth", "DayOfWeek")
    val H2OKMTimer = new Timer
    val kmeansParams = new KMeansParameters()
    kmeansParams._k = 5
    kmeansParams._max_iterations = 10
    kmeansParams._train = airlinesData
    kmeansParams._ignored_columns = ignore_columns
    kmeansParams._standardize = false
    val KmeansModel = new hex.kmeans.KMeans(kmeansParams).trainModel().get()
    val H2OKMBuildTime = H2OKMTimer.time

    // Score in H2O
    import org.apache.spark.sql.SQLContext
    implicit val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
    val pred = KmeansModel.score(airlinesData)
    val predDF = asDataFrame(pred)
    val clusterCounts = predDF.rdd.countByValue()

    // Run Kmeans in Spark
    val sqlQueryTimer = new water.util.Timer
    val airlinesDF = asDataFrame(airlinesData)(sqlContext)
    airlinesDF.registerTempTable("airlinesRDD")
    val airlinesTable = sqlContext.sql(
      """SELECT Month, DayofMonth, DayOfWeek FROM airlinesRDD"""
    )
    val sqlQueryTime = sqlQueryTimer.time

    assert (airlinesData.numRows == airlinesTable.count, "Transfer of H2ORDD to SparkRDD completed!")

    // Run Kmeans in Spark  on indices 10,19,26 (FlightNo, Distance, WeatherDelay)

    val airlinesVectorRDD = airlinesTable.map(row => Vectors.dense(row.getByte(0) * 1.0, row.getByte(1) * 1.0, row.getByte(2) * 1.0))
    val SparkKMTimer = new water.util.Timer
    val clusters = KMeans.train(airlinesVectorRDD, 5, 10)
    val SparkKMBuildTime = SparkKMTimer.time

    // Predict on Spark's Kmeans
    val sparkPredRDD = clusters.predict(airlinesVectorRDD)
    val srdd : DataFrame = sparkPredRDD.map(v => IntHolder(Option(v))).toDF
    val df : H2OFrame = srdd
    val sparkClusterCounts = sparkPredRDD.countByValue()

    // Get Within Set Sum of Squared Errors
    val sparkWSSSE = clusters.computeCost(airlinesVectorRDD)
    val h2oWSSSE = KmeansModel._output._withinss.fold(0.0)(_+_)

    println("Spark: Within Set Sum of Squared Errors = " + sparkWSSSE)
    println("Spark: Time to Build (s) = " + SparkKMBuildTime)
    println("H2O: Within Set Sum of Squared Errors = " + h2oWSSSE)
    println("H2O: Time to Build (s) = " + H2OKMBuildTime)

    val relativeMeanDiff = (sparkWSSSE - h2oWSSSE)/sparkWSSSE
    assert (relativeMeanDiff < 0.1, "Within Set Sum of Squared Errors matches!")

    // Shutdown Spark
    sc.stop()
    // Shutdown H2O explicitly
    water.H2O.exit(0)
  }
}
{'fraction_non_alphanumeric': 0.10628019323671498, 'fraction_numerical': 0.01728395061728395, 'mean_word_length': 3.3778195488721803, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3959503', 'n_tokens_mistral': 3389, 'n_tokens_neox': 3110, 'n_words': 733}