<commit_before>package edu.fuberlin.hotspots

import java.lang.Math.sqrt

import org.apache.commons.math3.distribution.NormalDistribution

import org.apache.spark.rdd.RDD
import edu.fuberlin.hotspots.SparkHelpers._

import scala.collection.mutable.ListBuffer

/**
  * Created by <NAME> on 01.07.16.
  */
object GetisOrd {
  val superCellSize = 25

  def calculate(trips:RDD[Trip], cellSize:Any, timeSpan:Any):RDD[(Cellid, Double, Double)] = {
    val observations = trips.toCells(cellSize, timeSpan).cache
    val stdDev = observations.values.stdev
    val mean = observations.values.mean
    val count = observations.count
    val norm = new NormalDistribution()
    val factory = new SuperCellFactory(superCellSize)
    val superCells = observations.flatMap(factory.create).mapValues(Seq(_)).reduceByKey(_ ++ _).map(c => new SuperCell(c._2, superCellSize, c._1))
    superCells.flatMap(superCell => {
      val buffer = new ListBuffer[(Cellid, Double, Double)]
      for((cellid, passengerCount) <- superCell.coreCells){
        val neighbours = superCell.neighbours(cellid)
        val radicant = ((count * neighbours.size) - Math.pow(neighbours.size, 2.0)) / (count - 1)
        val denominator = stdDev * sqrt(radicant)
        val numerator = neighbours.sum - (mean * neighbours.size)
        val zValue = numerator / denominator
        buffer.append((cellid, zValue, 1 - norm.cumulativeProbability(Math.abs(zValue))))
      }
      buffer
    })
  }<commit_msg>use aggregateBy instead of reductByKey<commit_after>package edu.fuberlin.hotspots

import java.lang.Math.sqrt

import org.apache.commons.math3.distribution.NormalDistribution

import org.apache.spark.rdd.RDD
import edu.fuberlin.hotspots.SparkHelpers._

import scala.collection.mutable.ListBuffer

/**
  * Created by <NAME> on 01.07.16.
  */
object GetisOrd {
  val superCellSize = 25

  def calculate(trips:RDD[Trip], cellSize:Any, timeSpan:Any):RDD[(Cellid, Double, Double)] = {
    val observations = trips.toCells(cellSize, timeSpan).cache
    val stdDev = observations.values.stdev
    val mean = observations.values.mean
    val count = observations.count
    val norm = new NormalDistribution()
    val factory = new SuperCellFactory(superCellSize)
    val superCells = observations.flatMap(factory.create).aggregateByKey(Seq[(Int, Int)]())(_ :+ _, _ ++ _).map(c => new SuperCell(c._2, superCellSize, c._1))
    superCells.flatMap(superCell => {
      val buffer = new ListBuffer[(Cellid, Double, Double)]
      for((cellid, passengerCount) <- superCell.coreCells){
        val neighbours = superCell.neighbours(cellid)
        val radicant = ((count * neighbours.size) - Math.pow(neighbours.size, 2.0)) / (count - 1)
        val denominator = stdDev * sqrt(radicant)
        val numerator = neighbours.sum - (mean * neighbours.size)
        val zValue = numerator / denominator
        buffer.append((cellid, zValue, 1 - norm.cumulativeProbability(Math.abs(zValue))))
      }
      buffer
    })
  }
{'fraction_non_alphanumeric': 0.12, 'fraction_numerical': 0.010084033613445379, 'mean_word_length': 3.664576802507837, 'pattern_counts': {'":': 0, '<': 7, '<?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': '249948', 'n_tokens_mistral': 983, 'n_tokens_neox': 919, 'n_words': 197}