<commit_before>/**
 * bisecting <master> <inputFile> <outputFile>
 *
 * divisive hierarchical clustering using bisecting k-means
 * writes result to JSON for display using d3
 *
 * (in progress)
 *
 */

import java.io.File
import spark.SparkContext
import spark.SparkContext._
import spark.util.Vector
import scala.collection.mutable.ArrayBuffer
import cc.spray.json._
import cc.spray.json.DefaultJsonProtocol._


object bisecting {

  case class Cluster(var key: Int, var center: List[Map[String,Double]], var children: Option[List[Cluster]])

  object MyJsonProtocol extends DefaultJsonProtocol {
    implicit val menuItemFormat: JsonFormat[Cluster] = lazyFormat(jsonFormat(Cluster, "key", "center", "children"))
  }

  import MyJsonProtocol._

  def insert(node: Cluster, key: Int, children: List[Cluster]) {
    // recursively search cluster tree for desired key and insert children
    if (node.key == key) {
      node.children = Some(children)
    } else {
      if (node.children.getOrElse(0) != 0) {
        insert(node.children.get(0),key,children)
        insert(node.children.get(1),key,children)
      }
    }
  }

  def makeXYmap(vec: Array[Double]): List[Map[String,Double]] = {
    return vec.toList.zipWithIndex.map(x => Map("x"->x._2.toDouble,"y"->x._1))
  }

  def parseVector(line: String): Vector = {
    val nums = line.split(' ')
    //val vec = nums.slice(1, nums.length).map(_.toDouble)
    var vec = nums.slice(3,nums.length).map(_.toDouble)
    val mean = vec.sum / vec.length
    vec = (vec.map(x => (x - mean)/(mean + 0.1)))
    return Vector(vec)
  }

  def printToFile(f: java.io.File)(op: java.io.PrintWriter => Unit) {
    val p = new java.io.PrintWriter(f)
    try {
      op(p)
    } finally {
      p.close()
    }
  }

  def closestPoint(p: Vector, centers: Array[Vector]): Int = {

    var bestIndex = 0
    var closest = Double.PositiveInfinity
    for (i <- 0 until centers.length) {
      val tempDist = p.squaredDist(centers(i))
      if (tempDist < closest) {
        closest = tempDist
        bestIndex = i
      }
    }
    return bestIndex
  }

  def split(cluster: spark.RDD[Vector]): Array[Vector] = {

    // use k-means with k=2 to split a cluster in two
    // try multiple splits and keep the best
    val convergeDist = 0.0001
    val iters = 5
    var best = Double.PositiveInfinity
    var centersFinal = cluster.takeSample(false,2,1).toArray
    for (iter <- 0 until iters) {
      val centers = cluster.takeSample(false, 2, iter).toArray
      var tempDist = 1.0
      // do k-means iterations till convergence
      while(tempDist > convergeDist) {
        val closest = cluster.map (p => (closestPoint(p, centers), (p, 1)))
        val pointStats = closest.reduceByKey{case ((x1, y1), (x2, y2)) => (x1 + x2, y1 + y2)}
        val newPoints = pointStats.map {pair => (pair._1, pair._2._1 / pair._2._2)}.collectAsMap()
        tempDist = 0.0
        for (i <- 0 until 2) {
          tempDist += centers(i).squaredDist(newPoints(i))
        }
        for (newP <- newPoints) {
          centers(newP._1) = newP._2
        }
      }
      // check within-class similarity of clusters
      var totalDist = 0.0
      for (i <- 0 until 2) {
        totalDist += cluster.filter(x => closestPoint(x,centers)==i).map(x => x.squaredDist(centers(i))).reduce(_+_)
      }
      if (totalDist < best) {
        best = totalDist
        centersFinal = centers
      }
    }
    return centersFinal
  }


