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

// Jackson
// import com.github.fge.jsonschema.core.report.ProcessingMessage

// Scalaz
import scalaz._
import Scalaz._

// Scalding
import com.twitter.scalding._

// Snowplow Common Enrich
import common._
import common.FatalEtlError
import common.outputs.CanonicalOutput

// This project
import inputs.EnrichedEventLoader
import shredder.Shredder
import outputs.{
  BadRow,
  ShreddedPartition
}
import utils.ProcessingMessageUtils

/**
 * Helpers for our data processing pipeline.
 */
object ShredJob {

  import utils.ProcessingMessageUtils._

  /**
   * Pipelines our loading of raw lines into
   * shredding the JSONs.
   *
   * @param line The incoming raw line (hopefully
   *        holding a Snowplow enriched event)
   * @return a Validation boxing either a Nel of
   *         ProcessingMessages on Failure, or a
   *         (possibly empty) List of JSON instances
   *         + schemas on Success
   */
  def loadAndShred(line: String): ValidatedJsonSchemaPairList =
    for {
      event <- EnrichedEventLoader.toEnrichedEvent(line).toProcessingMessages
      shred <- Shredder.shred(event)
    } yield shred

  /**
   * Isolates our Failures into a Some; Successes
   * become a None will be silently dropped by
   * Scalding in this pipeline.
   *
   * @param all The Validation containing either
   *        our Successes or our Failures
   * @return an Option boxing either our List of
   *         Processing Messages on Failure, or
   *         None on Success
   */
  def isolateBads(all: ValidatedJsonSchemaPairList): MaybeProcMsgNel =
    all.fold(
      e => Some(e), // Nel -> Some(List) of ProcessingMessages
      c => None)
}

/**
 * The Snowplow Shred job, written in Scalding
 * (the Scala DSL on top of Cascading).
 */ 
class ShredJob(args : Args) extends Job(args) {
<commit_msg>Added an isolateGoods function alongside isolateBads<commit_after>/*
 * Copyright (c) 2012 Twitter, Inc.
 *
 * This program is licensed to you under the Apache License Version 2.0,
 * and you may not use this file except in compliance with the Apache License Version 2.0.
 * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the Apache License Version 2.0 is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
 */
package com.snowplowanalytics.snowplow.enrich
package hadoop

// Jackson
// import com.github.fge.jsonschema.core.report.ProcessingMessage

// Scalaz
import scalaz._
import Scalaz._

// Scalding
import com.twitter.scalding._

// Snowplow Common Enrich
import common._
import common.FatalEtlError
import common.outputs.CanonicalOutput

// This project
import inputs.EnrichedEventLoader
import shredder.Shredder
import outputs.{
  BadRow,
  ShreddedPartition
}
import utils.ProcessingMessageUtils

/**
 * Helpers for our data processing pipeline.
 */
object ShredJob {

  import utils.ProcessingMessageUtils._

  /**
   * Pipelines our loading of raw lines into
   * shredding the JSONs.
   *
   * @param line The incoming raw line (hopefully
   *        holding a Snowplow enriched event)
   * @return a Validation boxing either a Nel of
   *         ProcessingMessages on Failure, or a
   *         (possibly empty) List of JSON instances
   *         + schemas on Success
   */
  def loadAndShred(line: String): ValidatedJsonSchemaPairList =
    for {
      event <- EnrichedEventLoader.toEnrichedEvent(line).toProcessingMessages
      shred <- Shredder.shred(event)
    } yield shred

  /**
   * Isolates our Failures into a Some; Successes
   * become a None will be silently dropped by
   * Scalding in this pipeline.
   *
   * @param all The Validation containing either
   *        our Successes or our Failures
   * @return an Option boxing either our List of
   *         Processing Messages on Failure, or
   *         None on Success
   */
  def isolateBads(all: ValidatedJsonSchemaPairList): MaybeProcMsgNel =
    all.fold(
      e => Some(e), // Nel -> Some(List) of ProcessingMessages
      c => None)    // Discard

  /**
   * Isolates our Failures into a Some; Successes
   * become a None will be silently dropped by
   * Scalding in this pipeline.
   *
   * @param all The Validation containing either
   *        our Successes or our Failures
   * @return an Option boxing either our List of
   *         Processing Messages on Failure, or
   *         None on Success
   */
  def isolateGoods(all: ValidatedJsonSchemaPairList): Option[List[JsonSchemaPair]] =
    all.fold(
      e => None, // Discard
      c => Some(c)) // List -> Some(List) of JsonSchemaPairs

}

/**
 * The Snowplow Shred job, written in Scalding
 * (the Scala DSL on top of Cascading).
 */ 
class ShredJob(args : Args) extends Job(args) {

{'fraction_non_alphanumeric': 0.07514854945823139, 'fraction_numerical': 0.005592450192240475, 'mean_word_length': 3.1172661870503595, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 12, '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': '4159690', 'n_tokens_mistral': 1668, 'n_tokens_neox': 1603, 'n_words': 622}