docs: def preprocess(pipeline, args):
    """Transfrom csv data into transfromed tf.example files.
  
    Outline:
      1) read the input data (as csv or bigquery) into a dict format
      2) replace image paths with base64 encoded image files
      3) build a csv input string with images paths replaced with base64. This 
         matches the serving csv that a trained model would expect.
      4) batch the csv strings
      5) run the transformations
      6) write the results to tf.example files and save any errors.
    """
    from tensorflow.python.lib.io import file_io
    from trainer import feature_transforms
  
    schema = json.loads(file_io.read_file_to_string(
        os.path.join(args.analysis, feature_transforms.SCHEMA_FILE)).decode())
    features = json.loads(file_io.read_file_to_string(
        os.path.join(args.analysis, feature_transforms.FEATURES_FILE)).decode())
    stats = json.loads(file_io.read_file_to_string(
        os.path.join(args.analysis, feature_transforms.STATS_FILE)).decode())
  
    column_names = [col['name'] for col in schema]
  
    if args.csv:
      all_files = []
      for i, file_pattern in enumerate(args.csv):
        all_files.append(pipeline | ('ReadCSVFile%d' % i) >> beam.io.ReadFromText(file_pattern))
      raw_data = (
          all_files
          | 'MergeCSVFiles' >> beam.Flatten()
          | 'ParseCSVData' >> beam.Map(decode_csv, column_names))
    else:
      columns = ', '.join(column_names)
      query = 'SELECT {columns} FROM `{table}`'.format(columns=columns,
                                                       table=args.bigquery)
      raw_data = (
          pipeline
          | 'ReadBiqQueryData'
          >> beam.io.Read(beam.io.BigQuerySource(query=query,
                                                 use_standard_sql=True)))
  
    # Note that prepare_image_transforms does not make embeddings, it justs reads
    # the image files and converts them to byte stings. TransformFeaturesDoFn()
    # will make the image embeddings.
    image_columns = image_transform_columns(features)
    
    clean_csv_data = (
        raw_data
        | 'PreprocessTransferredLearningTransformations'
        >> beam.Map(prepare_image_transforms, image_columns)
        | 'BuildCSVString'
        >> beam.Map(encode_csv, column_names))
  
    if args.shuffle:
      clean_csv_data = clean_csv_data | 'ShuffleData' >> shuffle()
  
    transform_dofn = TransformFeaturesDoFn(args.analysis, features, schema, stats)
    (transformed_data, errors) = (
         clean_csv_data
         | 'Batch Input' 
         >> beam.ParDo(EmitAsBatchDoFn(args.batch_size)) 
         | 'Run TF Graph on Batches' 
         >> beam.ParDo(transform_dofn).with_outputs('errors', main='main'))
  
    _ = (transformed_data
          | 'SerializeExamples' >> beam.Map(serialize_example, feature_transforms.get_transformed_feature_info(features, schema))
          | 'WriteExamples'
          >> beam.io.WriteToTFRecord(
              os.path.join(args.output, args.prefix),
              file_name_suffix='.tfrecord.gz'))
    _ = (errors
         | 'WriteErrors'
         >> beam.io.WriteToText(
             os.path.join(args.output, 'errors_' + args.prefix),
             file_name_suffix='.txt'))