<commit_before>package GeoTsRDD

import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext._

import scala.reflect.ClassTag

/**
 * Created by <NAME> on 6/17/15.
 */

/*
TODO: Internal RDD as IndexedRDD (mesh index for spatial data)
TODO: Convert to DataFrame representation
 */
case class Rasterizer[LocType, AttribType: ClassTag, MergeType](
  private val grid: LocType => Long,
  private val aggregator: Aggregator[AttribType, MergeType],
  private val postMapper: ((Long, MergeType)) => (Long, AttribType),
  private val rawData: RDD[LocEvent[LocType, AttribType]]) {

  private val rasters: RDD[(Long, AttribType)] = rawData
    .map(x =>(grid(x.getLoc()), x.getAttrib()))
    .combineByKey(aggregator.initializer, aggregator.valueMerger, aggregator.valueCombiner)
    .map(postMapper)

  def persist() = rasters.persist()

  def getValueAtLoc(loc: LocType): Seq[AttribType] ={
    rasters.lookup(grid(loc))
  }

  def collectRasters() = rasters.collect()

  def collectAsMap() = rasters.collectAsMap()

}<commit_msg>Solved bugged and tested with python notebook on NY data.<commit_after>package GeoTsRDD

import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext._
import org.apache.spark.sql._

import scala.reflect.ClassTag

/**
 * Created by <NAME> on 6/17/15.
 */

/*
TODO: Internal RDD as IndexedRDD (mesh index for spatial data)
TODO: Convert to DataFrame representation
 */
case class Rasterizer[LocType, HashedType: ClassTag, AttribType: ClassTag, MergeType](
  private val grid: LocType => HashedType,
  private val aggregator: Aggregator[AttribType, MergeType],
  private val postMapper: ((HashedType, MergeType)) => (HashedType, AttribType),
  private val rawData: RDD[LocEvent[LocType, AttribType]]) {

  val rasters: RDD[(HashedType, AttribType)] = rawData
    .map(x =>(grid(x.getLoc()), x.getAttrib()))
    .combineByKey(aggregator.initializer, aggregator.valueMerger, aggregator.valueCombiner)
    .map(postMapper)

  def persist() = rasters.persist()

  def getValueAtLoc(loc: LocType): Seq[AttribType] ={
    rasters.lookup(grid(loc))
  }

  def collectRasters() = rasters.collect()

  def collectAsMap() = rasters.collectAsMap()

}
{'fraction_non_alphanumeric': 0.12600896860986546, 'fraction_numerical': 0.004484304932735426, 'mean_word_length': 4.902116402116402, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '4316737', 'n_tokens_mistral': 777, 'n_tokens_neox': 754, 'n_words': 182}