<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.typed

import java.io.Serializable

import com.twitter.algebird.{Semigroup, MapAlgebra, Monoid, Ring, Aggregator}

import com.twitter.scalding.TupleConverter.{singleConverter, tuple2Converter, CTupleConverter, TupleEntryConverter}
import com.twitter.scalding.TupleSetter.{singleSetter, tup2Setter}

import com.twitter.scalding._

import cascading.flow.FlowDef
import cascading.pipe.Pipe
import cascading.tuple.{Fields, Tuple => CTuple, TupleEntry}
import util.Random

/**
 * factory methods for TypedPipe, which is the typed representation of distributed lists in scalding.
 * This object is here rather than in the typed package because a lot of code was written using
 * the functions in the object, which we do not see how to hide with package object tricks.
 */
object TypedPipe extends Serializable {
  def from[T](pipe: Pipe, fields: Fields)(implicit conv: TupleConverter[T]): TypedPipe[T] =
    TypedPipeInst[T](pipe, fields, Converter(conv))

  def from[T](mappable: TypedSource[T])(implicit flowDef: FlowDef, mode: Mode): TypedPipe[T] =
    TypedPipeInst[T](mappable.read, mappable.sourceFields, Converter(mappable.converter))

  def from[T](iter: Iterable[T])(implicit flowDef: FlowDef, mode: Mode): TypedPipe[T] =
    IterablePipe[T](iter.view, flowDef, mode)

  /** Input must be a Pipe with exactly one Field */
  def fromSingleField[T](pipe: Pipe): TypedPipe[T] =
    TypedPipeInst[T](pipe, new Fields(0), Converter(singleConverter[T]))

  def empty(implicit flowDef: FlowDef, mode: Mode): TypedPipe[Nothing] =
    EmptyTypedPipe(flowDef, mode)
}

/** Think of a TypedPipe as a distributed unordered list that may or may not yet
 * have been materialized in memory or disk.
 *
 * Represents a phase in a distributed computation on an input data source
 * Wraps a cascading Pipe object, and holds the transformation done up until that point
 */
trait TypedPipe[+T] extends Serializable {

  // Implements a cross product.  The right side should be tiny
  def cross[U](tiny: TypedPipe[U]): TypedPipe[(T,U)]

  def flatMap[U](f: T => TraversableOnce[U]): TypedPipe[U]

  /** If you are going to create two branches or forks,
   * it may be more efficient to call this method first
   * which will create a node in the cascading graph.
   * Without this, both full branches of the fork will be
   * put into separate cascading pipes, which can, in some cases,
   * be slower.
   *
   * Ideally the planner would see this
   */
  def fork: TypedPipe[T]

  /** limit the output to at most count items.
   * useful for debugging, but probably that's about it.
   * The number may be less than count, and not sampled particular method
   */
  def limit(count: Int): TypedPipe[T]

  /** This does a sum of values WITHOUT triggering a shuffle.
   * the contract is, if followed by a group.sum the result is the same
   * with or without this present, and it never increases the number of
   * items. BUT due to the cost of caching, it might not be faster if
   * there is poor key locality.
   *
   * It is only useful for expert tuning,
   * and best avoided unless you are struggling with performance problems.
   * If you are not sure you need this, you probably don't.
   *
   * The main use case is to reduce the values down before a key expansion
   * such as is often done in a data cube.
   */
  def sumByLocalKeys[K,V](implicit ev : T <:< (K,V), sg: Semigroup[V]): TypedPipe[(K,V)]

  def sample(percent : Double): TypedPipe[T]
  def sample(percent : Double, seed : Long): TypedPipe[T]

  /** Export back to a raw cascading Pipe. useful for interop with the scalding
   * Fields API or with Cascading code.
   */
  def toPipe[U >: T](fieldNames: Fields)(implicit setter: TupleSetter[U]): Pipe

/////////////////////////////////////////////
//
// The following have default implementations in terms of the above
//
/////////////////////////////////////////////

  import Dsl._

  def ++[U >: T](other: TypedPipe[U]): TypedPipe[U] = other match {
    case EmptyTypedPipe(_,_) => this
    case IterablePipe(thatIter,_,_) if thatIter.isEmpty => this
    case _ => MergedTypedPipe(this, other)
  }

