<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 com.twitter.chill.config.{ScalaMapConfig, ConfiguredInstantiator}

import cascading.pipe.assembly.AggregateBy
import cascading.flow.{Flow, FlowDef, FlowProps, FlowListener, FlowSkipStrategy, FlowStepStrategy}
import cascading.pipe.Pipe
import cascading.property.AppProps
import cascading.tuple.collect.SpillableProps

import org.apache.hadoop.io.serializer.{Serialization => HSerialization}

//For java -> scala implicits on collections
import scala.collection.JavaConversions._

import java.util.Calendar
import java.util.concurrent.{Executors, TimeUnit, ThreadFactory, Callable, TimeoutException}
import java.util.concurrent.atomic.AtomicInteger
import java.security.MessageDigest

object Job {
  // Uses reflection to create a job by name
  def apply(jobName : String, args : Args) : Job =
    Class.forName(jobName).
      getConstructor(classOf[Args]).
      newInstance(args).
      asInstanceOf[Job]
}

class Job(val args : Args) extends FieldConversions with java.io.Serializable {
  // Set specific Mode
  implicit def mode : Mode = Mode.getMode(args).getOrElse(sys.error("No Mode defined"))

  /**
  * you should never call these directly, there are here to make
  * the DSL work.  Just know, you can treat a Pipe as a RichPipe and
  * vice-versa within a Job
  */
  implicit def p2rp(pipe : Pipe) = new RichPipe(pipe)
  implicit def rp2p(rp : RichPipe) = rp.pipe
  implicit def source2rp(src : Source) : RichPipe = RichPipe(src.read)

  // This converts an interable into a Source with index (int-based) fields
  implicit def iterToSource[T](iter : Iterable[T])(implicit set: TupleSetter[T], conv : TupleConverter[T]) : Source = {
    IterableSource[T](iter)(set, conv)
  }
  //
  implicit def iterToPipe[T](iter : Iterable[T])(implicit set: TupleSetter[T], conv : TupleConverter[T]) : Pipe = {
    iterToSource(iter)(set, conv).read
  }
  implicit def iterToRichPipe[T](iter : Iterable[T])
    (implicit set: TupleSetter[T], conv : TupleConverter[T]) : RichPipe = {
    RichPipe(iterToPipe(iter)(set, conv))
  }

  // Override this if you want change how the mapred.job.name is written in Hadoop
  def name : String = getClass.getName

  //This is the FlowDef used by all Sources this job creates
  @transient
  implicit protected val flowDef = {
    val fd = new FlowDef
    fd.setName(name)
    fd
  }

  /** Copy this job
   * By default, this uses reflection and the single argument Args constructor
   */
  def clone(nextargs : Args) : Job = {
    this.getClass
    .getConstructor(classOf[Args])
    .newInstance(nextargs)
    .asInstanceOf[Job]
  }

  /**
  * Implement this method if you want some other jobs to run after the current
  * job. These will not execute until the current job has run successfully.
  */
  def next : Option[Job] = None

  /** Keep 100k tuples in memory by default before spilling
   * Turn this up as high as you can without getting OOM.
   *
   * This is ignored if there is a value set in the incoming mode.config
   */
  def defaultSpillThreshold: Int = 100 * 1000

  def fromInputStream(s: java.io.InputStream): Array[Byte] =
    Stream.continually(s.read).takeWhile(-1 !=).map(_.toByte).toArray

  def toHexString(bytes: Array[Byte]): String =
    bytes.map("%02X".format(_)).mkString

  def md5Hex(bytes: Array[Byte]): String = {
    val md = MessageDigest.getInstance("MD5")
    md.update(bytes)
    toHexString(md.digest)
  }

  // Generated the MD5 hex of the the bytes in the job classfile
  lazy val classIdentifier : String = {
    val classAsPath = getClass.getName.replace(".", "/") + ".class"
    val is = getClass.getClassLoader.getResourceAsStream(classAsPath)
    val bytes = fromInputStream(is)
    is.close()
    md5Hex(bytes)
  }

  /** This is the exact config that is passed to the Cascading FlowConnector.
   * By default:
   *   if there are no spill thresholds in mode.config, we replace with defaultSpillThreshold
   *   we overwrite io.serializations with ioSerializations
   *   we overwrite cascading.tuple.element.comparator.default to defaultComparator
   *   we add some scalding keys for debugging/logging
   *
   * Tip: override this method, call super, and ++ your additional
   * map to add or overwrite more options
   */
  def config: Map[AnyRef,AnyRef] = {
    // These are ignored if set in mode.config
    val lowPriorityDefaults =
      Map(SpillableProps.LIST_THRESHOLD -> defaultSpillThreshold.toString,
          SpillableProps.MAP_THRESHOLD -> defaultSpillThreshold.toString,
          AggregateBy.AGGREGATE_BY_THRESHOLD -> defaultSpillThreshold.toString
          )
    // Set up the keys for chill
    val chillConf = ScalaMapConfig(lowPriorityDefaults)
    ConfiguredInstantiator.setReflect(chillConf, classOf[serialization.KryoHadoop])

    chillConf.toMap ++
      mode.config ++
      // Optionally set a default Comparator
      (defaultComparator match {
        case Some(defcomp) => Map(FlowProps.DEFAULT_ELEMENT_COMPARATOR -> defcomp.getName)
        case None => Map.empty[AnyRef, AnyRef]
      }) ++
      Map(
        "io.serializations" -> ioSerializations.map { _.getName }.mkString(","),
        "scalding.version" -> "0.9.0-SNAPSHOT",
        "cascading.app.name" -> name,
        "cascading.app.id" -> name,
        "scalding.flow.class.name" -> getClass.getName,
        "scalding.flow.class.signature" -> classIdentifier,
        "scalding.job.args" -> args.toString,
        "scalding.flow.submitted.timestamp" ->
          Calendar.getInstance().getTimeInMillis().toString
      )
  }

