title: Read a csv file, clean it, then write out the result as a csv using Apache Beam dataflow
body: <p>I would like to read a csv file, clean it, then write out the result as a csv using Apache Beam dataflow. The purpose is to make the file loadable into BigQuery. The cleaning rule is to simply escape a double quote with a double quote.
  My cleaning rule works. I’m having trouble incorporating it into a pipeline. I am seeking advice on what my cleaning function should return and how to call it through the pipeline.</p>
  <pre><code>import apache_beam as beam
  import csv
  import logging
  from apache_beam.options.pipeline_options import PipelineOptions
  from apache_beam.io import WriteToText
  
  lines = p | ReadFromText(file_pattern=&quot;gs://dev/clean_input/input01.csv&quot;)
  
  def parse_method(line):
      
      CSV_PARSING_KWARGS = {
          'doublequote': True,
          'escapechar': '\\',
          'quotechar': '&quot;',
          'delimiter': ','
      }
  
      reader = csv.reader(csv_file, CSV_PARSING_KWARGS)
      for rec in reader:
          cw = csv.writer(out_file, escapechar='&quot;', quoting=csv.QUOTE_MINIMAL)
          cw.writerow(rec)
          return rec
          
  
  def run(region, project, bucket, temploc ):
      argv = [
             # Passed in args 
             '--region={}'.format(region),
             '--project={}'.format(project),
             '--temp_location={}'.format(temploc),
             # Constructs
             '--staging_location=gs://{}/clean_input/stg/'.format(bucket),
             # Mandatory constants
             '--job_name=cleammycsv',
             '--runner=DataflowRunner'     
             ]
        
      options = PipelineOptions(
      flags=argv
      )
  
      pipeline = beam.Pipeline(options=options)
    
      clean_csv = (pipeline
      lines = lines| 'Read' &gt;&gt; beam.Map(parse_method)
      line = lines | 'Output to file' &gt;&gt; WriteToText(file_pattern=&quot;gs://dev/clean_output/output_file.csv&quot;)
      )   
      pipeline.run()
  
  if __name__ == '__main__':
     import argparse
     
     # Create the parser  
     parser = argparse.ArgumentParser(description='Run the CSV cleaning pipeline')   
  
     parser.add_argument('-r','--region', help='Region ID where data flow job to run', default='australia-southeast1')
     parser.add_argument('-p','--project', help='Unique project ID', required=True)
     parser.add_argument('-b','--bucket', help='Bucket name', required=True)
     parser.add_argument('-t','--temploc', help='Bucket name and folder', required=True)
     
     # Execute the parse_args() method
     args = vars(parser.parse_args())
  
     run(project=args['project'], bucket=args['bucket'], region=args['region'],temploc=args['temploc'])
  </code></pre>