<commit_before>package org.project.thunder.streaming.analyses

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.project.thunder.streaming.analyses.Analysis.OutputListType
import org.project.thunder.streaming.rdds.StreamingSeries
import org.project.thunder.streaming.util.ThunderStreamingContext

import scala.util.{Failure, Success, Try}

abstract class SeriesTestAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends Analysis[Int, Array[Double]] {

  def getSeries(): StreamingSeries = {
    val dataPath = params.getOrElse(SeriesTestAnalysis.DATA_PATH_KEY, "series_data")
    tssc.loadStreamingSeries(dataPath, inputFormat = params.getOrElse(SeriesTestAnalysis.FORMAT_KEY, ""))
  }

  def register(outputs: OutputListType): Unit = {
    val data = getSeries()
    val outputFuncs = getOutputFunctions(outputs)
    val analyzedData  = analyze(data)
    analyzedData.outputWithKeys(outputFuncs)
  }

  def analyze(data: StreamingSeries): StreamingSeries

}

object SeriesTestAnalysis {
  final val DATA_PATH_KEY = "data_path"
  final val FORMAT_KEY = "format"
}

class SeriesMeanAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends SeriesTestAnalysis(tssc, params) {
  def analyze(data: StreamingSeries): StreamingSeries = {
    data.seriesMean()
  }
}

class SeriesCountAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends SeriesTestAnalysis(tssc, params) {
  def analyze(data: StreamingSeries): StreamingSeries = {
    data.count()
  }
}


<commit_msg>Added the SeriesNoopAnalysis which just passes the input through unmodified<commit_after>package org.project.thunder.streaming.analyses

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.project.thunder.streaming.analyses.Analysis.OutputListType
import org.project.thunder.streaming.rdds.StreamingSeries
import org.project.thunder.streaming.util.ThunderStreamingContext

import scala.util.{Failure, Success, Try}

abstract class SeriesTestAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends Analysis[Int, Array[Double]] {

  def getSeries(): StreamingSeries = {
    val dataPath = params.getOrElse(SeriesTestAnalysis.DATA_PATH_KEY, "series_data")
    tssc.loadStreamingSeries(dataPath, inputFormat = params.getOrElse(SeriesTestAnalysis.FORMAT_KEY, ""))
  }

  def register(outputs: OutputListType): Unit = {
    val data = getSeries()
    val outputFuncs = getOutputFunctions(outputs)
    val analyzedData  = analyze(data)
    analyzedData.outputWithKeys(outputFuncs)
  }

  def analyze(data: StreamingSeries): StreamingSeries

}

object SeriesTestAnalysis {
  final val DATA_PATH_KEY = "data_path"
  final val FORMAT_KEY = "format"
}

class SeriesMeanAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends SeriesTestAnalysis(tssc, params) {
  def analyze(data: StreamingSeries): StreamingSeries = {
    data.seriesMean()
  }
}

class SeriesCountAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends SeriesTestAnalysis(tssc, params) {
  def analyze(data: StreamingSeries): StreamingSeries = {
    data.count()
  }
}

class SeriesNoopAnalysis(tssc: ThunderStreamingContext, params: Map[String, String]) extends SeriesTestAnalysis(tssc, params) {
  def analyze(data: StreamingSeries): StreamingSeries = {
    data
  }
}



{'fraction_non_alphanumeric': 0.09737609329446065, 'fraction_numerical': 0.0, 'mean_word_length': 6.030737704918033, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '6038448', 'n_tokens_mistral': 1060, 'n_tokens_neox': 1023, 'n_words': 209}