docs: def parse_arguments(argv):
    """Parse command line arguments.
    Args:
      argv: list of command line arguments including program name.
    Returns:
      The parsed arguments as returned by argparse.ArgumentParser.
    """
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=textwrap.dedent("""\
            Runs preprocessing on raw data for TensorFlow training.
  
            This script applies some transformations to raw data to improve
            training performance. Some data transformations can be expensive
            such as the tf-idf text column transformation. During training, the
            same raw data row might be used multiply times to train a model. This
            means the same transformations are applied to the same data row
            multiple times. This can be very inefficient, so this script applies
            partial transformations to the raw data and writes an intermediate 
            preprocessed datasource to disk for training. 
  
            Running this transformation step is required for two usage paths:
              1) If the img_url_to_vec transform is used. This is because
                 preprocessing as image is expensive and TensorFlow cannot easily
                 read raw image files during training.
              2) If the raw data is in BigQuery. TensorFlow cannot read from a 
                 BigQuery source.
  
            Running this transformation step is recommended if a text transform is
            used (like tf-idf or bag-of-words), and the text value for each row
            is very long.
  
            Running this transformation step may not have an interesting training
            performance impact if the transforms are all simple like scaling
            numerical values."""))
  
    source_group = parser.add_mutually_exclusive_group(required=True)
  
    source_group.add_argument(
        '--csv',
        metavar='FILE',
        required=False,
        action='append',
        help='CSV data to transform.')
  
    source_group.add_argument(
        '--bigquery',
        metavar='PROJECT_ID.DATASET.TABLE_NAME',
        type=str,
        required=False,
        help=('Must be in the form `project.dataset.table_name`. BigQuery '
              'data to transform'))
  
    parser.add_argument(
        '--analysis',
        metavar='ANALYSIS_OUTPUT_DIR',
        required=True,
        help='The output folder of analyze')
  
    parser.add_argument(
        '--prefix',
        metavar='OUTPUT_FILENAME_PREFIX',
        required=True,
        type=str)
  
    parser.add_argument(
        '--output',
        metavar='DIR',
        default=None,
        required=True,
        help=('Google Cloud Storage or Local directory in which '
              'to place outputs.'))
  
    parser.add_argument(
        '--shuffle',
        action='store_true',
        default=False,
        help='If used, data source is shuffled. This is recommended for training data.')
  
    parser.add_argument(
        '--batch-size',
        metavar='N',
        type=int,
        default=100,
        help='Larger values increase performance and peak memory usage.')
  
    cloud_group = parser.add_argument_group(
        title='Cloud Parameters',
        description='These parameters are only used if --cloud is used.')
  
    cloud_group.add_argument(
        '--cloud',
        action='store_true',
        help='Run preprocessing on the cloud.')
  
    cloud_group.add_argument(
        '--job-name',
        type=str,
        help='Unique dataflow job name.')
  
    cloud_group.add_argument(
        '--project-id',
        help='The project to which the job will be submitted.')
  
    cloud_group.add_argument(
        '--num-workers',
        metavar='N',
        type=int,
        default=0,
        help='Set to 0 to use the default size determined by the Dataflow service.')
  
    cloud_group.add_argument(
        '--worker-machine-type',
        metavar='NAME',
        type=str,
        help='A machine name from https://cloud.google.com/compute/docs/machine-types. '
             ' If not given, the service uses the default machine type.')
  
    cloud_group.add_argument(
        '--async',
        action='store_true',
        help='If used, this script returns before the dataflow job is completed.')
  
    args = parser.parse_args(args=argv[1:])
  
    if args.cloud and not args.project_id:
      raise ValueError('--project-id is needed for --cloud')
  
    if args.async and not args.cloud:
      raise ValueError('--async should only be used with --cloud')
  
    if not args.job_name:
      args.job_name = ('dataflow-job-{}'.format(
          datetime.datetime.now().strftime('%Y%m%d%H%M%S')))
    return args