<commit_before># Copyright 2017 Google Inc. All Rights Reserved.
#
# 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.

"""Create a viewer for dsub pipelines using Apache Beam.

dview can be used as a shell command or a python library.

By running the dview command at the top of a script that executes
multiple dsub commands, a live-updating execution graph can be
displayed in the Google Dataflow UI using Beam's --runner option.

For example usage see the README and dview_example.sh.
"""
from __future__ import absolute_import

import os
import sys
import argparse
import yaml
import getpass
import time
from datetime import datetime
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
import logging
from dsub.providers import provider_base

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProviderOptions(object):
  def __init__(self, **kwargs):
    self.provider = 'local'
    self.project = 'your-project-id'
    self.__dict__.update(kwargs)

class BeamOptions(object):
  def __init__(self, **kwargs):
    now = datetime.now().strftime('%m%d%H%M%S')
    self.name = ('dview-%s-%s' % (getpass.getuser(), now))
    self.setup_file = (os.path.dirname(os.path.realpath(sys.argv[0])) + '/setup.py')
    self.runner = 'direct'
    self.max_num_workers = '1'
    self.__dict__.update(kwargs)

  def __iter__(self):
    return iter(self.__dict__.keys())

def parse_args(argv):
  """Parse command-line arguments"""

  provider_defaults = ProviderOptions()
  beam_defaults = BeamOptions()

  parser = argparse.ArgumentParser(
      prog='dview',
      description='Create a viewer for dsub pipelines using Apache Beam.')

  parser.add_argument(
      '--dag',
      required=True,
      help='YAML graph of job names.')

  dstat = parser.add_argument_group(
      title='dsub/dstat',
      description='Options for dsub/dstat.')
  dstat.add_argument(
      '--provider',
      choices=['google', 'local', 'test-fails'],
      default=provider_defaults.provider,
      help='Service provider for batch jobs.')
  dstat.add_argument(
      '--dry-run',
      default=False,
      action='store_true',
      help='For testing, construct the workflow graph without calling any provider methods.')

  beam = parser.add_argument_group(
      title='Beam',
      description='Options for Apache Beam.')
  beam.add_argument(
      '--name',
      default=beam_defaults.name,
      help='Display name for the job.')
  beam.add_argument(
      '--runner',
      choices=['direct', 'dataflow'],
      default=beam_defaults.runner,
      help='Apache Beam runner.')

  dataflow = parser.add_argument_group(
      title='Google',
      description='Options for the Google provider or Dataflow runner.\
          The --project is required for both; all others apply to Dataflow only.')
  dataflow.add_argument(
      '--project',
      default=provider_defaults.project,
      help='Google Cloud project ID, required for Google provider or Dataflow.')
  dataflow.add_argument(
      '--temp-location',
      default='your-temp-path',
      help='Storage path for temp files.')
  dataflow.add_argument(
      '--setup-file',
      default=beam_defaults.setup_file,
      help='Path to setup.py in the dsub root directory, if not installed as a package.')
  dataflow.add_argument(
      '--extra-package',
      default='abs-path',
      help='Absolute path to the dsub package, if installed as a package.')
  dataflow.add_argument(
      '--max-num-workers',
      default=beam_defaults.max_num_workers,
      help='Maximum number of worker VMs.')

  known_args, pipeline_args = parser.parse_known_args(argv)

  # Dataflow pipeline args
  pipeline_args.extend([
      '--runner=' + known_args.runner,
      '--project=' + known_args.project,
      '--staging_location=' + known_args.temp_location,
      '--temp_location=' + known_args.temp_location,
      '--max_num_workers=' + known_args.max_num_workers,
      '--job_name=' + known_args.name
  ])

  return known_args, pipeline_args

class MergeBranches(beam.PTransform):
  """Merge branches in the DAG"""

  def expand(self, pcoll):<commit_msg>cleanup<commit_after># Copyright 2017 Google Inc. All Rights Reserved.
#
# 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.

