<commit_before>/*
 * Copyright 2016 Spotify AB.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.spotify.bdrc.pipeline

import com.spotify.bdrc.util.Records.Rating
import com.spotify.scio.values.SCollection
import com.twitter.algebird.MultiAggregator
import com.twitter.scalding.TypedPipe
import org.apache.spark.rdd.RDD

/**
 * Compute basic statistics.
 *
 * Input is a collection of (user, item, score).
 */
object Statistics {

  case class Stats(max: Double, min: Double, sum: Double, count: Long, mean: Double, stdev: Double)

  // Algebird Aggregator
  def aggregator = {
    import com.twitter.algebird.Aggregator.{prepareMonoid, _}
    import com.twitter.algebird.Moments

    // 4 aggregators with different logic
    val maxOp = maxBy[Rating, Double](_.score)
    val minOp = minBy[Rating, Double](_.score)
    val sum = prepareMonoid[Rating, Double](_.score)
    val moments = Moments.aggregator.composePrepare[Rating](_.score)

    // Apply 4 aggregators on the same input, present result Tuple4 as Stats
    MultiAggregator(maxOp, minOp, sum, moments)
      .andThenPresent { case (mmax, mmin, ssum, moment) =>
        Stats(mmax.score, mmin.score, ssum, moment.count, moment.mean, moment.stddev)
      }
  }

  def scalding(input: TypedPipe[Rating]): TypedPipe[Stats] = {
    input.aggregate(aggregator)
  }

  def scio(input: SCollection[Rating]): SCollection[Stats] = {
    input
      .map(_.score)
      .stats()
      .map(s => Stats(s.max, s.min, s.sum, s.count, s.mean, s.stdev))
  }

  def scioAlgebird(input: SCollection[Rating]): SCollection[Stats] = {
    input.aggregate(aggregator)
  }

  def spark(input: RDD[Rating]): Stats = {
    val s = input.map(_.score).stats()
    Stats(s.max, s.min, s.sum, s.count, s.mean, s.stdev)
  }
<commit_msg>clean up Statistics.scala<commit_after>/*
 * Copyright 2016 Spotify AB.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.spotify.bdrc.pipeline

import com.spotify.bdrc.util.Records.Rating
import com.spotify.scio.values.SCollection
import com.twitter.scalding.TypedPipe
import org.apache.spark.rdd.RDD

/**
 * Compute basic statistics.
 *
 * Input is a collection of (user, item, score).
 */
object Statistics {

  case class Stats(max: Double, min: Double, sum: Double, count: Long, mean: Double, stddev: Double)

  // Algebird Aggregator
  def aggregator = {
    import com.twitter.algebird._

    /*
    4 aggregators with different logic

    The first 3 aggregators are of type Aggregator[Rating, _, Double] which means it takes Rating as
    input and outputs Double. The last aggregator is of type Aggregator[Rating, _, Moments].
    The input Rating is prepared with a (Rating => Double) function _.score.
     */
    val maxOp = Aggregator.max[Double].composePrepare[Rating](_.score)
    val minOp = Aggregator.min[Double].composePrepare[Rating](_.score)
    val sumOp = Aggregator.prepareMonoid[Rating, Double](_.score)
    val momentsOp = Moments.aggregator.composePrepare[Rating](_.score)

    // Apply 4 aggregators on the same input, present result Tuple4 as Stats
    MultiAggregator(maxOp, minOp, sumOp, momentsOp)
      .andThenPresent { case (max, min, sum, moments) =>
        Stats(max, min, sum, moments.count, moments.mean, moments.stddev)
      }
  }

  def scalding(input: TypedPipe[Rating]): TypedPipe[Stats] = {
    input.aggregate(aggregator)
  }

  def scio(input: SCollection[Rating]): SCollection[Stats] = {
    input
      .map(_.score)
      .stats()
      .map(s => Stats(s.max, s.min, s.sum, s.count, s.mean, s.stdev))
  }

  def scioAlgebird(input: SCollection[Rating]): SCollection[Stats] = {
    input.aggregate(aggregator)
  }

  def spark(input: RDD[Rating]): Stats = {
    val s = input.map(_.score).stats()
    Stats(s.max, s.min, s.sum, s.count, s.mean, s.stdev)
  }

{'fraction_non_alphanumeric': 0.1128565522943365, 'fraction_numerical': 0.0047540305911533695, 'mean_word_length': 3.543661971830986, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6199748', 'n_tokens_mistral': 1593, 'n_tokens_neox': 1508, 'n_words': 464}