  /** Same as groupAll.aggregate.values
   */
  def aggregate[B,C](agg: Aggregator[T, B, C]): ValuePipe[C] =
    ComputedValue(groupAll.aggregate(agg).values)

  def collect[U](fn: PartialFunction[T, U]): TypedPipe[U] =
    filter(fn.isDefinedAt(_)).map(fn(_))

  def cross[V](p: ValuePipe[V]) : TypedPipe[(T, V)] =
    p match {
      case e@EmptyValue() => e.toTypedPipe
      case LiteralValue(v) => map { (_, v) }
      case ComputedValue(pipe) => map(((), _)).hashJoin(pipe.groupAll).values
    }

  // prints the current pipe to stdout
  def debug: TypedPipe[T] = map { t => println(t); t }

  /**
   * Returns the set of distinct elements in the TypedPipe
   */
  @annotation.implicitNotFound(msg = "For distinct method to work, the type in TypedPipe must have an Ordering.")
  def distinct(implicit ord: Ordering[_ >: T]): TypedPipe[T] = {
    // cast because Ordering is not contravariant, but should be (and this cast is safe)
    implicit val ordT: Ordering[T] = ord.asInstanceOf[Ordering[T]]
    map{ (_, ()) }.group.sum.keys
  }

  def either[R](that: TypedPipe[R]): TypedPipe[Either[T, R]] =
    map(Left(_)) ++ (that.map(Right(_)))

  /** Sometimes useful for implementing custom joins with groupBy + mapValueStream when you know
   * that the value/key can fit in memory. Beware.
   */
  def eitherValues[K,V,R](that: TypedPipe[(K, R)])(implicit ev: T <:< (K,V)): TypedPipe[(K, Either[V, R])] =
    mapValues { (v: V) => Left(v) } ++ (that.mapValues { (r: R) => Right(r) })

  def map[U](f: T => U): TypedPipe[U] = flatMap { t => Iterable(f(t)) }<commit_msg>remove code duplication is cross<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.typed

import java.io.Serializable

import com.twitter.algebird.{Semigroup, MapAlgebra, Monoid, Ring, Aggregator}

import com.twitter.scalding.TupleConverter.{singleConverter, tuple2Converter, CTupleConverter, TupleEntryConverter}
import com.twitter.scalding.TupleSetter.{singleSetter, tup2Setter}

import com.twitter.scalding._

import cascading.flow.FlowDef
import cascading.pipe.Pipe
import cascading.tuple.{Fields, Tuple => CTuple, TupleEntry}
import util.Random

/**
 * factory methods for TypedPipe, which is the typed representation of distributed lists in scalding.
 * This object is here rather than in the typed package because a lot of code was written using
 * the functions in the object, which we do not see how to hide with package object tricks.
 */
object TypedPipe extends Serializable {
  def from[T](pipe: Pipe, fields: Fields)(implicit conv: TupleConverter[T]): TypedPipe[T] =
    TypedPipeInst[T](pipe, fields, Converter(conv))

  def from[T](mappable: TypedSource[T])(implicit flowDef: FlowDef, mode: Mode): TypedPipe[T] =
    TypedPipeInst[T](mappable.read, mappable.sourceFields, Converter(mappable.converter))

  def from[T](iter: Iterable[T])(implicit flowDef: FlowDef, mode: Mode): TypedPipe[T] =
    IterablePipe[T](iter.view, flowDef, mode)

  /** Input must be a Pipe with exactly one Field */
  def fromSingleField[T](pipe: Pipe): TypedPipe[T] =
    TypedPipeInst[T](pipe, new Fields(0), Converter(singleConverter[T]))

  def empty(implicit flowDef: FlowDef, mode: Mode): TypedPipe[Nothing] =
    EmptyTypedPipe(flowDef, mode)
}

/** Think of a TypedPipe as a distributed unordered list that may or may not yet
 * have been materialized in memory or disk.
 *
 * Represents a phase in a distributed computation on an input data source
 * Wraps a cascading Pipe object, and holds the transformation done up until that point
 */
trait TypedPipe[+T] extends Serializable {

  // Implements a cross product.  The right side should be tiny
  def cross[U](tiny: TypedPipe[U]): TypedPipe[(T,U)]