  def main(args: Array[String]) {

    if (args.length < 3) {
      System.err.println("Usage: bisecting <master> <inputFile> <outputFile> <k>")
      System.exit(1)
    }

    System.setProperty("spark.executor.memory", "120g")
    System.setProperty("spark.serializer", "spark.KryoSerializer")
    //System.setProperty("spark.default.parallelism", "50")
    val sc = new SparkContext(args(0), "hierarchical", System.getenv("SPARK_HOME"),
      List("target/scala-2.9.3/thunder_2.9.3-1.0.jar"))
    //sc.setCheckpointDir(System.getenv("CHECKPOINT"))

    val k = args(3).toDouble
    val data = sc.textFile(args(1)).map(parseVector _).cache()

    val clusters = ArrayBuffer((0,data))
    val center = data.reduce(_+_).elements.map(x => x / data.count())
    val tree = Cluster(0,makeXYmap(center),None)
    var count = 1

    while (clusters.size < k) {

      // find largest cluster for splitting
      val ind = clusters.map(_._2.count()).view.zipWithIndex.max._2

      // split into 2 clusters using k-means
      val centers = split(clusters(ind)._2) // find 2 cluster centers<commit_msg>Made iterations on each sub kmeans routine an arg<commit_after>/**
 * bisecting <master> <inputFile> <outputFile>
 *
 * divisive hierarchical clustering using bisecting k-means
 * writes result to JSON for display using d3
 *
 * (in progress)
 *
 */

import java.io.File
import spark.SparkContext
import spark.SparkContext._
import spark.util.Vector
import scala.collection.mutable.ArrayBuffer
import cc.spray.json._
import cc.spray.json.DefaultJsonProtocol._


object bisecting {

  case class Cluster(var key: Int, var center: List[Map[String,Double]], var children: Option[List[Cluster]])

  object MyJsonProtocol extends DefaultJsonProtocol {
    implicit val menuItemFormat: JsonFormat[Cluster] = lazyFormat(jsonFormat(Cluster, "key", "center", "children"))
  }

  import MyJsonProtocol._

  def insert(node: Cluster, key: Int, children: List[Cluster]) {
    // recursively search cluster tree for desired key and insert children
    if (node.key == key) {
      node.children = Some(children)
    } else {
      if (node.children.getOrElse(0) != 0) {
        insert(node.children.get(0),key,children)
        insert(node.children.get(1),key,children)
      }
    }
  }

  def makeXYmap(vec: Array[Double]): List[Map[String,Double]] = {
    return vec.toList.zipWithIndex.map(x => Map("x"->x._2.toDouble,"y"->x._1))
  }

  def parseVector(line: String): Vector = {
    val nums = line.split(' ')
    //val vec = nums.slice(1, nums.length).map(_.toDouble)
    var vec = nums.slice(3,nums.length).map(_.toDouble)
    val mean = vec.sum / vec.length
    vec = (vec.map(x => (x - mean)/(mean + 0.1)))
    return Vector(vec)
  }

  def printToFile(f: java.io.File)(op: java.io.PrintWriter => Unit) {
    val p = new java.io.PrintWriter(f)
    try {
      op(p)
    } finally {
      p.close()
    }
  }

  def closestPoint(p: Vector, centers: Array[Vector]): Int = {

    var bestIndex = 0
    var closest = Double.PositiveInfinity
    for (i <- 0 until centers.length) {
      val tempDist = p.squaredDist(centers(i))
      if (tempDist < closest) {
        closest = tempDist
        bestIndex = i
      }
    }
    return bestIndex
  }

  def split(cluster: spark.RDD[Vector], subIters: Int): Array[Vector] = {

    // use k-means with k=2 to split a cluster in two
    // try multiple splits and keep the best
    val convergeDist = 0.0001
    var best = Double.PositiveInfinity
    var centersFinal = cluster.takeSample(false,2,1).toArray
    for (iter <- 0 until subIters) {
      val centers = cluster.takeSample(false, 2, iter).toArray
      var tempDist = 1.0
      // do k-means iterations till convergence
      while(tempDist > convergeDist) {
        val closest = cluster.map (p => (closestPoint(p, centers), (p, 1)))
        val pointStats = closest.reduceByKey{case ((x1, y1), (x2, y2)) => (x1 + x2, y1 + y2)}
        val newPoints = pointStats.map {pair => (pair._1, pair._2._1 / pair._2._2)}.collectAsMap()
        tempDist = 0.0
        for (i <- 0 until 2) {
          tempDist += centers(i).squaredDist(newPoints(i))
        }
        for (newP <- newPoints) {
          centers(newP._1) = newP._2
        }
      }
      // check within-class similarity of clusters
      var totalDist = 0.0
      for (i <- 0 until 2) {
        totalDist += cluster.filter(x => closestPoint(x,centers)==i).map(x => x.squaredDist(centers(i))).reduce(_+_)
      }
      if (totalDist < best) {
        best = totalDist
        centersFinal = centers
      }
    }
    return centersFinal
  }


  def main(args: Array[String]) {

    if (args.length < 3) {
      System.err.println("Usage: bisecting <master> <inputFile> <outputFile> <k> <subIters>")
      System.exit(1)
    }

    System.setProperty("spark.executor.memory", "120g")
    System.setProperty("spark.serializer", "spark.KryoSerializer")
    //System.setProperty("spark.default.parallelism", "50")
    val sc = new SparkContext(args(0), "hierarchical", System.getenv("SPARK_HOME"),
      List("target/scala-2.9.3/thunder_2.9.3-1.0.jar"))
    //sc.setCheckpointDir(System.getenv("CHECKPOINT"))

    val k = args(3).toDouble
    val subIters = args(4).toInt
    val data = sc.textFile(args(1)).map(parseVector _).cache()

    val clusters = ArrayBuffer((0,data))
    val center = data.reduce(_+_).elements.map(x => x / data.count())
    val tree = Cluster(0,makeXYmap(center),None)
    var count = 1

    while (clusters.size < k) {

      // find largest cluster for splitting
      val ind = clusters.map(_._2.count()).view.zipWithIndex.max._2

      // split into 2 clusters using k-means
      val centers = split(clusters(ind)._2,subIters) // find 2 cluster centers
{'fraction_non_alphanumeric': 0.13985404640017426, 'fraction_numerical': 0.01655593072650038, 'mean_word_length': 2.843449141900377, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 42, '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': '7034388', 'n_tokens_mistral': 3153, 'n_tokens_neox': 2985, 'n_words': 720}