<s> """<STR_LIT>""" <EOL> import json <EOL> import logging <EOL> import os . path <EOL> import re <EOL> import time <EOL> import urllib2 <EOL> import jinja2 <EOL> from src . clients import gcs <EOL> from src . hadoop import datastore <EOL> jinja_environment = jinja2 . Environment ( <EOL> loader = jinja2 . FileSystemLoader ( os . path . dirname ( __file__ ) ) ) <EOL> class HadoopError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class HadoopCsvTransformer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , config ) : <EOL> """<STR_LIT>""" <EOL> self . config = config <EOL> self . boundary = '<STR_LIT>' <EOL> self . cloud_storage_client = gcs . Gcs ( ) <EOL> cluster_query = datastore . ClusterInfo . query ( ) <EOL> if not cluster_query . count ( ) : <EOL> raise HadoopError ( '<STR_LIT>' ) <EOL> hadoop_cluster = cluster_query . fetch ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> logging . info ( '<STR_LIT>' , <EOL> hadoop_cluster . name ) <EOL> self . master_ip = hadoop_cluster . GetMasterIpAddress ( ) <EOL> def StartTransform ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _LoadMapper ( ) <EOL> gcs_dir = self . config [ '<STR_LIT>' ] <EOL> hadoop_input_filename = '<STR_LIT>' % gcs_dir <EOL> logging . info ( '<STR_LIT>' , <EOL> self . config [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , self . config [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> logging . debug ( '<STR_LIT>' , hadoop_input_filename ) <EOL> output_file = self . cloud_storage_client . OpenObject ( <EOL> self . config [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , mode = '<STR_LIT:w>' ) <EOL> input_file = self . cloud_storage_client . OpenObject ( self . config [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> hadoop_input = self . cloud_storage_client . OpenObject ( <EOL> hadoop_input_filename , mode = '<STR_LIT:w>' ) <EOL> line_count = <NUM_LIT:0> <EOL> for line in input_file : <EOL> if line_count < self . config [ '<STR_LIT>' ] : <EOL> output_file . write ( line ) <EOL> else : <EOL> hadoop_input . write ( line ) <EOL> line_count += <NUM_LIT:1> <EOL> hadoop_input . close ( ) <EOL> input_file . close ( ) <EOL> mapreduce_id = self . _StartHadoopMapReduce ( gcs_dir ) <EOL> self . _WaitForMapReduce ( mapreduce_id ) <EOL> bucket , hadoop_dir = gcs . Gcs . UrlToBucketAndName ( gcs_dir ) <EOL> tab_strip_pattern = re . compile ( '<STR_LIT>' ) <EOL> for hadoop_result in self . cloud_storage_client . ListBucket ( <EOL> '<STR_LIT>' % bucket , prefix = '<STR_LIT>' % hadoop_dir ) : <EOL> logging . debug ( '<STR_LIT>' , hadoop_result ) <EOL> hadoop_output = self . cloud_storage_client . OpenObject ( hadoop_result ) <EOL> for line in hadoop_output : <EOL> output_file . write ( tab_strip_pattern . sub ( '<STR_LIT:\n>' , line ) ) <EOL> output_file . close ( ) <EOL> def _LoadMapper ( self ) : <EOL> """<STR_LIT>""" <EOL> template = jinja_environment . get_template ( '<STR_LIT>' ) <EOL> self . mapper = template . render ( { '<STR_LIT>' : json . dumps ( self . config ) } ) <EOL> def _AddParameter ( self , body , name , value ) : <EOL> """<STR_LIT>""" <EOL> body += '<STR_LIT>' % self . boundary <EOL> body += '<STR_LIT>' % name <EOL> body += '<STR_LIT:\r\n>' <EOL> body += '<STR_LIT>' % value <EOL> return body <EOL> def _AttachFile ( self , body , name , filename , content ) : <EOL> """<STR_LIT>""" <EOL> body += '<STR_LIT>' % self . boundary <EOL> body += '<STR_LIT>' % ( <EOL> name , filename ) <EOL> body += '<STR_LIT>' <EOL> body += '<STR_LIT:\r\n>' <EOL> body += '<STR_LIT>' % content <EOL> return body <EOL> def _StartHadoopMapReduce ( self , gcs_dir ) : <EOL> """<STR_LIT>""" <EOL> http_request = urllib2 . Request ( '<STR_LIT>' % self . master_ip ) <EOL> http_request . add_header ( '<STR_LIT:Content-Type>' , <EOL> '<STR_LIT>' % self . boundary ) <EOL> bare_gcs_dir = gcs_dir <EOL> if gcs_dir . startswith ( '<STR_LIT>' ) : <EOL> bare_gcs_dir = gcs_dir [ <NUM_LIT:5> : ] <EOL> mapreduce_params = { <EOL> '<STR_LIT>' : '<STR_LIT:file>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT:input>' : bare_gcs_dir + '<STR_LIT>' , <EOL> '<STR_LIT>' : bare_gcs_dir + '<STR_LIT>' , <EOL> } <EOL> body = '<STR_LIT>' <EOL> for key , value in mapreduce_params . items ( ) : <EOL> body = self . _AddParameter ( body , key , value ) <EOL> body = self . _AttachFile ( <EOL> body , '<STR_LIT>' , '<STR_LIT>' , self . mapper ) <EOL> body += '<STR_LIT>' % self . boundary <EOL> logging . debug ( body ) <EOL> response = urllib2 . urlopen ( http_request , body , timeout = <NUM_LIT> ) <EOL> mapreduce_id = response . read ( ) <EOL> logging . info ( '<STR_LIT>' , mapreduce_id ) <EOL> return mapreduce_id <EOL> def _WaitForMapReduce ( self , mapreduce_id ) : <EOL> """<STR_LIT>""" <EOL> for _ in xrange ( <NUM_LIT> ) : <EOL> time . sleep ( <NUM_LIT:30> ) <EOL> try : <EOL> response = urllib2 . urlopen ( <EOL> '<STR_LIT>' % ( self . master_ip , mapreduce_id ) , <EOL> timeout = <NUM_LIT> ) <EOL> logging . debug ( response . read ( ) ) <EOL> return <EOL> except urllib2 . URLError as e : <EOL> if e . code == <NUM_LIT> : <EOL> logging . info ( '<STR_LIT>' ) <EOL> else : <EOL> logging . error ( '<STR_LIT>' , e . code , e . read ( ) ) <EOL> raise HadoopError ( '<STR_LIT>' ) <EOL> raise HadoopError ( '<STR_LIT>' ) </s>
