code: def prepare_image_transforms(element, image_columns):
    
    import base64
    import cStringIO
    from PIL import Image
    from tensorflow.python.lib.io import file_io as tf_file_io
    from apache_beam.metrics import Metrics
  
    img_error_count = Metrics.counter('main', 'ImgErrorCount')
    img_missing_count = Metrics.counter('main', 'ImgMissingCount')
  
    for name in image_columns:
      uri = element[name]
      if not uri:
        img_missing_count.inc()
        continue
      try:
        with tf_file_io.FileIO(uri, 'r') as f:
          img = Image.open(f).convert('RGB')
  
      # A variety of different calling libraries throw different exceptions here.
      # They all correspond to an unreadable file so we treat them equivalently.
      # pylint: disable broad-except
      except Exception as e:
        logging.exception('Error processing image %s: %s', uri, str(e))
        img_error_count.inc()
        return
  
      # Convert to desired format and output.
      output = cStringIO.StringIO()
      img.save(output, 'jpeg')
      element[name] = base64.urlsafe_b64encode(output.getvalue())
  
    return element
docs: Replace an images url with its jpeg bytes.
  
    Args: 
      element: one input row, as a dict
      image_columns: list of columns that are image paths
  
    Return:
      element, where each image file path has been replaced by a base64 image.