  def skipStrategy: Option[FlowSkipStrategy] = None

  def stepStrategy: Option[FlowStepStrategy[_]] = None

  /**
   * combine the config, flowDef and the Mode to produce a flow
   */
  def buildFlow: Flow[_] = {
    val flow = mode.newFlowConnector(config).connect(flowDef)
    listeners.foreach { flow.addListener(_) }
    skipStrategy.foreach { flow.setFlowSkipStrategy(_) }
    stepStrategy.foreach { flow.setFlowStepStrategy(_) }

    // set the framework property for cascading
    AppProps.addApplicationFramework(new java.util.HashMap(config),
      String.format("scalding:%s", config.get("scalding.version").get))

    flow
  }

  // called before run
  // only override if you do not use flowDef
  def validate {
    FlowStateMap.validateSources(flowDef, mode)
  }

  // called after successfull run
  // only override if you do not use flowDef
  def clear {
    FlowStateMap.clear(flowDef)
  }

  //Override this if you need to do some extra processing other than complete the flow
  def run : Boolean = {
    val flow = buildFlow
    flow.complete
    flow.getFlowStats.isSuccessful
  }

  //override this to add any listeners you need
  def listeners : List[FlowListener] = Nil

  /** The exact list of Hadoop serializations passed into the config
   * These replace the config serializations
   * Cascading tuple serialization should be in this list, and probably
   * before any custom code
   */
  def ioSerializations: List[Class[_ <: HSerialization[_]]] = List(<commit_msg>set the cascading framework system property instead of using the API, since the API has side effects.<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 com.twitter.chill.config.{ScalaMapConfig, ConfiguredInstantiator}

import cascading.pipe.assembly.AggregateBy
import cascading.flow.{Flow, FlowDef, FlowProps, FlowListener, FlowSkipStrategy, FlowStepStrategy}
import cascading.pipe.Pipe
import cascading.property.AppProps
import cascading.tuple.collect.SpillableProps

import org.apache.hadoop.io.serializer.{Serialization => HSerialization}

//For java -> scala implicits on collections
import scala.collection.JavaConversions._

import java.util.Calendar
import java.util.concurrent.{Executors, TimeUnit, ThreadFactory, Callable, TimeoutException}
import java.util.concurrent.atomic.AtomicInteger
import java.security.MessageDigest

object Job {
  // Uses reflection to create a job by name
  def apply(jobName : String, args : Args) : Job =
    Class.forName(jobName).
      getConstructor(classOf[Args]).
      newInstance(args).
      asInstanceOf[Job]
}

class Job(val args : Args) extends FieldConversions with java.io.Serializable {
  // Set specific Mode
  implicit def mode : Mode = Mode.getMode(args).getOrElse(sys.error("No Mode defined"))

  /**
  * you should never call these directly, there are here to make
  * the DSL work.  Just know, you can treat a Pipe as a RichPipe and
  * vice-versa within a Job
  */
  implicit def p2rp(pipe : Pipe) = new RichPipe(pipe)
  implicit def rp2p(rp : RichPipe) = rp.pipe
  implicit def source2rp(src : Source) : RichPipe = RichPipe(src.read)

  // This converts an interable into a Source with index (int-based) fields
  implicit def iterToSource[T](iter : Iterable[T])(implicit set: TupleSetter[T], conv : TupleConverter[T]) : Source = {
    IterableSource[T](iter)(set, conv)
  }
  //
  implicit def iterToPipe[T](iter : Iterable[T])(implicit set: TupleSetter[T], conv : TupleConverter[T]) : Pipe = {
    iterToSource(iter)(set, conv).read
  }
  implicit def iterToRichPipe[T](iter : Iterable[T])
    (implicit set: TupleSetter[T], conv : TupleConverter[T]) : RichPipe = {
    RichPipe(iterToPipe(iter)(set, conv))
  }

  // Override this if you want change how the mapred.job.name is written in Hadoop
  def name : String = getClass.getName

  //This is the FlowDef used by all Sources this job creates
  @transient
  implicit protected val flowDef = {
    val fd = new FlowDef
    fd.setName(name)
    fd
  }

  /** Copy this job
   * By default, this uses reflection and the single argument Args constructor
   */
  def clone(nextargs : Args) : Job = {
    this.getClass
    .getConstructor(classOf[Args])
    .newInstance(nextargs)
    .asInstanceOf[Job]
  }

  /**
  * Implement this method if you want some other jobs to run after the current
  * job. These will not execute until the current job has run successfully.
  */
  def next : Option[Job] = None

  /** Keep 100k tuples in memory by default before spilling
   * Turn this up as high as you can without getting OOM.
   *
   * This is ignored if there is a value set in the incoming mode.config
   */
  def defaultSpillThreshold: Int = 100 * 1000

  def fromInputStream(s: java.io.InputStream): Array[Byte] =
    Stream.continually(s.read).takeWhile(-1 !=).map(_.toByte).toArray

  def toHexString(bytes: Array[Byte]): String =
    bytes.map("%02X".format(_)).mkString

  def md5Hex(bytes: Array[Byte]): String = {
    val md = MessageDigest.getInstance("MD5")
    md.update(bytes)
    toHexString(md.digest)
  }

  // Generated the MD5 hex of the the bytes in the job classfile
  lazy val classIdentifier : String = {
    val classAsPath = getClass.getName.replace(".", "/") + ".class"
    val is = getClass.getClassLoader.getResourceAsStream(classAsPath)
    val bytes = fromInputStream(is)
    is.close()
    md5Hex(bytes)
  }

  /** This is the exact config that is passed to the Cascading FlowConnector.
   * By default:
   *   if there are no spill thresholds in mode.config, we replace with defaultSpillThreshold
   *   we overwrite io.serializations with ioSerializations
   *   we overwrite cascading.tuple.element.comparator.default to defaultComparator
   *   we add some scalding keys for debugging/logging
   *
   * Tip: override this method, call super, and ++ your additional
   * map to add or overwrite more options
   */
  def config: Map[AnyRef,AnyRef] = {
    // These are ignored if set in mode.config
    val lowPriorityDefaults =
      Map(SpillableProps.LIST_THRESHOLD -> defaultSpillThreshold.toString,
          SpillableProps.MAP_THRESHOLD -> defaultSpillThreshold.toString,
          AggregateBy.AGGREGATE_BY_THRESHOLD -> defaultSpillThreshold.toString
          )
    // Set up the keys for chill
    val chillConf = ScalaMapConfig(lowPriorityDefaults)
    ConfiguredInstantiator.setReflect(chillConf, classOf[serialization.KryoHadoop])

    val scaldingVersion = "0.9-SNAPSHOT"
    System.setProperty(AppProps.APP_FRAMEWORKS,
          String.format("scalding:%s", scaldingVersion))

    chillConf.toMap ++
      mode.config ++
      // Optionally set a default Comparator
      (defaultComparator match {
        case Some(defcomp) => Map(FlowProps.DEFAULT_ELEMENT_COMPARATOR -> defcomp.getName)
        case None => Map.empty[AnyRef, AnyRef]
      }) ++
      Map(
        "io.serializations" -> ioSerializations.map { _.getName }.mkString(","),
        "scalding.version" -> scaldingVersion,
        "cascading.app.name" -> name,
        "cascading.app.id" -> name,
        "scalding.flow.class.name" -> getClass.getName,
        "scalding.flow.class.signature" -> classIdentifier,
        "scalding.job.args" -> args.toString,
        "scalding.flow.submitted.timestamp" ->
          Calendar.getInstance().getTimeInMillis().toString
      )
  }

  def skipStrategy: Option[FlowSkipStrategy] = None

  def stepStrategy: Option[FlowStepStrategy[_]] = None

  /**
   * combine the config, flowDef and the Mode to produce a flow
   */
  def buildFlow: Flow[_] = {
    val flow = mode.newFlowConnector(config).connect(flowDef)
    listeners.foreach { flow.addListener(_) }
    skipStrategy.foreach { flow.setFlowSkipStrategy(_) }
    stepStrategy.foreach { flow.setFlowStepStrategy(_) }
    flow
  }

  // called before run
  // only override if you do not use flowDef
  def validate {
    FlowStateMap.validateSources(flowDef, mode)
  }

  // called after successfull run
  // only override if you do not use flowDef
  def clear {
    FlowStateMap.clear(flowDef)
  }

  //Override this if you need to do some extra processing other than complete the flow
  def run : Boolean = {
    val flow = buildFlow
    flow.complete
    flow.getFlowStats.isSuccessful
  }

  //override this to add any listeners you need
  def listeners : List[FlowListener] = Nil

  /** The exact list of Hadoop serializations passed into the config
   * These replace the config serializations
   * Cascading tuple serialization should be in this list, and probably
   * before any custom code
   */
  def ioSerializations: List[Class[_ <: HSerialization[_]]] = List(
{'fraction_non_alphanumeric': 0.09518814737118134, 'fraction_numerical': 0.003998951094794808, 'mean_word_length': 3.515985790408526, 'pattern_counts': {'":': 0, '<': 5, '<?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': '6200410', 'n_tokens_mistral': 4614, 'n_tokens_neox': 4498, 'n_words': 1405}