<commit_before>/*
Copyright 2012 Twitter, Inc.

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.twitter.scalding

import cascading.tuple.Fields
import cascading.tuple.{Tuple => CTuple, TupleEntry}

import com.twitter.algebird.{
  Monoid,
  Ring,
  AveragedValue,
  Moments,
  SortedTakeListMonoid,
  HyperLogLogMonoid,
  HLL,
  Aggregator
}

import scala.collection.JavaConverters._

import Dsl._ //Get the conversion implicits

/** Implements reductions on top of a simple abstraction for the Fields-API
 * This is for associative and commutive operations (particularly Monoids play a big role here)
 *
 * We use the f-bounded polymorphism trick to return the type called Self
 * in each operation.
 */
trait ReduceOperations[+Self <: ReduceOperations[Self]] extends java.io.Serializable {
 /**
  * Type T is the type of the input field (input to map, T => X)
  * Type X is the intermediate type, which your reduce function operates on
  * (reduce is (X,X) => X)
  * Type U is the final result type, (final map is: X => U)
  *
  * The previous output goes into the reduce function on the left, like foldLeft,
  * so if your operation is faster for the accumulator to be on one side, be aware.
  */
  def mapReduceMap[T,X,U](fieldDef : (Fields, Fields))(mapfn : T => X )(redfn : (X, X) => X)
      (mapfn2 : X => U)(implicit startConv : TupleConverter[T],
                        middleSetter : TupleSetter[X],
                        middleConv : TupleConverter[X],
                        endSetter : TupleSetter[U]) : Self

  /////////////////////////////////////////
  // All the below functions are implemented in terms of the above
  /////////////////////////////////////////

  /** Pretty much a synonym for mapReduceMap with the methods collected into a trait. */
  def aggregate[A,B,C](fieldDef : (Fields, Fields))(ag: Aggregator[A,B,C])
    (implicit startConv : TupleConverter[A],
                        middleSetter : TupleSetter[B],
                        middleConv : TupleConverter[B],
                        endSetter : TupleSetter[C]): Self =
    mapReduceMap[A,B,C](fieldDef)(ag.prepare _)(ag.reduce _)(ag.present _)

  /**
   * uses a more stable online algorithm which should
   * be suitable for large numbers of records
   *
   * == Similar To ==
   * <a href="http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Parallel_algorithm">http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Parallel_algorithm</a>
   */
  def average(f : (Fields, Fields)) = mapPlusMap(f) { (x : Double) => AveragedValue(1L, x) } { _.value }
  def average(f : Symbol) : Self = average(f->f)

  /**
   * Approximate number of unique values
   * We use about m = (104/errPercent)^2 bytes of memory per key
   * Uses `.toString.getBytes` to serialize the data so you MUST
   * ensure that .toString is an equivalance on your counted fields
   * (i.e. `x.toString == y.toString` if and only if `x == y`)
   *
   * For each key:
   * {{{
   * 10% error ~ 256 bytes
   * 5% error ~ 1kb
   * 1% error ~ 8kb
   * 0.5% error ~ 64kb
   * 0.25% error ~ 256kb
   * }}}
   */
  def approxUniques(f : (Fields, Fields), errPercent : Double = 1.0) = {
    hyperLogLogMap(f, errPercent) { (monoid,hll) => monoid.estimateSize(hll) }
      { _.toString.getBytes("UTF-8") }
  }

  def hyperLogLog(f : (Fields, Fields), errPercent : Double = 1.0) = {
    hyperLogLogMap(f, errPercent) { (_,hll) => hll }
      { _.toString.getBytes("UTF-8") }
  }

  def hyperLogLogMap[T,U](f : (Fields, Fields), errPercent : Double = 1.0)
                         (fn : (HyperLogLogMonoid,HLL) => U)
                         (toBytes : (T) => Array[Byte]) = {
    //bits = log(m) == 2 *log(104/errPercent) = 2log(104) - 2*log(errPercent)
    def log2(x : Double) = scala.math.log(x)/scala.math.log(2.0)
    val bits = 2 * scala.math.ceil(log2(104) - log2(errPercent)).toInt
    implicit val hmm = new HyperLogLogMonoid(bits)
    mapPlusMap(f) { (t : T) => hmm.create(toBytes(t)) }
      { hll => fn(hmm,hll) }
  }

