def METHOD_NAME(argv=None):
  """Main entry point; defines and runs the wordcount pipeline."""
  parser = argparse.ArgumentParser()
  parser.add_argument(
      '--input',
      dest='input',
      default='gs://dataflow-samples/shakespeare/kinglear.txt',
      help='Input file to process.')
  parser.add_argument(
      '--output',
      dest='output',
      required=True,
      help='Output file to write results to.')
  parser.add_argument(
      '--format',
      dest='format',
      default='text',
      help='Supported output file formats: %s.' % FORMATS)
  known_args, pipeline_args = parser.parse_known_args(argv)
  if known_args.format not in FORMATS:
    raise ValueError('--format should be one of: %s' % FORMATS)
  # We use the save_main_session option because one or more DoFn's in this
  # workflow rely on global context (e.g., a module imported at module level).
  pipeline_options = PipelineOptions(pipeline_args)
  pipeline_options.view_as(SetupOptions).save_main_session = True
  p = beam.Pipeline(options=pipeline_options)
  # Read the text file[pattern] into a PCollection.
  lines = p | 'read' >> ReadFromText(known_args.input)
  # Count the occurrences of each word.
  def count_ones(word_ones):
    (word, ones) = word_ones
    return (word, sum(ones))
  counts = (
      lines
      | 'split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(unicode))
      | 'pair_with_one' >> beam.Map(lambda x: (x, 1))
      | 'group' >> beam.GroupByKey()
      | 'count' >> beam.Map(count_ones))
  # Format the counts into a PCollection of strings.
  def format_text(word_count):
    (word, count) = word_count
    return '%s: %d' % (word, count)
  # Format the counts into a PCollection of dictionary strings.
  def format_dict(word_count):
    (word, count) = word_count
    row = dict(zip(HEADER, [word, count]))
    return row
  if known_args.format == 'text':
    output = counts | 'format text' >> beam.Map(format_text)
    # Write the output using a "Write" transform that has side effects.
    # pylint: disable=expression-not-assigned
    output | 'write text' >> WriteToText(known_args.output)
  elif known_args.format == 'avro':
    output = counts | 'format avro' >> beam.Map(format_dict)
    schema = avro.schema.parse(json.dumps(AVRO_SCHEMA))
    # Write the output using a "Write" transform that has side effects.
    # pylint: disable=expression-not-assigned
    output | 'write avro' >> WriteToAvro(
        file_path_prefix=known_args.output, schema=schema, codec=DEFAULT_CODEC)
  else:
    output = counts | 'format parquet' >> beam.Map(format_dict)
    # Write the output using a "Write" transform that has side effects.
    # pylint: disable=expression-not-assigned
    output | 'write parquet' >> WriteToParquet(
        file_path_prefix=known_args.output,
        schema=PARQUET_SCHEMA,
        codec=DEFAULT_CODEC)
  result = p.METHOD_NAME()
  result.wait_until_finish()
  # Do not query metrics when creating a template which doesn't run
  if (not hasattr(result, 'has_job')  # direct runner
      or result.has_job):  # not just a template creation
    empty_lines_filter = MetricsFilter().with_name('empty_lines')
    query_result = result.metrics().query(empty_lines_filter)
    if query_result['counters']:
      empty_lines_counter = query_result['counters'][0]
      logging.info('number of empty lines: %d', empty_lines_counter.result)
    word_lengths_filter = MetricsFilter().with_name('word_len_dist')
    query_result = result.metrics().query(word_lengths_filter)
    if query_result['distributions']:
      word_lengths_dist = query_result['distributions'][0]
      logging.info('average word length: %d', word_lengths_dist.result.mean)