def _StartHadoopMapReduce(self, gcs_dir):
    """Sends HTTP request to initiate Hadoop MapReduce for CSV transformation.

    Args:
      gcs_dir: Directory on Cloud Storage where inputs and outputs of MapReduce
          are stored.
    Returns:
      MapReduce ID generated by Hadoop master.
    Raises:
      HadoopError: When no Hadoop cluster is available.
    """
    http_request = urllib2.Request('http://%s/mapreduceasync' % self.master_ip)
    # To attach a file, Content-Type must be multipart/form-data.
    # Python libraries (urllib, urllib2, httplib) don't support
    # multipart/form-data by themselves.  Build multipart request manually.
    http_request.add_header('Content-Type',
                            'multipart/form-data; boundary=%s' % self.boundary)

    # gcs_dir starts with 'gs://'.  Remove it, since MapReduce script
    # expects path without prefix ('gs://')
    bare_gcs_dir = gcs_dir
    if gcs_dir.startswith('gs://'):
      bare_gcs_dir = gcs_dir[5:]

    mapreduce_params = {
        'mapper_type': 'file',
        'mapper_count': 5,
        'reducer_type': 'identity',
        'reducer_count': 0,
        'input': bare_gcs_dir + '/inputs/',
        'output': bare_gcs_dir + '/outputs/',
    }

    body = ''
    for key, value in mapreduce_params.items():
      body = self._AddParameter(body, key, value)

    # Attach mapper and reducer as files.
    body = self._AttachFile(
        body, 'mapper_file', 'datapipeline_transform.py', self.mapper)

    body += '--%s--\r\n' % self.boundary

    logging.debug(body)

    response = urllib2.urlopen(http_request, body, timeout=60)
    mapreduce_id = response.read()
    logging.info('MapReduce ID: %s', mapreduce_id)
    return mapreduce_id
dataset/ETHPy150Open GoogleCloudPlatform/Data-Pipeline/app/src/hadoop/hadoop_csv_transformer.py/HadoopCsvTransformer._StartHadoopMapReduce/original