  def flatMap[U](f: T => TraversableOnce[U]): TypedPipe[U]

  /** If you are going to create two branches or forks,
   * it may be more efficient to call this method first
   * which will create a node in the cascading graph.
   * Without this, both full branches of the fork will be
   * put into separate cascading pipes, which can, in some cases,
   * be slower.
   *
   * Ideally the planner would see this
   */
  def fork: TypedPipe[T]

  /** limit the output to at most count items.
   * useful for debugging, but probably that's about it.
   * The number may be less than count, and not sampled particular method
   */
  def limit(count: Int): TypedPipe[T]

  /** This does a sum of values WITHOUT triggering a shuffle.
   * the contract is, if followed by a group.sum the result is the same
   * with or without this present, and it never increases the number of
   * items. BUT due to the cost of caching, it might not be faster if
   * there is poor key locality.
   *
   * It is only useful for expert tuning,
   * and best avoided unless you are struggling with performance problems.
   * If you are not sure you need this, you probably don't.
   *
   * The main use case is to reduce the values down before a key expansion
   * such as is often done in a data cube.
   */
  def sumByLocalKeys[K,V](implicit ev : T <:< (K,V), sg: Semigroup[V]): TypedPipe[(K,V)]

  def sample(percent : Double): TypedPipe[T]
  def sample(percent : Double, seed : Long): TypedPipe[T]

  /** Export back to a raw cascading Pipe. useful for interop with the scalding
   * Fields API or with Cascading code.
   */
  def toPipe[U >: T](fieldNames: Fields)(implicit setter: TupleSetter[U]): Pipe

/////////////////////////////////////////////
//
// The following have default implementations in terms of the above
//
/////////////////////////////////////////////

  import Dsl._

  def ++[U >: T](other: TypedPipe[U]): TypedPipe[U] = other match {
    case EmptyTypedPipe(_,_) => this
    case IterablePipe(thatIter,_,_) if thatIter.isEmpty => this
    case _ => MergedTypedPipe(this, other)
  }

  /** Same as groupAll.aggregate.values
   */
  def aggregate[B,C](agg: Aggregator[T, B, C]): ValuePipe[C] =
    ComputedValue(groupAll.aggregate(agg).values)

  def collect[U](fn: PartialFunction[T, U]): TypedPipe[U] =
    filter(fn.isDefinedAt(_)).map(fn(_))

  def cross[V](p: ValuePipe[V]) : TypedPipe[(T, V)] =
    p match {
      case e@EmptyValue() => e.toTypedPipe
      case LiteralValue(v) => map { (_, v) }
      case ComputedValue(pipe) => cross(pipe)
    }

  // prints the current pipe to stdout
  def debug: TypedPipe[T] = map { t => println(t); t }

  /**
   * Returns the set of distinct elements in the TypedPipe
   */
  @annotation.implicitNotFound(msg = "For distinct method to work, the type in TypedPipe must have an Ordering.")
  def distinct(implicit ord: Ordering[_ >: T]): TypedPipe[T] = {
    // cast because Ordering is not contravariant, but should be (and this cast is safe)
    implicit val ordT: Ordering[T] = ord.asInstanceOf[Ordering[T]]
    map{ (_, ()) }.group.sum.keys
  }

  def either[R](that: TypedPipe[R]): TypedPipe[Either[T, R]] =
    map(Left(_)) ++ (that.map(Right(_)))

  /** Sometimes useful for implementing custom joins with groupBy + mapValueStream when you know
   * that the value/key can fit in memory. Beware.
   */
  def eitherValues[K,V,R](that: TypedPipe[(K, R)])(implicit ev: T <:< (K,V)): TypedPipe[(K, Either[V, R])] =
    mapValues { (v: V) => Left(v) } ++ (that.mapValues { (r: R) => Right(r) })

  def map[U](f: T => U): TypedPipe[U] = flatMap { t => Iterable(f(t)) }
{'fraction_non_alphanumeric': 0.12287831699737031, 'fraction_numerical': 0.0017531277392620927, 'mean_word_length': 3.688083675756444, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 35, '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': '680411', 'n_tokens_mistral': 4058, 'n_tokens_neox': 3900, 'n_words': 1439}