def StartTransform(self):
    """Starts CSV transformation on Hadoop cluster."""
    self._LoadMapper()

    gcs_dir = self.config['hadoopTmpDir']
    hadoop_input_filename = '%s/inputs/input.csv' % gcs_dir

    logging.info('Starting Hadoop transform from %s to %s',
                 self.config['sources'][0], self.config['sinks'][0])
    logging.debug('Hadoop input file: %s', hadoop_input_filename)

    output_file = self.cloud_storage_client.OpenObject(
        self.config['sinks'][0], mode='w')
    # Copy input file to Cloud Storage MapReduce input directory.
    input_file = self.cloud_storage_client.OpenObject(self.config['sources'][0])
    hadoop_input = self.cloud_storage_client.OpenObject(
        hadoop_input_filename, mode='w')

    line_count = 0
    for line in input_file:
      if line_count < self.config['skipLeadingRows']:
        output_file.write(line)
      else:
        hadoop_input.write(line)
      line_count += 1

    hadoop_input.close()
    input_file.close()

    mapreduce_id = self._StartHadoopMapReduce(gcs_dir)

    self._WaitForMapReduce(mapreduce_id)

    bucket, hadoop_dir = gcs.Gcs.UrlToBucketAndName(gcs_dir)

    tab_strip_pattern = re.compile('\t\r?\n')

    for hadoop_result in self.cloud_storage_client.ListBucket(
        '/%s' % bucket, prefix='%s/outputs/part-' % hadoop_dir):
      logging.debug('Hadoop result file: %s', hadoop_result)
      hadoop_output = self.cloud_storage_client.OpenObject(hadoop_result)
      for line in hadoop_output:
        # Since Hadoop MapReduce output is key-value pair separated by a tab,
        # and Hadoop transformer always outputs a line as a key,
        # trim the trailing tab before EOL.
        output_file.write(tab_strip_pattern.sub('\n', line))

    output_file.close()
dataset/ETHPy150Open GoogleCloudPlatform/Data-Pipeline/app/src/hadoop/hadoop_csv_transformer.py/HadoopCsvTransformer.StartTransform/original