code: def main(argv=None):
    
    args = parse_arguments(sys.argv if argv is None else argv)
    temp_dir = os.path.join(args.output, 'tmp')
  
    if args.cloud:
      pipeline_name = 'DataflowRunner'
    else:
      pipeline_name = 'DirectRunner'
      # Suppress TF warnings.
      os.environ['TF_CPP_MIN_LOG_LEVEL']='3'
  
    options = {
        'job_name': args.job_name,
        'temp_location': temp_dir,
        'project': args.project_id,
        'setup_file':
            os.path.abspath(os.path.join(
                os.path.dirname(__file__),
                'setup.py')),
    }
    if args.num_workers:
      options['num_workers'] = args.num_workers
    if args.worker_machine_type:
      options['worker_machine_type'] = args.worker_machine_type
  
    pipeline_options = beam.pipeline.PipelineOptions(flags=[], **options)
  
    p = beam.Pipeline(pipeline_name, options=pipeline_options)
    preprocess(pipeline=p, args=args)
    pipeline_result = p.run()
  
    if not args.async:
      pipeline_result.wait_until_finish()
    if args.async and args.cloud:
      print('View job at https://console.developers.google.com/dataflow/job/%s?project=%s' %
          (pipeline_result.job_id(), args.project_id))
docs: Run Preprocessing as a Dataflow.