<s> """<STR_LIT>""" <EOL> __all__ = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> import base64 <EOL> import copy <EOL> import logging <EOL> import pickle <EOL> import random <EOL> import string <EOL> import StringIO <EOL> import time <EOL> import zipfile <EOL> from google . net . proto import ProtocolBuffer <EOL> from google . appengine . ext import ndb <EOL> from google . appengine . api import datastore <EOL> from google . appengine . api import logservice <EOL> from google . appengine . api . logservice import log_service_pb <EOL> from google . appengine . ext import blobstore <EOL> from google . appengine . ext import db <EOL> from google . appengine . ext import key_range <EOL> from google . appengine . ext . db import metadata <EOL> from google . appengine . ext . mapreduce import context <EOL> from google . appengine . ext . mapreduce import datastore_range_iterators as db_iters <EOL> from google . appengine . ext . mapreduce import errors <EOL> from google . appengine . ext . mapreduce import json_util <EOL> from google . appengine . ext . mapreduce import key_ranges <EOL> from google . appengine . ext . mapreduce import kv_pb <EOL> from google . appengine . ext . mapreduce import model <EOL> from google . appengine . ext . mapreduce import namespace_range <EOL> from google . appengine . ext . mapreduce import operation <EOL> from google . appengine . ext . mapreduce import property_range <EOL> from google . appengine . ext . mapreduce import records <EOL> from google . appengine . ext . mapreduce import util <EOL> try : <EOL> cloudstorage = None <EOL> from google . appengine . _internal import cloudstorage <EOL> if hasattr ( cloudstorage , "<STR_LIT>" ) : <EOL> cloudstorage = None <EOL> except ImportError : <EOL> pass <EOL> if cloudstorage is None : <EOL> try : <EOL> import cloudstorage <EOL> except ImportError : <EOL> pass <EOL> Error = errors . Error <EOL> BadReaderParamsError = errors . BadReaderParamsError <EOL> COUNTER_IO_READ_BYTES = "<STR_LIT>" <EOL> COUNTER_IO_READ_MSEC = "<STR_LIT>" <EOL> ALLOW_CHECKPOINT = util . ALLOW_CHECKPOINT <EOL> class InputReader ( json_util . JsonMixin ) : <EOL> """<STR_LIT>""" <EOL> expand_parameters = False <EOL> _APP_PARAM = "<STR_LIT>" <EOL> NAMESPACE_PARAM = "<STR_LIT>" <EOL> NAMESPACES_PARAM = "<STR_LIT>" <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % self . __class__ ) <EOL> @ classmethod <EOL> def from_json ( cls , input_shard_state ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % cls ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % cls ) <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> def _get_params ( mapper_spec , allowed_keys = None , allow_old = True ) : <EOL> """<STR_LIT>""" <EOL> if "<STR_LIT>" not in mapper_spec . params : <EOL> message = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if not allow_old or allowed_keys : <EOL> raise errors . BadReaderParamsError ( message ) <EOL> params = mapper_spec . params <EOL> params = dict ( ( str ( n ) , v ) for n , v in params . iteritems ( ) ) <EOL> else : <EOL> if not isinstance ( mapper_spec . params . get ( "<STR_LIT>" ) , dict ) : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" ) <EOL> params = mapper_spec . params . get ( "<STR_LIT>" ) <EOL> params = dict ( ( str ( n ) , v ) for n , v in params . iteritems ( ) ) <EOL> if allowed_keys : <EOL> params_diff = set ( params . keys ( ) ) - allowed_keys <EOL> if params_diff : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" % "<STR_LIT:U+002C>" . join ( params_diff ) ) <EOL> return params <EOL> class AbstractDatastoreInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> _BATCH_SIZE = <NUM_LIT:50> <EOL> _MAX_SHARD_COUNT = <NUM_LIT> <EOL> MAX_NAMESPACES_FOR_KEY_SHARD = <NUM_LIT:10> <EOL> ENTITY_KIND_PARAM = "<STR_LIT>" <EOL> KEYS_ONLY_PARAM = "<STR_LIT>" <EOL> BATCH_SIZE_PARAM = "<STR_LIT>" <EOL> KEY_RANGE_PARAM = "<STR_LIT>" <EOL> FILTERS_PARAM = "<STR_LIT>" <EOL> WHOLE_EG_PARAM = "<STR_LIT>" <EOL> _KEY_RANGE_ITER_CLS = db_iters . AbstractKeyRangeIterator <EOL> def __init__ ( self , iterator ) : <EOL> """<STR_LIT>""" <EOL> self . _iter = iterator <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> for o in self . _iter : <EOL> yield o <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return repr ( self . _iter ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _iter . to_json ( ) <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> return cls ( db_iters . RangeIteratorFactory . from_json ( json ) ) <EOL> @ classmethod <EOL> def _get_query_spec ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> entity_kind = params [ cls . ENTITY_KIND_PARAM ] <EOL> filters = params . get ( cls . FILTERS_PARAM ) <EOL> app = params . get ( cls . _APP_PARAM ) <EOL> ns = params . get ( cls . NAMESPACE_PARAM ) <EOL> return model . QuerySpec ( <EOL> entity_kind = cls . _get_raw_entity_kind ( entity_kind ) , <EOL> keys_only = bool ( params . get ( cls . KEYS_ONLY_PARAM , False ) ) , <EOL> filters = filters , <EOL> batch_size = int ( params . get ( cls . BATCH_SIZE_PARAM , cls . _BATCH_SIZE ) ) , <EOL> model_class_path = entity_kind , <EOL> app = app , <EOL> ns = ns ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> shard_count = mapper_spec . shard_count <EOL> query_spec = cls . _get_query_spec ( mapper_spec ) <EOL> params = _get_params ( mapper_spec ) <EOL> shard_whole_eg = params . get ( cls . WHOLE_EG_PARAM , False ) <EOL> namespaces = None <EOL> if query_spec . ns is not None : <EOL> k_ranges = cls . _to_key_ranges_by_shard ( <EOL> query_spec . app , [ query_spec . ns ] , shard_count , query_spec , <EOL> shard_whole_eg = shard_whole_eg ) <EOL> else : <EOL> ns_keys = namespace_range . get_namespace_keys ( <EOL> query_spec . app , cls . MAX_NAMESPACES_FOR_KEY_SHARD + <NUM_LIT:1> ) <EOL> if not ns_keys : <EOL> return <EOL> elif len ( ns_keys ) <= cls . MAX_NAMESPACES_FOR_KEY_SHARD : <EOL> namespaces = [ ns_key . name ( ) or "<STR_LIT>" for ns_key in ns_keys ] <EOL> k_ranges = cls . _to_key_ranges_by_shard ( <EOL> query_spec . app , namespaces , shard_count , query_spec , <EOL> shard_whole_eg = shard_whole_eg ) <EOL> else : <EOL> ns_ranges = namespace_range . NamespaceRange . split ( n = shard_count , <EOL> contiguous = False , <EOL> can_query = lambda : True , <EOL> _app = query_spec . app ) <EOL> k_ranges = [ key_ranges . KeyRangesFactory . create_from_ns_range ( ns_range ) <EOL> for ns_range in ns_ranges ] <EOL> iters = [ db_iters . RangeIteratorFactory . create_key_ranges_iterator ( <EOL> r , query_spec , cls . _KEY_RANGE_ITER_CLS ) for r in k_ranges ] <EOL> return [ cls ( i ) for i in iters ] <EOL> @ classmethod <EOL> def _to_key_ranges_by_shard ( cls , app , namespaces , shard_count , query_spec , <EOL> shard_whole_eg = False ) : <EOL> """<STR_LIT>""" <EOL> key_ranges_by_ns = [ ] <EOL> for namespace in namespaces : <EOL> ranges = cls . _split_ns_by_scatter ( <EOL> shard_count , <EOL> namespace , <EOL> query_spec . entity_kind , <EOL> app , <EOL> shard_whole_eg = shard_whole_eg ) <EOL> random . shuffle ( ranges ) <EOL> key_ranges_by_ns . append ( ranges ) <EOL> ranges_by_shard = [ [ ] for _ in range ( shard_count ) ] <EOL> for ranges in key_ranges_by_ns : <EOL> for i , k_range in enumerate ( ranges ) : <EOL> if k_range : <EOL> ranges_by_shard [ i ] . append ( k_range ) <EOL> key_ranges_by_shard = [ ] <EOL> for ranges in ranges_by_shard : <EOL> if ranges : <EOL> key_ranges_by_shard . append ( key_ranges . KeyRangesFactory . create_from_list ( <EOL> ranges ) ) <EOL> return key_ranges_by_shard <EOL> @ classmethod <EOL> def _split_ns_by_scatter ( cls , shard_count , namespace , raw_entity_kind , app , <EOL> shard_whole_eg = False ) : <EOL> """<STR_LIT>""" <EOL> if shard_count == <NUM_LIT:1> : <EOL> return [ key_range . KeyRange ( namespace = namespace , _app = app ) ] <EOL> ds_query = datastore . Query ( kind = raw_entity_kind , <EOL> namespace = namespace , <EOL> _app = app , <EOL> keys_only = True ) <EOL> ds_query . Order ( "<STR_LIT>" ) <EOL> oversampling_factor = <NUM_LIT:32> <EOL> random_keys = ds_query . Get ( shard_count * oversampling_factor ) <EOL> if not random_keys : <EOL> return ( [ key_range . KeyRange ( namespace = namespace , _app = app ) ] + <EOL> [ None ] * ( shard_count - <NUM_LIT:1> ) ) <EOL> random_keys . sort ( ) <EOL> if shard_whole_eg : <EOL> random_keys = cls . align_to_entity_groups ( random_keys ) <EOL> if len ( random_keys ) >= shard_count : <EOL> random_keys = cls . _choose_split_points ( random_keys , shard_count ) <EOL> k_ranges = [ ] <EOL> k_ranges . append ( key_range . KeyRange ( <EOL> key_start = None , <EOL> key_end = random_keys [ <NUM_LIT:0> ] , <EOL> direction = key_range . KeyRange . ASC , <EOL> include_start = False , <EOL> include_end = False , <EOL> namespace = namespace , <EOL> _app = app ) ) <EOL> for i in range ( <NUM_LIT:0> , len ( random_keys ) - <NUM_LIT:1> ) : <EOL> k_ranges . append ( key_range . KeyRange ( <EOL> key_start = random_keys [ i ] , <EOL> key_end = random_keys [ i + <NUM_LIT:1> ] , <EOL> direction = key_range . KeyRange . ASC , <EOL> include_start = True , <EOL> include_end = False , <EOL> namespace = namespace , <EOL> _app = app ) ) <EOL> k_ranges . append ( key_range . KeyRange ( <EOL> key_start = random_keys [ - <NUM_LIT:1> ] , <EOL> key_end = None , <EOL> direction = key_range . KeyRange . ASC , <EOL> include_start = True , <EOL> include_end = False , <EOL> namespace = namespace , <EOL> _app = app ) ) <EOL> if len ( k_ranges ) < shard_count : <EOL> k_ranges += [ None ] * ( shard_count - len ( k_ranges ) ) <EOL> return k_ranges <EOL> @ classmethod <EOL> def _choose_split_points ( cls , sorted_keys , shard_count ) : <EOL> """<STR_LIT>""" <EOL> assert len ( sorted_keys ) >= shard_count <EOL> index_stride = len ( sorted_keys ) / float ( shard_count ) <EOL> return [ sorted_keys [ int ( round ( index_stride * i ) ) ] <EOL> for i in range ( <NUM_LIT:1> , shard_count ) ] <EOL> @ classmethod <EOL> def align_to_entity_groups ( cls , keys ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> for key in keys : <EOL> if key is None : <EOL> break <EOL> eg_key = key . entity_group ( ) <EOL> if not result or result [ - <NUM_LIT:1> ] != eg_key : <EOL> result . append ( eg_key ) <EOL> return result <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . ENTITY_KIND_PARAM not in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if cls . BATCH_SIZE_PARAM in params : <EOL> try : <EOL> batch_size = int ( params [ cls . BATCH_SIZE_PARAM ] ) <EOL> if batch_size < <NUM_LIT:1> : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % batch_size ) <EOL> except ValueError , e : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % e ) <EOL> if cls . NAMESPACE_PARAM in params : <EOL> if not isinstance ( params [ cls . NAMESPACE_PARAM ] , <EOL> ( str , unicode , type ( None ) ) ) : <EOL> raise BadReaderParamsError ( <EOL> "<STR_LIT>" ) <EOL> if cls . NAMESPACES_PARAM in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if cls . FILTERS_PARAM in params : <EOL> filters = params [ cls . FILTERS_PARAM ] <EOL> if not isinstance ( filters , list ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> for f in filters : <EOL> if not isinstance ( f , ( tuple , list ) ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , f ) <EOL> if len ( f ) != <NUM_LIT:3> : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , f ) <EOL> prop , op , _ = f <EOL> if not isinstance ( prop , basestring ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , prop ) <EOL> if not isinstance ( op , basestring ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , op ) <EOL> @ classmethod <EOL> def _get_raw_entity_kind ( cls , entity_kind_or_model_classpath ) : <EOL> """<STR_LIT>""" <EOL> return entity_kind_or_model_classpath <EOL> class RawDatastoreInputReader ( AbstractDatastoreInputReader ) : <EOL> """<STR_LIT>""" <EOL> _KEY_RANGE_ITER_CLS = db_iters . KeyRangeEntityIterator <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> super ( RawDatastoreInputReader , cls ) . validate ( mapper_spec ) <EOL> params = _get_params ( mapper_spec ) <EOL> entity_kind = params [ cls . ENTITY_KIND_PARAM ] <EOL> if "<STR_LIT:.>" in entity_kind : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> entity_kind , cls . __name__ ) <EOL> if cls . FILTERS_PARAM in params : <EOL> filters = params [ cls . FILTERS_PARAM ] <EOL> for f in filters : <EOL> if f [ <NUM_LIT:1> ] != "<STR_LIT:=>" : <EOL> raise BadReaderParamsError ( <EOL> "<STR_LIT>" , f ) <EOL> class DatastoreInputReader ( AbstractDatastoreInputReader ) : <EOL> """<STR_LIT>""" <EOL> _KEY_RANGE_ITER_CLS = db_iters . KeyRangeModelIterator <EOL> @ classmethod <EOL> def _get_raw_entity_kind ( cls , model_classpath ) : <EOL> entity_type = util . for_name ( model_classpath ) <EOL> if isinstance ( entity_type , db . Model ) : <EOL> return entity_type . kind ( ) <EOL> elif isinstance ( entity_type , ( ndb . Model , ndb . MetaModel ) ) : <EOL> return entity_type . _get_kind ( ) <EOL> else : <EOL> return util . get_short_name ( model_classpath ) <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> super ( DatastoreInputReader , cls ) . validate ( mapper_spec ) <EOL> params = _get_params ( mapper_spec ) <EOL> entity_kind = params [ cls . ENTITY_KIND_PARAM ] <EOL> try : <EOL> model_class = util . for_name ( entity_kind ) <EOL> except ImportError , e : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % e ) <EOL> if cls . FILTERS_PARAM in params : <EOL> filters = params [ cls . FILTERS_PARAM ] <EOL> if issubclass ( model_class , db . Model ) : <EOL> cls . _validate_filters ( filters , model_class ) <EOL> else : <EOL> cls . _validate_filters_ndb ( filters , model_class ) <EOL> property_range . PropertyRange ( filters , entity_kind ) <EOL> @ classmethod <EOL> def _validate_filters ( cls , filters , model_class ) : <EOL> """<STR_LIT>""" <EOL> if not filters : <EOL> return <EOL> properties = model_class . properties ( ) <EOL> for f in filters : <EOL> prop , _ , val = f <EOL> if prop not in properties : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" , <EOL> prop , model_class . kind ( ) ) <EOL> try : <EOL> properties [ prop ] . validate ( val ) <EOL> except db . BadValueError , e : <EOL> raise errors . BadReaderParamsError ( e ) <EOL> @ classmethod <EOL> def _validate_filters_ndb ( cls , filters , model_class ) : <EOL> """<STR_LIT>""" <EOL> if not filters : <EOL> return <EOL> properties = model_class . _properties <EOL> for f in filters : <EOL> prop , _ , val = f <EOL> if prop not in properties : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" , <EOL> prop , model_class . _get_kind ( ) ) <EOL> try : <EOL> properties [ prop ] . _do_validate ( val ) <EOL> except db . BadValueError , e : <EOL> raise errors . BadReaderParamsError ( e ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> shard_count = mapper_spec . shard_count <EOL> query_spec = cls . _get_query_spec ( mapper_spec ) <EOL> if not property_range . should_shard_by_property_range ( query_spec . filters ) : <EOL> return super ( DatastoreInputReader , cls ) . split_input ( mapper_spec ) <EOL> p_range = property_range . PropertyRange ( query_spec . filters , <EOL> query_spec . model_class_path ) <EOL> p_ranges = p_range . split ( shard_count ) <EOL> if query_spec . ns : <EOL> ns_range = namespace_range . NamespaceRange ( <EOL> namespace_start = query_spec . ns , <EOL> namespace_end = query_spec . ns , <EOL> _app = query_spec . app ) <EOL> ns_ranges = [ copy . copy ( ns_range ) for _ in p_ranges ] <EOL> else : <EOL> ns_keys = namespace_range . get_namespace_keys ( <EOL> query_spec . app , cls . MAX_NAMESPACES_FOR_KEY_SHARD + <NUM_LIT:1> ) <EOL> if not ns_keys : <EOL> return <EOL> if len ( ns_keys ) <= cls . MAX_NAMESPACES_FOR_KEY_SHARD : <EOL> ns_ranges = [ namespace_range . NamespaceRange ( _app = query_spec . app ) <EOL> for _ in p_ranges ] <EOL> else : <EOL> ns_ranges = namespace_range . NamespaceRange . split ( n = shard_count , <EOL> contiguous = False , <EOL> can_query = lambda : True , <EOL> _app = query_spec . app ) <EOL> p_ranges = [ copy . copy ( p_range ) for _ in ns_ranges ] <EOL> assert len ( p_ranges ) == len ( ns_ranges ) <EOL> iters = [ <EOL> db_iters . RangeIteratorFactory . create_property_range_iterator ( <EOL> p , ns , query_spec ) for p , ns in zip ( p_ranges , ns_ranges ) ] <EOL> return [ cls ( i ) for i in iters ] <EOL> class DatastoreKeyInputReader ( RawDatastoreInputReader ) : <EOL> """<STR_LIT>""" <EOL> _KEY_RANGE_ITER_CLS = db_iters . KeyRangeKeyIterator <EOL> DatastoreEntityInputReader = RawDatastoreInputReader <EOL> class _OldAbstractDatastoreInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> _BATCH_SIZE = <NUM_LIT:50> <EOL> _MAX_SHARD_COUNT = <NUM_LIT> <EOL> _OVERSAMPLING_FACTOR = <NUM_LIT:32> <EOL> MAX_NAMESPACES_FOR_KEY_SHARD = <NUM_LIT:10> <EOL> ENTITY_KIND_PARAM = "<STR_LIT>" <EOL> KEYS_ONLY_PARAM = "<STR_LIT>" <EOL> BATCH_SIZE_PARAM = "<STR_LIT>" <EOL> KEY_RANGE_PARAM = "<STR_LIT>" <EOL> NAMESPACE_RANGE_PARAM = "<STR_LIT>" <EOL> CURRENT_KEY_RANGE_PARAM = "<STR_LIT>" <EOL> FILTERS_PARAM = "<STR_LIT>" <EOL> def __init__ ( self , <EOL> entity_kind , <EOL> key_ranges = None , <EOL> ns_range = None , <EOL> batch_size = _BATCH_SIZE , <EOL> current_key_range = None , <EOL> filters = None ) : <EOL> """<STR_LIT>""" <EOL> assert key_ranges is not None or ns_range is not None , ( <EOL> "<STR_LIT>" ) <EOL> assert key_ranges is None or ns_range is None , ( <EOL> "<STR_LIT>" ) <EOL> self . _entity_kind = entity_kind <EOL> self . _key_ranges = key_ranges and list ( reversed ( key_ranges ) ) <EOL> self . _ns_range = ns_range <EOL> self . _batch_size = int ( batch_size ) <EOL> self . _current_key_range = current_key_range <EOL> self . _filters = filters <EOL> @ classmethod <EOL> def _get_raw_entity_kind ( cls , entity_kind ) : <EOL> if "<STR_LIT:.>" in entity_kind : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> entity_kind , cls . __name__ ) <EOL> return entity_kind <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _key_ranges is not None : <EOL> for o in self . _iter_key_ranges ( ) : <EOL> yield o <EOL> elif self . _ns_range is not None : <EOL> for o in self . _iter_ns_range ( ) : <EOL> yield o <EOL> else : <EOL> assert False , "<STR_LIT>" <EOL> def _iter_key_ranges ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> if self . _current_key_range is None : <EOL> if self . _key_ranges : <EOL> self . _current_key_range = self . _key_ranges . pop ( ) <EOL> continue <EOL> else : <EOL> break <EOL> for key , o in self . _iter_key_range ( <EOL> copy . deepcopy ( self . _current_key_range ) ) : <EOL> self . _current_key_range . advance ( key ) <EOL> yield o <EOL> self . _current_key_range = None <EOL> def _iter_ns_range ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> if self . _current_key_range is None : <EOL> query = self . _ns_range . make_datastore_query ( ) <EOL> namespace_result = query . Get ( <NUM_LIT:1> ) <EOL> if not namespace_result : <EOL> break <EOL> namespace = namespace_result [ <NUM_LIT:0> ] . name ( ) or "<STR_LIT>" <EOL> self . _current_key_range = key_range . KeyRange ( <EOL> namespace = namespace , _app = self . _ns_range . app ) <EOL> yield ALLOW_CHECKPOINT <EOL> for key , o in self . _iter_key_range ( <EOL> copy . deepcopy ( self . _current_key_range ) ) : <EOL> self . _current_key_range . advance ( key ) <EOL> yield o <EOL> if ( self . _ns_range . is_single_namespace or <EOL> self . _current_key_range . namespace == self . _ns_range . namespace_end ) : <EOL> break <EOL> self . _ns_range = self . _ns_range . with_start_after ( <EOL> self . _current_key_range . namespace ) <EOL> self . _current_key_range = None <EOL> def _iter_key_range ( self , k_range ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _ns_range is None : <EOL> return repr ( self . _key_ranges ) <EOL> else : <EOL> return repr ( self . _ns_range ) <EOL> @ classmethod <EOL> def _choose_split_points ( cls , sorted_keys , shard_count ) : <EOL> """<STR_LIT>""" <EOL> assert len ( sorted_keys ) >= shard_count <EOL> index_stride = len ( sorted_keys ) / float ( shard_count ) <EOL> return [ sorted_keys [ int ( round ( index_stride * i ) ) ] <EOL> for i in range ( <NUM_LIT:1> , shard_count ) ] <EOL> @ classmethod <EOL> def _split_input_from_namespace ( cls , app , namespace , entity_kind , <EOL> shard_count ) : <EOL> """<STR_LIT>""" <EOL> raw_entity_kind = cls . _get_raw_entity_kind ( entity_kind ) <EOL> if shard_count == <NUM_LIT:1> : <EOL> return [ key_range . KeyRange ( namespace = namespace , _app = app ) ] <EOL> ds_query = datastore . Query ( kind = raw_entity_kind , <EOL> namespace = namespace , <EOL> _app = app , <EOL> keys_only = True ) <EOL> ds_query . Order ( "<STR_LIT>" ) <EOL> random_keys = ds_query . Get ( shard_count * cls . _OVERSAMPLING_FACTOR ) <EOL> if not random_keys : <EOL> return ( [ key_range . KeyRange ( namespace = namespace , _app = app ) ] + <EOL> [ None ] * ( shard_count - <NUM_LIT:1> ) ) <EOL> random_keys . sort ( ) <EOL> if len ( random_keys ) >= shard_count : <EOL> random_keys = cls . _choose_split_points ( random_keys , shard_count ) <EOL> key_ranges = [ ] <EOL> key_ranges . append ( key_range . KeyRange ( <EOL> key_start = None , <EOL> key_end = random_keys [ <NUM_LIT:0> ] , <EOL> direction = key_range . KeyRange . ASC , <EOL> include_start = False , <EOL> include_end = False , <EOL> namespace = namespace , <EOL> _app = app ) ) <EOL> for i in range ( <NUM_LIT:0> , len ( random_keys ) - <NUM_LIT:1> ) : <EOL> key_ranges . append ( key_range . KeyRange ( <EOL> key_start = random_keys [ i ] , <EOL> key_end = random_keys [ i + <NUM_LIT:1> ] , <EOL> direction = key_range . KeyRange . ASC , <EOL> include_start = True , <EOL> include_end = False , <EOL> namespace = namespace , <EOL> _app = app ) ) <EOL> key_ranges . append ( key_range . KeyRange ( <EOL> key_start = random_keys [ - <NUM_LIT:1> ] , <EOL> key_end = None , <EOL> direction = key_range . KeyRange . ASC , <EOL> include_start = True , <EOL> include_end = False , <EOL> namespace = namespace , <EOL> _app = app ) ) <EOL> if len ( key_ranges ) < shard_count : <EOL> key_ranges += [ None ] * ( shard_count - len ( key_ranges ) ) <EOL> return key_ranges <EOL> @ classmethod <EOL> def _split_input_from_params ( cls , app , namespaces , entity_kind_name , <EOL> params , shard_count ) : <EOL> """<STR_LIT>""" <EOL> key_ranges = [ ] <EOL> for namespace in namespaces : <EOL> key_ranges . extend ( <EOL> cls . _split_input_from_namespace ( app , <EOL> namespace , <EOL> entity_kind_name , <EOL> shard_count ) ) <EOL> shared_ranges = [ [ ] for _ in range ( shard_count ) ] <EOL> for i , k_range in enumerate ( key_ranges ) : <EOL> shared_ranges [ i % shard_count ] . append ( k_range ) <EOL> batch_size = int ( params . get ( cls . BATCH_SIZE_PARAM , cls . _BATCH_SIZE ) ) <EOL> return [ cls ( entity_kind_name , <EOL> key_ranges = key_ranges , <EOL> ns_range = None , <EOL> batch_size = batch_size ) <EOL> for key_ranges in shared_ranges if key_ranges ] <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . ENTITY_KIND_PARAM not in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if cls . BATCH_SIZE_PARAM in params : <EOL> try : <EOL> batch_size = int ( params [ cls . BATCH_SIZE_PARAM ] ) <EOL> if batch_size < <NUM_LIT:1> : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % batch_size ) <EOL> except ValueError , e : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % e ) <EOL> if cls . NAMESPACE_PARAM in params : <EOL> if not isinstance ( params [ cls . NAMESPACE_PARAM ] , <EOL> ( str , unicode , type ( None ) ) ) : <EOL> raise BadReaderParamsError ( <EOL> "<STR_LIT>" ) <EOL> if cls . NAMESPACES_PARAM in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if cls . FILTERS_PARAM in params : <EOL> filters = params [ cls . FILTERS_PARAM ] <EOL> if not isinstance ( filters , list ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> for f in filters : <EOL> if not isinstance ( f , ( tuple , list ) ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , f ) <EOL> if len ( f ) != <NUM_LIT:3> : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , f ) <EOL> if not isinstance ( f [ <NUM_LIT:0> ] , basestring ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" , f ) <EOL> if f [ <NUM_LIT:1> ] != "<STR_LIT:=>" : <EOL> raise BadReaderParamsError ( <EOL> "<STR_LIT>" , f ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> entity_kind_name = params [ cls . ENTITY_KIND_PARAM ] <EOL> batch_size = int ( params . get ( cls . BATCH_SIZE_PARAM , cls . _BATCH_SIZE ) ) <EOL> shard_count = mapper_spec . shard_count <EOL> namespace = params . get ( cls . NAMESPACE_PARAM ) <EOL> app = params . get ( cls . _APP_PARAM ) <EOL> filters = params . get ( cls . FILTERS_PARAM ) <EOL> if namespace is None : <EOL> namespace_query = datastore . Query ( "<STR_LIT>" , <EOL> keys_only = True , <EOL> _app = app ) <EOL> namespace_keys = namespace_query . Get ( <EOL> limit = cls . MAX_NAMESPACES_FOR_KEY_SHARD + <NUM_LIT:1> ) <EOL> if len ( namespace_keys ) > cls . MAX_NAMESPACES_FOR_KEY_SHARD : <EOL> ns_ranges = namespace_range . NamespaceRange . split ( n = shard_count , <EOL> contiguous = True , <EOL> _app = app ) <EOL> return [ cls ( entity_kind_name , <EOL> key_ranges = None , <EOL> ns_range = ns_range , <EOL> batch_size = batch_size , <EOL> filters = filters ) <EOL> for ns_range in ns_ranges ] <EOL> elif not namespace_keys : <EOL> return [ cls ( entity_kind_name , <EOL> key_ranges = None , <EOL> ns_range = namespace_range . NamespaceRange ( _app = app ) , <EOL> batch_size = shard_count , <EOL> filters = filters ) ] <EOL> else : <EOL> namespaces = [ namespace_key . name ( ) or "<STR_LIT>" <EOL> for namespace_key in namespace_keys ] <EOL> else : <EOL> namespaces = [ namespace ] <EOL> readers = cls . _split_input_from_params ( <EOL> app , namespaces , entity_kind_name , params , shard_count ) <EOL> if filters : <EOL> for reader in readers : <EOL> reader . _filters = filters <EOL> return readers <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _key_ranges is None : <EOL> key_ranges_json = None <EOL> else : <EOL> key_ranges_json = [ ] <EOL> for k in self . _key_ranges : <EOL> if k : <EOL> key_ranges_json . append ( k . to_json ( ) ) <EOL> else : <EOL> key_ranges_json . append ( None ) <EOL> if self . _ns_range is None : <EOL> namespace_range_json = None <EOL> else : <EOL> namespace_range_json = self . _ns_range . to_json_object ( ) <EOL> if self . _current_key_range is None : <EOL> current_key_range_json = None <EOL> else : <EOL> current_key_range_json = self . _current_key_range . to_json ( ) <EOL> json_dict = { self . KEY_RANGE_PARAM : key_ranges_json , <EOL> self . NAMESPACE_RANGE_PARAM : namespace_range_json , <EOL> self . CURRENT_KEY_RANGE_PARAM : current_key_range_json , <EOL> self . ENTITY_KIND_PARAM : self . _entity_kind , <EOL> self . BATCH_SIZE_PARAM : self . _batch_size , <EOL> self . FILTERS_PARAM : self . _filters } <EOL> return json_dict <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> if json [ cls . KEY_RANGE_PARAM ] is None : <EOL> key_ranges = None <EOL> else : <EOL> key_ranges = [ ] <EOL> for k in json [ cls . KEY_RANGE_PARAM ] : <EOL> if k : <EOL> key_ranges . append ( key_range . KeyRange . from_json ( k ) ) <EOL> else : <EOL> key_ranges . append ( None ) <EOL> if json [ cls . NAMESPACE_RANGE_PARAM ] is None : <EOL> ns_range = None <EOL> else : <EOL> ns_range = namespace_range . NamespaceRange . from_json_object ( <EOL> json [ cls . NAMESPACE_RANGE_PARAM ] ) <EOL> if json [ cls . CURRENT_KEY_RANGE_PARAM ] is None : <EOL> current_key_range = None <EOL> else : <EOL> current_key_range = key_range . KeyRange . from_json ( <EOL> json [ cls . CURRENT_KEY_RANGE_PARAM ] ) <EOL> return cls ( <EOL> json [ cls . ENTITY_KIND_PARAM ] , <EOL> key_ranges , <EOL> ns_range , <EOL> json [ cls . BATCH_SIZE_PARAM ] , <EOL> current_key_range , <EOL> filters = json . get ( cls . FILTERS_PARAM ) ) <EOL> class BlobstoreLineInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> _BLOB_BUFFER_SIZE = <NUM_LIT> <EOL> _MAX_SHARD_COUNT = <NUM_LIT> <EOL> _MAX_BLOB_KEYS_COUNT = <NUM_LIT> <EOL> BLOB_KEYS_PARAM = "<STR_LIT>" <EOL> INITIAL_POSITION_PARAM = "<STR_LIT>" <EOL> END_POSITION_PARAM = "<STR_LIT>" <EOL> BLOB_KEY_PARAM = "<STR_LIT>" <EOL> def __init__ ( self , blob_key , start_position , end_position ) : <EOL> """<STR_LIT>""" <EOL> self . _blob_key = blob_key <EOL> self . _blob_reader = blobstore . BlobReader ( blob_key , <EOL> self . _BLOB_BUFFER_SIZE , <EOL> start_position ) <EOL> self . _end_position = end_position <EOL> self . _has_iterated = False <EOL> self . _read_before_start = bool ( start_position ) <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _has_iterated = True <EOL> if self . _read_before_start : <EOL> self . _blob_reader . readline ( ) <EOL> self . _read_before_start = False <EOL> start_position = self . _blob_reader . tell ( ) <EOL> if start_position > self . _end_position : <EOL> raise StopIteration ( ) <EOL> line = self . _blob_reader . readline ( ) <EOL> if not line : <EOL> raise StopIteration ( ) <EOL> return start_position , line . rstrip ( "<STR_LIT:\n>" ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> new_pos = self . _blob_reader . tell ( ) <EOL> if self . _has_iterated : <EOL> new_pos -= <NUM_LIT:1> <EOL> return { self . BLOB_KEY_PARAM : self . _blob_key , <EOL> self . INITIAL_POSITION_PARAM : new_pos , <EOL> self . END_POSITION_PARAM : self . _end_position } <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> self . _blob_key , self . _blob_reader . tell ( ) , self . _end_position ) <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> return cls ( json [ cls . BLOB_KEY_PARAM ] , <EOL> json [ cls . INITIAL_POSITION_PARAM ] , <EOL> json [ cls . END_POSITION_PARAM ] ) <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . BLOB_KEYS_PARAM not in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> blob_keys = params [ cls . BLOB_KEYS_PARAM ] <EOL> if isinstance ( blob_keys , basestring ) : <EOL> blob_keys = blob_keys . split ( "<STR_LIT:U+002C>" ) <EOL> if len ( blob_keys ) > cls . _MAX_BLOB_KEYS_COUNT : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if not blob_keys : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> for blob_key in blob_keys : <EOL> blob_info = blobstore . BlobInfo . get ( blobstore . BlobKey ( blob_key ) ) <EOL> if not blob_info : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % <EOL> blob_key ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> blob_keys = params [ cls . BLOB_KEYS_PARAM ] <EOL> if isinstance ( blob_keys , basestring ) : <EOL> blob_keys = blob_keys . split ( "<STR_LIT:U+002C>" ) <EOL> blob_sizes = { } <EOL> for blob_key in blob_keys : <EOL> blob_info = blobstore . BlobInfo . get ( blobstore . BlobKey ( blob_key ) ) <EOL> blob_sizes [ blob_key ] = blob_info . size <EOL> shard_count = min ( cls . _MAX_SHARD_COUNT , mapper_spec . shard_count ) <EOL> shards_per_blob = shard_count // len ( blob_keys ) <EOL> if shards_per_blob == <NUM_LIT:0> : <EOL> shards_per_blob = <NUM_LIT:1> <EOL> chunks = [ ] <EOL> for blob_key , blob_size in blob_sizes . items ( ) : <EOL> blob_chunk_size = blob_size // shards_per_blob <EOL> for i in xrange ( shards_per_blob - <NUM_LIT:1> ) : <EOL> chunks . append ( BlobstoreLineInputReader . from_json ( <EOL> { cls . BLOB_KEY_PARAM : blob_key , <EOL> cls . INITIAL_POSITION_PARAM : blob_chunk_size * i , <EOL> cls . END_POSITION_PARAM : blob_chunk_size * ( i + <NUM_LIT:1> ) } ) ) <EOL> chunks . append ( BlobstoreLineInputReader . from_json ( <EOL> { cls . BLOB_KEY_PARAM : blob_key , <EOL> cls . INITIAL_POSITION_PARAM : blob_chunk_size * ( shards_per_blob - <NUM_LIT:1> ) , <EOL> cls . END_POSITION_PARAM : blob_size } ) ) <EOL> return chunks <EOL> class BlobstoreZipInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> _MAX_SHARD_COUNT = <NUM_LIT> <EOL> BLOB_KEY_PARAM = "<STR_LIT>" <EOL> START_INDEX_PARAM = "<STR_LIT>" <EOL> END_INDEX_PARAM = "<STR_LIT>" <EOL> def __init__ ( self , blob_key , start_index , end_index , <EOL> _reader = blobstore . BlobReader ) : <EOL> """<STR_LIT>""" <EOL> self . _blob_key = blob_key <EOL> self . _start_index = start_index <EOL> self . _end_index = end_index <EOL> self . _reader = _reader <EOL> self . _zip = None <EOL> self . _entries = None <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _zip : <EOL> self . _zip = zipfile . ZipFile ( self . _reader ( self . _blob_key ) ) <EOL> self . _entries = self . _zip . infolist ( ) [ self . _start_index : self . _end_index ] <EOL> self . _entries . reverse ( ) <EOL> if not self . _entries : <EOL> raise StopIteration ( ) <EOL> entry = self . _entries . pop ( ) <EOL> self . _start_index += <NUM_LIT:1> <EOL> return ( entry , lambda : self . _read ( entry ) ) <EOL> def _read ( self , entry ) : <EOL> """<STR_LIT>""" <EOL> start_time = time . time ( ) <EOL> content = self . _zip . read ( entry . filename ) <EOL> ctx = context . get ( ) <EOL> if ctx : <EOL> operation . counters . Increment ( COUNTER_IO_READ_BYTES , len ( content ) ) ( ctx ) <EOL> operation . counters . Increment ( <EOL> COUNTER_IO_READ_MSEC , int ( ( time . time ( ) - start_time ) * <NUM_LIT:1000> ) ) ( ctx ) <EOL> return content <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> return cls ( json [ cls . BLOB_KEY_PARAM ] , <EOL> json [ cls . START_INDEX_PARAM ] , <EOL> json [ cls . END_INDEX_PARAM ] ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return { self . BLOB_KEY_PARAM : self . _blob_key , <EOL> self . START_INDEX_PARAM : self . _start_index , <EOL> self . END_INDEX_PARAM : self . _end_index } <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> self . _blob_key , self . _start_index , self . _end_index ) <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . BLOB_KEY_PARAM not in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> blob_key = params [ cls . BLOB_KEY_PARAM ] <EOL> blob_info = blobstore . BlobInfo . get ( blobstore . BlobKey ( blob_key ) ) <EOL> if not blob_info : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % <EOL> blob_key ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec , _reader = blobstore . BlobReader ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> blob_key = params [ cls . BLOB_KEY_PARAM ] <EOL> zip_input = zipfile . ZipFile ( _reader ( blob_key ) ) <EOL> zfiles = zip_input . infolist ( ) <EOL> total_size = sum ( x . file_size for x in zfiles ) <EOL> num_shards = min ( mapper_spec . shard_count , cls . _MAX_SHARD_COUNT ) <EOL> size_per_shard = total_size // num_shards <EOL> shard_start_indexes = [ <NUM_LIT:0> ] <EOL> current_shard_size = <NUM_LIT:0> <EOL> for i , fileinfo in enumerate ( zfiles ) : <EOL> current_shard_size += fileinfo . file_size <EOL> if current_shard_size >= size_per_shard : <EOL> shard_start_indexes . append ( i + <NUM_LIT:1> ) <EOL> current_shard_size = <NUM_LIT:0> <EOL> if shard_start_indexes [ - <NUM_LIT:1> ] != len ( zfiles ) : <EOL> shard_start_indexes . append ( len ( zfiles ) ) <EOL> return [ cls ( blob_key , start_index , end_index , _reader ) <EOL> for start_index , end_index <EOL> in zip ( shard_start_indexes , shard_start_indexes [ <NUM_LIT:1> : ] ) ] <EOL> class BlobstoreZipLineInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> _MAX_SHARD_COUNT = <NUM_LIT> <EOL> _MAX_BLOB_KEYS_COUNT = <NUM_LIT> <EOL> BLOB_KEYS_PARAM = "<STR_LIT>" <EOL> BLOB_KEY_PARAM = "<STR_LIT>" <EOL> START_FILE_INDEX_PARAM = "<STR_LIT>" <EOL> END_FILE_INDEX_PARAM = "<STR_LIT>" <EOL> OFFSET_PARAM = "<STR_LIT>" <EOL> def __init__ ( self , blob_key , start_file_index , end_file_index , offset , <EOL> _reader = blobstore . BlobReader ) : <EOL> """<STR_LIT>""" <EOL> self . _blob_key = blob_key <EOL> self . _start_file_index = start_file_index <EOL> self . _end_file_index = end_file_index <EOL> self . _initial_offset = offset <EOL> self . _reader = _reader <EOL> self . _zip = None <EOL> self . _entries = None <EOL> self . _filestream = None <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . BLOB_KEYS_PARAM not in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> blob_keys = params [ cls . BLOB_KEYS_PARAM ] <EOL> if isinstance ( blob_keys , basestring ) : <EOL> blob_keys = blob_keys . split ( "<STR_LIT:U+002C>" ) <EOL> if len ( blob_keys ) > cls . _MAX_BLOB_KEYS_COUNT : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if not blob_keys : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> for blob_key in blob_keys : <EOL> blob_info = blobstore . BlobInfo . get ( blobstore . BlobKey ( blob_key ) ) <EOL> if not blob_info : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % <EOL> blob_key ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec , _reader = blobstore . BlobReader ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> blob_keys = params [ cls . BLOB_KEYS_PARAM ] <EOL> if isinstance ( blob_keys , basestring ) : <EOL> blob_keys = blob_keys . split ( "<STR_LIT:U+002C>" ) <EOL> blob_files = { } <EOL> total_size = <NUM_LIT:0> <EOL> for blob_key in blob_keys : <EOL> zip_input = zipfile . ZipFile ( _reader ( blob_key ) ) <EOL> blob_files [ blob_key ] = zip_input . infolist ( ) <EOL> total_size += sum ( x . file_size for x in blob_files [ blob_key ] ) <EOL> shard_count = min ( cls . _MAX_SHARD_COUNT , mapper_spec . shard_count ) <EOL> size_per_shard = total_size // shard_count <EOL> readers = [ ] <EOL> for blob_key in blob_keys : <EOL> bfiles = blob_files [ blob_key ] <EOL> current_shard_size = <NUM_LIT:0> <EOL> start_file_index = <NUM_LIT:0> <EOL> next_file_index = <NUM_LIT:0> <EOL> for fileinfo in bfiles : <EOL> next_file_index += <NUM_LIT:1> <EOL> current_shard_size += fileinfo . file_size <EOL> if current_shard_size >= size_per_shard : <EOL> readers . append ( cls ( blob_key , start_file_index , next_file_index , <NUM_LIT:0> , <EOL> _reader ) ) <EOL> current_shard_size = <NUM_LIT:0> <EOL> start_file_index = next_file_index <EOL> if current_shard_size != <NUM_LIT:0> : <EOL> readers . append ( cls ( blob_key , start_file_index , next_file_index , <NUM_LIT:0> , <EOL> _reader ) ) <EOL> return readers <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _filestream : <EOL> if not self . _zip : <EOL> self . _zip = zipfile . ZipFile ( self . _reader ( self . _blob_key ) ) <EOL> self . _entries = self . _zip . infolist ( ) [ self . _start_file_index : <EOL> self . _end_file_index ] <EOL> self . _entries . reverse ( ) <EOL> if not self . _entries : <EOL> raise StopIteration ( ) <EOL> entry = self . _entries . pop ( ) <EOL> value = self . _zip . read ( entry . filename ) <EOL> self . _filestream = StringIO . StringIO ( value ) <EOL> if self . _initial_offset : <EOL> self . _filestream . seek ( self . _initial_offset ) <EOL> self . _filestream . readline ( ) <EOL> start_position = self . _filestream . tell ( ) <EOL> line = self . _filestream . readline ( ) <EOL> if not line : <EOL> self . _filestream . close ( ) <EOL> self . _filestream = None <EOL> self . _start_file_index += <NUM_LIT:1> <EOL> self . _initial_offset = <NUM_LIT:0> <EOL> return self . next ( ) <EOL> return ( ( self . _blob_key , self . _start_file_index , start_position ) , <EOL> line . rstrip ( "<STR_LIT:\n>" ) ) <EOL> def _next_offset ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _filestream : <EOL> offset = self . _filestream . tell ( ) <EOL> if offset : <EOL> offset -= <NUM_LIT:1> <EOL> else : <EOL> offset = self . _initial_offset <EOL> return offset <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return { self . BLOB_KEY_PARAM : self . _blob_key , <EOL> self . START_FILE_INDEX_PARAM : self . _start_file_index , <EOL> self . END_FILE_INDEX_PARAM : self . _end_file_index , <EOL> self . OFFSET_PARAM : self . _next_offset ( ) } <EOL> @ classmethod <EOL> def from_json ( cls , json , _reader = blobstore . BlobReader ) : <EOL> """<STR_LIT>""" <EOL> return cls ( json [ cls . BLOB_KEY_PARAM ] , <EOL> json [ cls . START_FILE_INDEX_PARAM ] , <EOL> json [ cls . END_FILE_INDEX_PARAM ] , <EOL> json [ cls . OFFSET_PARAM ] , <EOL> _reader ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> self . _blob_key , self . _start_file_index , self . _end_file_index , <EOL> self . _next_offset ( ) ) <EOL> class RandomStringInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> COUNT = "<STR_LIT:count>" <EOL> STRING_LENGTH = "<STR_LIT>" <EOL> DEFAULT_STRING_LENGTH = <NUM_LIT:10> <EOL> def __init__ ( self , count , string_length ) : <EOL> """<STR_LIT>""" <EOL> self . _count = count <EOL> self . _string_length = string_length <EOL> def __iter__ ( self ) : <EOL> ctx = context . get ( ) <EOL> while self . _count : <EOL> self . _count -= <NUM_LIT:1> <EOL> start_time = time . time ( ) <EOL> content = "<STR_LIT>" . join ( random . choice ( string . ascii_lowercase ) <EOL> for _ in range ( self . _string_length ) ) <EOL> if ctx : <EOL> operation . counters . Increment ( <EOL> COUNTER_IO_READ_MSEC , int ( ( time . time ( ) - start_time ) * <NUM_LIT:1000> ) ) ( ctx ) <EOL> operation . counters . Increment ( COUNTER_IO_READ_BYTES , len ( content ) ) ( ctx ) <EOL> yield content <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> params = _get_params ( mapper_spec ) <EOL> count = params [ cls . COUNT ] <EOL> string_length = cls . DEFAULT_STRING_LENGTH <EOL> if cls . STRING_LENGTH in params : <EOL> string_length = params [ cls . STRING_LENGTH ] <EOL> shard_count = mapper_spec . shard_count <EOL> count_per_shard = count // shard_count <EOL> mr_input_readers = [ <EOL> cls ( count_per_shard , string_length ) for _ in range ( shard_count ) ] <EOL> left = count - count_per_shard * shard_count <EOL> if left > <NUM_LIT:0> : <EOL> mr_input_readers . append ( cls ( left , string_length ) ) <EOL> return mr_input_readers <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . COUNT not in params : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % cls . COUNT ) <EOL> if not isinstance ( params [ cls . COUNT ] , int ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % <EOL> ( cls . COUNT , type ( params [ cls . COUNT ] ) ) ) <EOL> if params [ cls . COUNT ] <= <NUM_LIT:0> : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> if cls . STRING_LENGTH in params and not ( <EOL> isinstance ( params [ cls . STRING_LENGTH ] , int ) and <EOL> params [ cls . STRING_LENGTH ] > <NUM_LIT:0> ) : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % <EOL> ( cls . STRING_LENGTH , params [ cls . STRING_LENGTH ] ) ) <EOL> if ( not isinstance ( mapper_spec . shard_count , int ) or <EOL> mapper_spec . shard_count <= <NUM_LIT:0> ) : <EOL> raise BadReaderParamsError ( <EOL> "<STR_LIT>" % <EOL> mapper_spec . shard_count ) <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> return cls ( json [ cls . COUNT ] , json [ cls . STRING_LENGTH ] ) <EOL> def to_json ( self ) : <EOL> return { self . COUNT : self . _count , self . STRING_LENGTH : self . _string_length } <EOL> class NamespaceInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> NAMESPACE_RANGE_PARAM = "<STR_LIT>" <EOL> BATCH_SIZE_PARAM = "<STR_LIT>" <EOL> _BATCH_SIZE = <NUM_LIT:10> <EOL> def __init__ ( self , ns_range , batch_size = _BATCH_SIZE ) : <EOL> self . ns_range = ns_range <EOL> self . _batch_size = batch_size <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return { self . NAMESPACE_RANGE_PARAM : self . ns_range . to_json_object ( ) , <EOL> self . BATCH_SIZE_PARAM : self . _batch_size } <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> return cls ( <EOL> namespace_range . NamespaceRange . from_json_object ( <EOL> json [ cls . NAMESPACE_RANGE_PARAM ] ) , <EOL> json [ cls . BATCH_SIZE_PARAM ] ) <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec ) <EOL> if cls . BATCH_SIZE_PARAM in params : <EOL> try : <EOL> batch_size = int ( params [ cls . BATCH_SIZE_PARAM ] ) <EOL> if batch_size < <NUM_LIT:1> : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % batch_size ) <EOL> except ValueError , e : <EOL> raise BadReaderParamsError ( "<STR_LIT>" % e ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> batch_size = int ( _get_params ( mapper_spec ) . get ( <EOL> cls . BATCH_SIZE_PARAM , cls . _BATCH_SIZE ) ) <EOL> shard_count = mapper_spec . shard_count <EOL> namespace_ranges = namespace_range . NamespaceRange . split ( shard_count , <EOL> contiguous = True ) <EOL> return [ NamespaceInputReader ( ns_range , batch_size ) <EOL> for ns_range in namespace_ranges ] <EOL> def __iter__ ( self ) : <EOL> while True : <EOL> keys = self . ns_range . make_datastore_query ( ) . Get ( limit = self . _batch_size ) <EOL> if not keys : <EOL> break <EOL> for key in keys : <EOL> namespace = metadata . Namespace . key_to_namespace ( key ) <EOL> self . ns_range = self . ns_range . with_start_after ( namespace ) <EOL> yield namespace <EOL> def __str__ ( self ) : <EOL> return repr ( self . ns_range ) <EOL> class LogInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> START_TIME_PARAM = "<STR_LIT>" <EOL> END_TIME_PARAM = "<STR_LIT>" <EOL> MINIMUM_LOG_LEVEL_PARAM = "<STR_LIT>" <EOL> INCLUDE_INCOMPLETE_PARAM = "<STR_LIT>" <EOL> INCLUDE_APP_LOGS_PARAM = "<STR_LIT>" <EOL> VERSION_IDS_PARAM = "<STR_LIT>" <EOL> MODULE_VERSIONS_PARAM = "<STR_LIT>" <EOL> _OFFSET_PARAM = "<STR_LIT>" <EOL> _PROTOTYPE_REQUEST_PARAM = "<STR_LIT>" <EOL> _PARAMS = frozenset ( [ START_TIME_PARAM , END_TIME_PARAM , _OFFSET_PARAM , <EOL> MINIMUM_LOG_LEVEL_PARAM , INCLUDE_INCOMPLETE_PARAM , <EOL> INCLUDE_APP_LOGS_PARAM , VERSION_IDS_PARAM , <EOL> MODULE_VERSIONS_PARAM , _PROTOTYPE_REQUEST_PARAM ] ) <EOL> _KWARGS = frozenset ( [ _OFFSET_PARAM , _PROTOTYPE_REQUEST_PARAM ] ) <EOL> def __init__ ( self , <EOL> start_time = None , <EOL> end_time = None , <EOL> minimum_log_level = None , <EOL> include_incomplete = False , <EOL> include_app_logs = False , <EOL> version_ids = None , <EOL> module_versions = None , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> InputReader . __init__ ( self ) <EOL> self . __params = dict ( kwargs ) <EOL> if start_time is not None : <EOL> self . __params [ self . START_TIME_PARAM ] = start_time <EOL> if end_time is not None : <EOL> self . __params [ self . END_TIME_PARAM ] = end_time <EOL> if minimum_log_level is not None : <EOL> self . __params [ self . MINIMUM_LOG_LEVEL_PARAM ] = minimum_log_level <EOL> if include_incomplete is not None : <EOL> self . __params [ self . INCLUDE_INCOMPLETE_PARAM ] = include_incomplete <EOL> if include_app_logs is not None : <EOL> self . __params [ self . INCLUDE_APP_LOGS_PARAM ] = include_app_logs <EOL> if version_ids : <EOL> self . __params [ self . VERSION_IDS_PARAM ] = version_ids <EOL> if module_versions : <EOL> self . __params [ self . MODULE_VERSIONS_PARAM ] = module_versions <EOL> if self . _PROTOTYPE_REQUEST_PARAM in self . __params : <EOL> prototype_request = log_service_pb . LogReadRequest ( <EOL> self . __params [ self . _PROTOTYPE_REQUEST_PARAM ] ) <EOL> self . __params [ self . _PROTOTYPE_REQUEST_PARAM ] = prototype_request <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> for log in logservice . fetch ( ** self . __params ) : <EOL> self . __params [ self . _OFFSET_PARAM ] = log . offset <EOL> yield log <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> params = dict ( ( str ( k ) , v ) for k , v in json . iteritems ( ) <EOL> if k in cls . _PARAMS ) <EOL> if cls . _OFFSET_PARAM in params : <EOL> params [ cls . _OFFSET_PARAM ] = base64 . b64decode ( params [ cls . _OFFSET_PARAM ] ) <EOL> return cls ( ** params ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> params = dict ( self . __params ) <EOL> if self . _PROTOTYPE_REQUEST_PARAM in params : <EOL> prototype_request = params [ self . _PROTOTYPE_REQUEST_PARAM ] <EOL> params [ self . _PROTOTYPE_REQUEST_PARAM ] = prototype_request . Encode ( ) <EOL> if self . _OFFSET_PARAM in params : <EOL> params [ self . _OFFSET_PARAM ] = base64 . b64encode ( params [ self . _OFFSET_PARAM ] ) <EOL> return params <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> params = _get_params ( mapper_spec ) <EOL> shard_count = mapper_spec . shard_count <EOL> start_time = params [ cls . START_TIME_PARAM ] <EOL> end_time = params [ cls . END_TIME_PARAM ] <EOL> seconds_per_shard = ( end_time - start_time ) / shard_count <EOL> shards = [ ] <EOL> for _ in xrange ( shard_count - <NUM_LIT:1> ) : <EOL> params [ cls . END_TIME_PARAM ] = ( params [ cls . START_TIME_PARAM ] + <EOL> seconds_per_shard ) <EOL> shards . append ( LogInputReader ( ** params ) ) <EOL> params [ cls . START_TIME_PARAM ] = params [ cls . END_TIME_PARAM ] <EOL> params [ cls . END_TIME_PARAM ] = end_time <EOL> return shards + [ LogInputReader ( ** params ) ] <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> if mapper_spec . input_reader_class ( ) != cls : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" ) <EOL> params = _get_params ( mapper_spec , allowed_keys = cls . _PARAMS ) <EOL> if ( cls . VERSION_IDS_PARAM not in params and <EOL> cls . MODULE_VERSIONS_PARAM not in params ) : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if ( cls . VERSION_IDS_PARAM in params and <EOL> cls . MODULE_VERSIONS_PARAM in params ) : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if ( cls . START_TIME_PARAM not in params or <EOL> params [ cls . START_TIME_PARAM ] is None ) : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if cls . END_TIME_PARAM not in params or params [ cls . END_TIME_PARAM ] is None : <EOL> params [ cls . END_TIME_PARAM ] = time . time ( ) <EOL> if params [ cls . START_TIME_PARAM ] >= params [ cls . END_TIME_PARAM ] : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if cls . _PROTOTYPE_REQUEST_PARAM in params : <EOL> try : <EOL> params [ cls . _PROTOTYPE_REQUEST_PARAM ] = log_service_pb . LogReadRequest ( <EOL> params [ cls . _PROTOTYPE_REQUEST_PARAM ] ) <EOL> except ( TypeError , ProtocolBuffer . ProtocolBufferDecodeError ) : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> try : <EOL> logservice . fetch ( ** params ) <EOL> except logservice . InvalidArgumentError , e : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" <EOL> "<STR_LIT>" % e ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> params = [ ] <EOL> for key in sorted ( self . __params . keys ( ) ) : <EOL> value = self . __params [ key ] <EOL> if key is self . _PROTOTYPE_REQUEST_PARAM : <EOL> params . append ( "<STR_LIT>" % ( key , value ) ) <EOL> elif key is self . _OFFSET_PARAM : <EOL> params . append ( "<STR_LIT>" % ( key , value ) ) <EOL> else : <EOL> params . append ( "<STR_LIT>" % ( key , value ) ) <EOL> return "<STR_LIT>" % "<STR_LIT:U+002CU+0020>" . join ( params ) <EOL> class _GoogleCloudStorageInputReader ( InputReader ) : <EOL> """<STR_LIT>""" <EOL> BUCKET_NAME_PARAM = "<STR_LIT>" <EOL> OBJECT_NAMES_PARAM = "<STR_LIT>" <EOL> BUFFER_SIZE_PARAM = "<STR_LIT>" <EOL> DELIMITER_PARAM = "<STR_LIT>" <EOL> FAIL_ON_MISSING_INPUT = "<STR_LIT>" <EOL> _ACCOUNT_ID_PARAM = "<STR_LIT>" <EOL> _JSON_PICKLE = "<STR_LIT>" <EOL> _JSON_FAIL_ON_MISSING_INPUT = "<STR_LIT>" <EOL> _STRING_MAX_FILES_LISTED = <NUM_LIT:10> <EOL> def __init__ ( self , filenames , index = <NUM_LIT:0> , buffer_size = None , _account_id = None , <EOL> delimiter = None ) : <EOL> """<STR_LIT>""" <EOL> self . _filenames = filenames <EOL> self . _index = index <EOL> self . _buffer_size = buffer_size <EOL> self . _account_id = _account_id <EOL> self . _delimiter = delimiter <EOL> self . _bucket = None <EOL> self . _bucket_iter = None <EOL> self . _fail_on_missing_input = None <EOL> def _next_file ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> if self . _bucket_iter : <EOL> try : <EOL> return self . _bucket_iter . next ( ) . filename <EOL> except StopIteration : <EOL> self . _bucket_iter = None <EOL> self . _bucket = None <EOL> if self . _index >= len ( self . _filenames ) : <EOL> return <EOL> filename = self . _filenames [ self . _index ] <EOL> self . _index += <NUM_LIT:1> <EOL> if self . _delimiter is None or not filename . endswith ( self . _delimiter ) : <EOL> return filename <EOL> self . _bucket = cloudstorage . listbucket ( filename , <EOL> delimiter = self . _delimiter ) <EOL> self . _bucket_iter = iter ( self . _bucket ) <EOL> @ classmethod <EOL> def get_params ( cls , mapper_spec , allowed_keys = None , allow_old = True ) : <EOL> params = _get_params ( mapper_spec , allowed_keys , allow_old ) <EOL> if ( mapper_spec . params . get ( cls . BUCKET_NAME_PARAM ) is not None and <EOL> params . get ( cls . BUCKET_NAME_PARAM ) is None ) : <EOL> params [ cls . BUCKET_NAME_PARAM ] = mapper_spec . params [ cls . BUCKET_NAME_PARAM ] <EOL> return params <EOL> @ classmethod <EOL> def validate ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> reader_spec = cls . get_params ( mapper_spec , allow_old = False ) <EOL> if cls . BUCKET_NAME_PARAM not in reader_spec : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" % <EOL> cls . BUCKET_NAME_PARAM ) <EOL> try : <EOL> cloudstorage . validate_bucket_name ( <EOL> reader_spec [ cls . BUCKET_NAME_PARAM ] ) <EOL> except ValueError , error : <EOL> raise errors . BadReaderParamsError ( "<STR_LIT>" % ( error ) ) <EOL> if cls . OBJECT_NAMES_PARAM not in reader_spec : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" % <EOL> cls . OBJECT_NAMES_PARAM ) <EOL> filenames = reader_spec [ cls . OBJECT_NAMES_PARAM ] <EOL> if not isinstance ( filenames , list ) : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" % <EOL> filenames . __class__ . __name__ ) <EOL> for filename in filenames : <EOL> if not isinstance ( filename , basestring ) : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" % <EOL> filename . __class__ . __name__ ) <EOL> if cls . DELIMITER_PARAM in reader_spec : <EOL> delimiter = reader_spec [ cls . DELIMITER_PARAM ] <EOL> if not isinstance ( delimiter , basestring ) : <EOL> raise errors . BadReaderParamsError ( <EOL> "<STR_LIT>" % <EOL> ( cls . DELIMITER_PARAM , type ( delimiter ) ) ) <EOL> @ classmethod <EOL> def split_input ( cls , mapper_spec ) : <EOL> """<STR_LIT>""" <EOL> reader_spec = cls . get_params ( mapper_spec , allow_old = False ) <EOL> bucket = reader_spec [ cls . BUCKET_NAME_PARAM ] <EOL> filenames = reader_spec [ cls . OBJECT_NAMES_PARAM ] <EOL> delimiter = reader_spec . get ( cls . DELIMITER_PARAM ) <EOL> account_id = reader_spec . get ( cls . _ACCOUNT_ID_PARAM ) <EOL> buffer_size = reader_spec . get ( cls . BUFFER_SIZE_PARAM ) <EOL> fail_on_missing_input = reader_spec . get ( cls . FAIL_ON_MISSING_INPUT ) <EOL> all_filenames = [ ] <EOL> for filename in filenames : <EOL> if filename . endswith ( "<STR_LIT:*>" ) : <EOL> all_filenames . extend ( <EOL> [ file_stat . filename for file_stat in cloudstorage . listbucket ( <EOL> "<STR_LIT:/>" + bucket + "<STR_LIT:/>" + filename [ : - <NUM_LIT:1> ] , delimiter = delimiter , <EOL> _account_id = account_id ) ] ) <EOL> else : <EOL> all_filenames . append ( "<STR_LIT>" % ( bucket , filename ) ) <EOL> readers = [ ] <EOL> for shard in range ( <NUM_LIT:0> , mapper_spec . shard_count ) : <EOL> shard_filenames = all_filenames [ shard : : mapper_spec . shard_count ] <EOL> if shard_filenames : <EOL> reader = cls ( <EOL> shard_filenames , buffer_size = buffer_size , _account_id = account_id , <EOL> delimiter = delimiter ) <EOL> reader . _fail_on_missing_input = fail_on_missing_input <EOL> readers . append ( reader ) <EOL> return readers <EOL> @ classmethod <EOL> def from_json ( cls , state ) : <EOL> obj = pickle . loads ( state [ cls . _JSON_PICKLE ] ) <EOL> obj . _fail_on_missing_input = state . get ( <EOL> cls . _JSON_FAIL_ON_MISSING_INPUT , False ) <EOL> if obj . _bucket : <EOL> obj . _bucket_iter = iter ( obj . _bucket ) <EOL> return obj <EOL> def to_json ( self ) : <EOL> before_iter = self . _bucket_iter <EOL> self . _bucket_iter = None <EOL> try : <EOL> return { <EOL> self . _JSON_PICKLE : pickle . dumps ( self ) , <EOL> self . _JSON_FAIL_ON_MISSING_INPUT : <EOL> getattr ( self , "<STR_LIT>" , False ) <EOL> } <EOL> finally : <EOL> self . _bucket_itr = before_iter <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> options = { } <EOL> if self . _buffer_size : <EOL> options [ "<STR_LIT>" ] = self . _buffer_size <EOL> if self . _account_id : <EOL> options [ "<STR_LIT>" ] = self . _account_id <EOL> while True : <EOL> filename = self . _next_file ( ) <EOL> if filename is None : <EOL> raise StopIteration ( ) <EOL> try : <EOL> start_time = time . time ( ) <EOL> handle = cloudstorage . open ( filename , ** options ) <EOL> ctx = context . get ( ) <EOL> if ctx : <EOL> operation . counters . Increment ( <EOL> COUNTER_IO_READ_MSEC , int ( ( time . time ( ) - start_time ) * <NUM_LIT:1000> ) ) ( ctx ) <EOL> return handle <EOL> except cloudstorage . NotFoundError : <EOL> self . _on_missing_input_file ( filename ) <EOL> if getattr ( self , "<STR_LIT>" , False ) : <EOL> raise errors . FailJobError ( <EOL> "<STR_LIT>" % filename ) <EOL> logging . warning ( "<STR_LIT>" , <EOL> filename ) <EOL> def _on_missing_input_file ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def __str__ ( self ) : <EOL> num_files = len ( self . _filenames ) <EOL> if num_files > self . _STRING_MAX_FILES_LISTED : <EOL> names = "<STR_LIT>" % ( <EOL> "<STR_LIT:U+002C>" . join ( self . _filenames [ <NUM_LIT:0> : self . _STRING_MAX_FILES_LISTED - <NUM_LIT:1> ] ) , <EOL> self . _filenames [ - <NUM_LIT:1> ] , <EOL> num_files - self . _STRING_MAX_FILES_LISTED ) <EOL> else : <EOL> names = "<STR_LIT:U+002C>" . join ( self . _filenames ) <EOL> if self . _index > num_files : <EOL> status = "<STR_LIT>" <EOL> else : <EOL> status = "<STR_LIT>" % ( <EOL> self . _filenames [ self . _index ] , <EOL> self . _index + <NUM_LIT:1> , <EOL> num_files ) <EOL> return "<STR_LIT>" % ( status , names ) <EOL> GoogleCloudStorageInputReader = _GoogleCloudStorageInputReader <EOL> class _GoogleCloudStorageRecordInputReader ( _GoogleCloudStorageInputReader ) : <EOL> """<STR_LIT>""" <EOL> def __getstate__ ( self ) : <EOL> result = self . __dict__ . copy ( ) <EOL> if "<STR_LIT>" in result : <EOL> result . pop ( "<STR_LIT>" ) <EOL> return result <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> if not hasattr ( self , "<STR_LIT>" ) or self . _cur_handle is None : <EOL> self . _cur_handle = super ( _GoogleCloudStorageRecordInputReader , <EOL> self ) . next ( ) <EOL> if not hasattr ( self , "<STR_LIT>" ) or self . _record_reader is None : <EOL> self . _record_reader = records . RecordsReader ( self . _cur_handle ) <EOL> try : <EOL> start_time = time . time ( ) <EOL> content = self . _record_reader . read ( ) <EOL> ctx = context . get ( ) <EOL> if ctx : <EOL> operation . counters . Increment ( COUNTER_IO_READ_BYTES , len ( content ) ) ( ctx ) <EOL> operation . counters . Increment ( <EOL> COUNTER_IO_READ_MSEC , int ( ( time . time ( ) - start_time ) * <NUM_LIT:1000> ) ) ( ctx ) <EOL> return content <EOL> except EOFError : <EOL> self . _cur_handle = None <EOL> self . _record_reader = None <EOL> GoogleCloudStorageRecordInputReader = _GoogleCloudStorageRecordInputReader <EOL> class _ReducerReader ( _GoogleCloudStorageRecordInputReader ) : <EOL> """<STR_LIT>""" <EOL> expand_parameters = True <EOL> def __init__ ( self , filenames , index = <NUM_LIT:0> , buffer_size = None , _account_id = None , <EOL> delimiter = None ) : <EOL> super ( _ReducerReader , self ) . __init__ ( filenames , index , buffer_size , <EOL> _account_id , delimiter ) <EOL> self . current_key = None <EOL> self . current_values = None <EOL> def __iter__ ( self ) : <EOL> ctx = context . get ( ) <EOL> combiner = None <EOL> if ctx : <EOL> combiner_spec = ctx . mapreduce_spec . mapper . params . get ( "<STR_LIT>" ) <EOL> if combiner_spec : <EOL> combiner = util . handler_for_name ( combiner_spec ) <EOL> try : <EOL> while True : <EOL> binary_record = super ( _ReducerReader , self ) . next ( ) <EOL> proto = kv_pb . KeyValues ( ) <EOL> proto . ParseFromString ( binary_record ) <EOL> to_yield = None <EOL> if self . current_key is not None and self . current_key != proto . key ( ) : <EOL> to_yield = ( self . current_key , self . current_values ) <EOL> self . current_key = None <EOL> self . current_values = None <EOL> if self . current_key is None : <EOL> self . current_key = proto . key ( ) <EOL> self . current_values = [ ] <EOL> if combiner : <EOL> combiner_result = combiner ( <EOL> self . current_key , proto . value_list ( ) , self . current_values ) <EOL> if not util . is_generator ( combiner_result ) : <EOL> raise errors . BadCombinerOutputError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( combiner , combiner_result ) ) <EOL> self . current_values = [ ] <EOL> for value in combiner_result : <EOL> if isinstance ( value , operation . Operation ) : <EOL> value ( ctx ) <EOL> else : <EOL> self . current_values . append ( value ) <EOL> if not to_yield : <EOL> yield ALLOW_CHECKPOINT <EOL> else : <EOL> self . current_values . extend ( proto . value_list ( ) ) <EOL> if to_yield : <EOL> yield to_yield <EOL> yield ALLOW_CHECKPOINT <EOL> except StopIteration : <EOL> pass <EOL> if self . current_key is not None : <EOL> to_yield = ( self . current_key , self . current_values ) <EOL> self . current_key = None <EOL> self . current_values = None <EOL> yield to_yield <EOL> @ staticmethod <EOL> def encode_data ( data ) : <EOL> """<STR_LIT>""" <EOL> return base64 . b64encode ( pickle . dumps ( data ) ) <EOL> @ staticmethod <EOL> def decode_data ( data ) : <EOL> """<STR_LIT>""" <EOL> return pickle . loads ( base64 . b64decode ( data ) ) <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> result = super ( _ReducerReader , self ) . to_json ( ) <EOL> result [ "<STR_LIT>" ] = self . encode_data ( self . current_key ) <EOL> result [ "<STR_LIT>" ] = self . encode_data ( self . current_values ) <EOL> return result <EOL> @ classmethod <EOL> def from_json ( cls , json ) : <EOL> """<STR_LIT>""" <EOL> result = super ( _ReducerReader , cls ) . from_json ( json ) <EOL> result . current_key = _ReducerReader . decode_data ( json [ "<STR_LIT>" ] ) <EOL> result . current_values = _ReducerReader . decode_data ( json [ "<STR_LIT>" ] ) <EOL> return result </s>