  /**<commit_msg>fix unit test<commit_after>/*
Copyright 2012 Twitter, Inc.

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.twitter.scalding

import cascading.tuple.Fields
import cascading.tuple.{Tuple => CTuple, TupleEntry}

import com.twitter.algebird.{
  Monoid,
  Ring,
  AveragedValue,
  Moments,
  SortedTakeListMonoid,
  HyperLogLogMonoid,
  HLL,
  Aggregator
}

import scala.collection.JavaConverters._

import Dsl._ //Get the conversion implicits

/** Implements reductions on top of a simple abstraction for the Fields-API
 * This is for associative and commutive operations (particularly Monoids play a big role here)
 *
 * We use the f-bounded polymorphism trick to return the type called Self
 * in each operation.
 */
trait ReduceOperations[+Self <: ReduceOperations[Self]] extends java.io.Serializable {
 /**
  * Type T is the type of the input field (input to map, T => X)
  * Type X is the intermediate type, which your reduce function operates on
  * (reduce is (X,X) => X)
  * Type U is the final result type, (final map is: X => U)
  *
  * The previous output goes into the reduce function on the left, like foldLeft,
  * so if your operation is faster for the accumulator to be on one side, be aware.
  */
  def mapReduceMap[T,X,U](fieldDef : (Fields, Fields))(mapfn : T => X )(redfn : (X, X) => X)
      (mapfn2 : X => U)(implicit startConv : TupleConverter[T],
                        middleSetter : TupleSetter[X],
                        middleConv : TupleConverter[X],
                        endSetter : TupleSetter[U]) : Self

  /////////////////////////////////////////
  // All the below functions are implemented in terms of the above
  /////////////////////////////////////////

  /** Pretty much a synonym for mapReduceMap with the methods collected into a trait. */
  def aggregate[A,B,C](fieldDef : (Fields, Fields))(ag: Aggregator[A,B,C])
    (implicit startConv : TupleConverter[A],
                        middleSetter : TupleSetter[B],
                        middleConv : TupleConverter[B],
                        endSetter : TupleSetter[C]): Self =
    mapReduceMap[A,B,C](fieldDef)(ag.prepare _)(ag.reduce _)(ag.present _)

  /**
   * uses a more stable online algorithm which should
   * be suitable for large numbers of records
   *
   * == Similar To ==
   * <a href="http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Parallel_algorithm">http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Parallel_algorithm</a>
   */
  def average(f : (Fields, Fields)) = mapPlusMap(f) { (x : Double) => AveragedValue(1L, x) } { _.value }
  def average(f : Symbol) : Self = average(f->f)

  /**
   * Approximate number of unique values
   * We use about m = (104/errPercent)^2 bytes of memory per key
   * Uses `.toString.getBytes` to serialize the data so you MUST
   * ensure that .toString is an equivalance on your counted fields
   * (i.e. `x.toString == y.toString` if and only if `x == y`)
   *
   * For each key:
   * {{{
   * 10% error ~ 256 bytes
   * 5% error ~ 1kb
   * 1% error ~ 8kb
   * 0.5% error ~ 64kb
   * 0.25% error ~ 256kb
   * }}}
   */
  def approxUniques(f : (Fields, Fields), errPercent : Double = 1.0) = {
    hyperLogLogMap(f, errPercent) { (monoid,hll) => monoid.estimateSize(hll) }
      { x: Any => x.toString.getBytes("UTF-8") }
  }

  def hyperLogLog(f : (Fields, Fields), errPercent : Double = 1.0) = {
    hyperLogLogMap(f, errPercent) { (_,hll) => hll }
      { x: Any => x.toString.getBytes("UTF-8") }
  }

  def hyperLogLogMap[T,U](f : (Fields, Fields), errPercent : Double = 1.0)
                         (fn : (HyperLogLogMonoid,HLL) => U)
                         (toBytes : (T) => Array[Byte]) = {
    //bits = log(m) == 2 *log(104/errPercent) = 2log(104) - 2*log(errPercent)
    def log2(x : Double) = scala.math.log(x)/scala.math.log(2.0)
    val bits = 2 * scala.math.ceil(log2(104) - log2(errPercent)).toInt
    implicit val hmm = new HyperLogLogMonoid(bits)
    mapPlusMap(f) { (t : T) => hmm.create(toBytes(t)) }
      { hll => fn(hmm,hll) }
  }

  /**
{'fraction_non_alphanumeric': 0.13561946902654867, 'fraction_numerical': 0.013053097345132743, 'mean_word_length': 2.819602872834812, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 39, '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': '2889559', 'n_tokens_mistral': 2989, 'n_tokens_neox': 2808, 'n_words': 902}