"""Create a viewer for dsub pipelines using Apache Beam.

dview can be used as a shell command or a python library.

By running the dview command at the top of a script that executes
multiple dsub commands, a live-updating execution graph can be
displayed in the Google Dataflow UI using Beam's --runner option.

For example usage see the README and dview_example.sh.
"""
from __future__ import absolute_import

import os
import sys
import argparse
import getpass
import logging
import time
import yaml
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from datetime import datetime
from dsub.providers import provider_base

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProviderOptions(object):
  def __init__(self, **kwargs):
    self.provider = 'local'
    self.project = 'your-project-id'
    self.__dict__.update(kwargs)

class BeamOptions(object):
  def __init__(self, **kwargs):
    now = datetime.now().strftime('%m%d%H%M%S')
    self.name = ('dview-%s-%s' % (getpass.getuser(), now))
    self.setup_file = (os.path.dirname(os.path.realpath(sys.argv[0])) + '/setup.py')
    self.runner = 'direct'
    self.max_num_workers = '1'
    self.__dict__.update(kwargs)

  def __iter__(self):
    return iter(self.__dict__.keys())

def parse_args(argv):
  """Parse command-line arguments"""

  provider_defaults = ProviderOptions()
  beam_defaults = BeamOptions()

  parser = argparse.ArgumentParser(
      prog='dview',
      description='Create a viewer for dsub pipelines using Apache Beam.')

  parser.add_argument(
      '--dag',
      required=True,
      help='YAML graph of job names.')

  dstat = parser.add_argument_group(
      title='dsub/dstat',
      description='Options for dsub/dstat.')
  dstat.add_argument(
      '--provider',
      choices=['google', 'local', 'test-fails'],
      default=provider_defaults.provider,
      help='Service provider for batch jobs.')
  dstat.add_argument(
      '--dry-run',
      default=False,
      action='store_true',
      help='For testing, construct the workflow graph without calling any provider methods.')

  beam = parser.add_argument_group(
      title='Beam',
      description='Options for Apache Beam.')
  beam.add_argument(
      '--name',
      default=beam_defaults.name,
      help='Display name for the job.')
  beam.add_argument(
      '--runner',
      choices=['direct', 'dataflow'],
      default=beam_defaults.runner,
      help='Apache Beam runner.')

  dataflow = parser.add_argument_group(
      title='Google',
      description='Options for the Google provider or Dataflow runner.\
          The --project is required for both; all others apply to Dataflow only.')
  dataflow.add_argument(
      '--project',
      default=provider_defaults.project,
      help='Google Cloud project ID, required for Google provider or Dataflow.')
  dataflow.add_argument(
      '--temp-location',
      default='your-temp-path',
      help='Storage path for temp files.')
  dataflow.add_argument(
      '--setup-file',
      default=beam_defaults.setup_file,
      help='Path to setup.py in the dsub root directory, if not installed as a package.')
  dataflow.add_argument(
      '--extra-package',
      default='abs-path',
      help='Absolute path to the dsub package, if installed as a package.')
  dataflow.add_argument(
      '--max-num-workers',
      default=beam_defaults.max_num_workers,
      help='Maximum number of worker VMs.')

  known_args, beam_args = parser.parse_known_args(argv)

  beam_args.extend([
      '--runner=' + known_args.runner,
      '--project=' + known_args.project,
      '--staging_location=' + known_args.temp_location,
      '--temp_location=' + known_args.temp_location,
      '--max_num_workers=' + known_args.max_num_workers,
      '--job_name=' + known_args.name
  ])

  return known_args, beam_args

class MergeBranches(beam.PTransform):
  """Merge branches in the DAG"""

  def expand(self, pcoll):
{'fraction_non_alphanumeric': 0.1086701503679215, 'fraction_numerical': 0.0021328783192918845, 'mean_word_length': 3.621981271562346, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1604262', 'n_tokens_mistral': 2796, 'n_tokens_neox': 2722, 'n_words': 836}