mapreduce/handlers.py
def _schedule_shards(cls, spec, input_readers, queue_name, base_path):
    """Prepares shard states and schedules their execution.

    Args:
      spec: mapreduce specification as MapreduceSpec.
      input_readers: list of InputReaders describing shard splits.
      queue_name: The queue to run this job on.
      base_path: The base url path of mapreduce callbacks.
    """
    # Note: it's safe to re-attempt this handler because:
    # - shard state has deterministic and unique key.
    # - schedule_slice will fall back gracefully if a task already exists.
    shard_states = []
    for shard_number, input_reader in enumerate(input_readers):
      shard = model.ShardState.create_new(spec.mapreduce_id, shard_number)
      shard.shard_description = str(input_reader)
      shard_states.append(shard)

    # Retrievs already existing shards.
    existing_shard_states = db.get(shard.key() for shard in shard_states)
    existing_shard_keys = set(shard.key() for shard in existing_shard_states
                              if shard is not None)

    # Puts only non-existing shards.
    db.put(shard for shard in shard_states
           if shard.key() not in existing_shard_keys)

    for shard_number, input_reader in enumerate(input_readers):
      shard_id = model.ShardState.shard_id_from_number(
          spec.mapreduce_id, shard_number)
      MapperWorkerCallbackHandler.schedule_slice(
          base_path, spec, shard_id, 0, input_reader, queue_name=queue_name)
Prepares shard states and schedules their execution.

    Args:
      spec: mapreduce specification as MapreduceSpec.
      input_readers: list of InputReaders describing shard splits.
      queue_name: The queue to run this job on.
      base_path: The base url path of mapreduce callbacks.
    
Prepares shard states and schedules their execution.
Prepares shard states and schedules their execution .
def _schedule_shards(cls, spec, input_readers, queue_name, base_path):
    shard_states = []
    for shard_number, input_reader in enumerate(input_readers):
      shard = model.ShardState.create_new(spec.mapreduce_id, shard_number)
      shard.shard_description = str(input_reader)
      shard_states.append(shard)
    existing_shard_states = db.get(shard.key() for shard in shard_states)
    existing_shard_keys = set(shard.key() for shard in existing_shard_states
                              if shard is not None)
    db.put(shard for shard in shard_states
           if shard.key() not in existing_shard_keys)
    for shard_number, input_reader in enumerate(input_readers):
      shard_id = model.ShardState.shard_id_from_number(
          spec.mapreduce_id, shard_number)
      MapperWorkerCallbackHandler.schedule_slice(
          base_path, spec, shard_id, 0, input_reader, queue_name=queue_name)
def _schedule_shards ( cls , spec , input_readers , queue_name , base_path ) : shard_states = [ ] for shard_number , input_reader in enumerate ( input_readers ) : shard = model . ShardState . create_new ( spec . mapreduce_id , shard_number ) shard . shard_description = str ( input_reader ) shard_states . append ( shard ) existing_shard_states = db . get ( shard . key ( ) for shard in shard_states ) existing_shard_keys = set ( shard . key ( ) for shard in existing_shard_states if shard is not None ) db . put ( shard for shard in shard_states if shard . key ( ) not in existing_shard_keys ) for shard_number , input_reader in enumerate ( input_readers ) : shard_id = model . ShardState . shard_id_from_number ( spec . mapreduce_id , shard_number ) MapperWorkerCallbackHandler . schedule_slice ( base_path , spec , shard_id , 0 , input_reader , queue_name = queue_name )
Prepares shard states and schedules their execution.
Prepares shard states and schedules their execution .
"""Prepares shard states and schedules their execution.

    Args:
      spec: mapreduce specification as MapreduceSpec.
      input_readers: list of InputReaders describing shard splits.
      queue_name: The queue to run this job on.
      base_path: The base url path of mapreduce callbacks.
    """ # Note: it's safe to re-attempt this handler because: # - shard state has deterministic and unique key. # - schedule_slice will fall back gracefully if a task already exists. # Retrievs already existing shards. # Puts only non-existing shards.
cls
spec
mapreduce specification as MapreduceSpec.
mapreduce specification as MapreduceSpec .
input_readers
list of InputReaders describing shard splits.
list of InputReaders describing shard splits .
queue_name
The queue to run this job on.
The queue to run this job on .
base_path
The base url path of mapreduce callbacks.
The base url path of mapreduce callbacks .