<commit_before>package com.spotify.scio

import java.beans.Introspector
import java.io.File
import java.net.URI

import com.google.api.services.bigquery.model.TableReference
import com.google.api.services.datastore.DatastoreV1.{Entity, Query}
import com.google.cloud.dataflow.sdk.PipelineResult.State
import com.google.cloud.dataflow.sdk.coders.TableRowJsonCoder
import com.google.cloud.dataflow.sdk.io.{AvroIO => GAvroIO, BigQueryIO => GBigQueryIO, DatastoreIO => GDatastoreIO, PubsubIO => GPubsubIO, TextIO => GTextIO}
import com.google.cloud.dataflow.sdk.options.PipelineOptions.CheckEnabled
import com.google.cloud.dataflow.sdk.options.{DataflowPipelineOptions, PipelineOptions, PipelineOptionsFactory}
import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner
import com.google.cloud.dataflow.sdk.testing.TestPipeline
import com.google.cloud.dataflow.sdk.transforms.Combine.CombineFn
import com.google.cloud.dataflow.sdk.transforms.{Create, PTransform}
import com.google.cloud.dataflow.sdk.util.IntervalBoundedExponentialBackOff
import com.google.cloud.dataflow.sdk.values.{PBegin, PCollection, POutput, TimestampedValue}
import com.google.cloud.dataflow.sdk.{Pipeline, PipelineResult}
import com.spotify.scio.bigquery._
import com.spotify.scio.testing._
import com.spotify.scio.util.CallSites
import com.spotify.scio.values._
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.joda.time.Instant

import scala.collection.JavaConverters._
import scala.collection.mutable.{ListBuffer, Set => MSet}
import scala.concurrent.Future
import scala.concurrent.duration.{Duration, MINUTES, SECONDS}
import scala.reflect.ClassTag

/** Convenience object for creating [[ScioContext]] and [[Args]]. */
object ContextAndArgs {
  /** Create [[ScioContext]] and [[Args]] for command line arguments. */
  def apply(args: Array[String]): (ScioContext, Args) = {
    val (_opts, _args) = ScioContext.parseArguments[DataflowPipelineOptions](args)
    (new ScioContext(_opts, _args.optional("testId")), _args)
  }
}

/** Companion object for [[ScioContext]]. */
object ScioContext {

  /**
   * Create a new [[ScioContext]] instance.
   *
   * @param cmdlineArgs command line arguments including both Dataflow and job specific ones.
   * Dataflow specific ones will be parsed as
   * [[com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions DataflowPipelineOptions]] in
   * field `options`.
   */
  def apply(cmdlineArgs: Array[String]): ScioContext = {
    val (_opts, _args) = ScioContext.parseArguments[DataflowPipelineOptions](cmdlineArgs)
    new ScioContext(_opts, _args.optional("testId"))
  }

  /** Create a new [[ScioContext]] instance. */
  def apply(): ScioContext = ScioContext(Array.empty[String])

  /** Create a new [[ScioContext]] instance. */
  def apply(options: DataflowPipelineOptions): ScioContext = new ScioContext(options, None)


  /** Parse PipelineOptions and application arguments from command line arguments. */
  def parseArguments[T <: PipelineOptions : ClassTag](cmdlineArgs: Array[String]): (T, Args) = {
    val cls = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]
    val dfPatterns = cls.getMethods.flatMap { m =>
      val n = m.getName
      if ((!n.startsWith("get") && !n.startsWith("is")) ||
        m.getParameterTypes.nonEmpty || m.getReturnType == classOf[Unit]) {
        None
      } else {
        Some(Introspector.decapitalize(n.substring(if (n.startsWith("is")) 2 else 3)))
      }
    }.map(s => s"--$s($$|=)".r)
    val (dfArgs, appArgs) = cmdlineArgs.partition(arg => dfPatterns.exists(_.findFirstIn(arg).isDefined))

    (PipelineOptionsFactory.fromArgs(dfArgs).as(cls), Args(appArgs))
  }

}

/**
 * Main entry point for Dataflow functionality. A ScioContext represents a Dataflow pipeline,
 * and can be used to create SCollections and distributed caches on that cluster.
 *
 * @groupname accumulator Accumulators
 * @groupname dist_cache Distributed Cache
 * @groupname in_memory In-memory Collections
 * @groupname input Input Sources
 * @groupname Ungrouped Other Members
 */
class ScioContext private[scio] (val options: DataflowPipelineOptions, testId: Option[String]) {

  import Implicits._

  options.setAppName(CallSites.getAppName)

  /** Dataflow pipeline. */
  val pipeline: Pipeline = this.newPipeline()

  /* Mutable members */
  private var _isClosed: Boolean = false
  private val _callbacks: ListBuffer[State => Any] = ListBuffer.empty
  private val _accumulators: MSet[String] = MSet.empty

  /** Wrap a [[com.google.cloud.dataflow.sdk.values.PCollection PCollection]]. */
  def wrap[T: ClassTag](p: PCollection[T]): SCollection[T] =
    new SCollectionImpl[T](p, this)

  // =======================================================================
  // Miscellaneous
  // =======================================================================

  private lazy val bigQueryClient: BigQueryClient =
    BigQueryClient(options.getProject, options.getGcpCredential)

  private def newPipeline(): Pipeline = {
    val p = if (testId.isEmpty) {
      Pipeline.create(options)
    } else {
      TestPipeline.create()
    }
    p.getOptions.setStableUniqueNames(CheckEnabled.WARNING)
    p.getCoderRegistry.registerScalaCoders()
    p
  }

  // =======================================================================
  // States
  // =======================================================================

  /** Close the context. No operation can be performed once the context is closed. */
  def close(): ScioResult = {<commit_msg>fix app name and job name<commit_after>package com.spotify.scio

import java.beans.Introspector
import java.io.File
import java.net.URI

import com.google.api.services.bigquery.model.TableReference
import com.google.api.services.datastore.DatastoreV1.{Entity, Query}
import com.google.cloud.dataflow.sdk.PipelineResult.State
import com.google.cloud.dataflow.sdk.coders.TableRowJsonCoder
import com.google.cloud.dataflow.sdk.io.{AvroIO => GAvroIO, BigQueryIO => GBigQueryIO, DatastoreIO => GDatastoreIO, PubsubIO => GPubsubIO, TextIO => GTextIO}
import com.google.cloud.dataflow.sdk.options.PipelineOptions.CheckEnabled
import com.google.cloud.dataflow.sdk.options.{DataflowPipelineOptions, PipelineOptions, PipelineOptionsFactory}
import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner
import com.google.cloud.dataflow.sdk.testing.TestPipeline
import com.google.cloud.dataflow.sdk.transforms.Combine.CombineFn
import com.google.cloud.dataflow.sdk.transforms.{Create, PTransform}
import com.google.cloud.dataflow.sdk.util.IntervalBoundedExponentialBackOff
import com.google.cloud.dataflow.sdk.values.{PBegin, PCollection, POutput, TimestampedValue}
import com.google.cloud.dataflow.sdk.{Pipeline, PipelineResult}
import com.spotify.scio.bigquery._
import com.spotify.scio.testing._
import com.spotify.scio.util.CallSites
import com.spotify.scio.values._
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.joda.time.Instant

import scala.collection.JavaConverters._
import scala.collection.mutable.{ListBuffer, Set => MSet}
import scala.concurrent.Future
import scala.concurrent.duration.{Duration, MINUTES, SECONDS}
import scala.reflect.ClassTag

/** Convenience object for creating [[ScioContext]] and [[Args]]. */
object ContextAndArgs {
  /** Create [[ScioContext]] and [[Args]] for command line arguments. */
  def apply(args: Array[String]): (ScioContext, Args) = {
    val (_opts, _args) = ScioContext.parseArguments[DataflowPipelineOptions](args)
    (new ScioContext(_opts, _args.optional("testId")), _args)
  }
}

/** Companion object for [[ScioContext]]. */
object ScioContext {

  /**
   * Create a new [[ScioContext]] instance.
   *
   * @param cmdlineArgs command line arguments including both Dataflow and job specific ones.
   * Dataflow specific ones will be parsed as
   * [[com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions DataflowPipelineOptions]] in
   * field `options`.
   */
  def apply(cmdlineArgs: Array[String]): ScioContext = {
    val (_opts, _args) = ScioContext.parseArguments[DataflowPipelineOptions](cmdlineArgs)
    new ScioContext(_opts, _args.optional("testId"))
  }

  /** Create a new [[ScioContext]] instance. */
  def apply(): ScioContext = ScioContext(Array.empty[String])

  /** Create a new [[ScioContext]] instance. */
  def apply(options: DataflowPipelineOptions): ScioContext = new ScioContext(options, None)


  /** Parse PipelineOptions and application arguments from command line arguments. */
  def parseArguments[T <: PipelineOptions : ClassTag](cmdlineArgs: Array[String]): (T, Args) = {
    val cls = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]
    val dfPatterns = cls.getMethods.flatMap { m =>
      val n = m.getName
      if ((!n.startsWith("get") && !n.startsWith("is")) ||
        m.getParameterTypes.nonEmpty || m.getReturnType == classOf[Unit]) {
        None
      } else {
        Some(Introspector.decapitalize(n.substring(if (n.startsWith("is")) 2 else 3)))
      }
    }.map(s => s"--$s($$|=)".r)
    val (dfArgs, appArgs) = cmdlineArgs.partition(arg => dfPatterns.exists(_.findFirstIn(arg).isDefined))

    (PipelineOptionsFactory.fromArgs(dfArgs).as(cls), Args(appArgs))
  }

}

/**
 * Main entry point for Dataflow functionality. A ScioContext represents a Dataflow pipeline,
 * and can be used to create SCollections and distributed caches on that cluster.
 *
 * @groupname accumulator Accumulators
 * @groupname dist_cache Distributed Cache
 * @groupname in_memory In-memory Collections
 * @groupname input Input Sources
 * @groupname Ungrouped Other Members
 */
class ScioContext private[scio] (val options: DataflowPipelineOptions, testId: Option[String]) {

  import Implicits._

  /** Dataflow pipeline. */
  val pipeline: Pipeline = this.newPipeline()

  /* Mutable members */
  private var _isClosed: Boolean = false
  private val _callbacks: ListBuffer[State => Any] = ListBuffer.empty
  private val _accumulators: MSet[String] = MSet.empty

  /** Wrap a [[com.google.cloud.dataflow.sdk.values.PCollection PCollection]]. */
  def wrap[T: ClassTag](p: PCollection[T]): SCollection[T] =
    new SCollectionImpl[T](p, this)

  // =======================================================================
  // Miscellaneous
  // =======================================================================

  private lazy val bigQueryClient: BigQueryClient =
    BigQueryClient(options.getProject, options.getGcpCredential)

  private def newPipeline(): Pipeline = {
    // override app name and job name
    options.setAppName(CallSites.getAppName)
    options.setJobName(new DataflowPipelineOptions.JobNameFactory().create(options))

    val p = if (testId.isEmpty) {
      Pipeline.create(options)
    } else {
      TestPipeline.create()
    }
    p.getOptions.setStableUniqueNames(CheckEnabled.WARNING)
    p.getCoderRegistry.registerScalaCoders()
    p
  }

  // =======================================================================
  // States
  // =======================================================================

  /** Close the context. No operation can be performed once the context is closed. */
  def close(): ScioResult = {
{'fraction_non_alphanumeric': 0.1654061749265949, 'fraction_numerical': 0.0005338553252068689, 'mean_word_length': 4.817805383022774, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6642881', 'n_tokens_mistral': 3397, 'n_tokens_neox': 3302, 'n_words': 725}