def _sort_records_map(records):
  """Map function sorting records.

  Converts records to KeyValue protos, sorts them by key and writes them
  into new GCS file. Creates _OutputFile entity to record resulting
  file name.

  Args:
    records: list of records which are serialized KeyValue protos.
  """
  ctx = context.get()
  l = len(records)
  key_records = [None] * l

  logging.debug("Parsing")
  for i in range(l):
    proto = kv_pb.KeyValue()
    proto.ParseFromString(records[i])
    key_records[i] = (proto.key(), records[i])

  logging.debug("Sorting")
  key_records.sort(cmp=_compare_keys)

  logging.debug("Writing")
  mapper_spec = ctx.mapreduce_spec.mapper
  params = input_readers._get_params(mapper_spec)
  bucket_name = params.get("bucket_name")
  filename = (ctx.mapreduce_spec.name + "/" + ctx.mapreduce_id + "/output-" +
              ctx.shard_id + "-" + str(int(time.time())))
  full_filename = "/%s/%s" % (bucket_name, filename)
  filehandle = cloudstorage.open(full_filename, mode="w")
  with output_writers.GCSRecordsPool(filehandle, ctx=ctx) as pool:
    for key_record in key_records:
      pool.append(key_record[1])

  logging.debug("Finalizing")
  filehandle.close()

  entity = _OutputFile(key_name=full_filename,
                       parent=_OutputFile.get_root_key(ctx.mapreduce_id))
  entity.put()
def _sort_records_map ( records ) : ctx = context . get ( ) l = len ( records ) key_records = [ None ] * l logging . debug ( "Parsing" ) for i in range ( l ) : proto = kv_pb . KeyValue ( ) proto . ParseFromString ( records [ i ] ) key_records [ i ] = ( proto . key ( ) , records [ i ] ) logging . debug ( "Sorting" ) key_records . sort ( cmp = _compare_keys ) logging . debug ( "Writing" ) mapper_spec = ctx . mapreduce_spec . mapper params = input_readers . _get_params ( mapper_spec ) bucket_name = params . get ( "bucket_name" ) filename = ( ctx . mapreduce_spec . name + "/" + ctx . mapreduce_id + "/output-" + ctx . shard_id + "-" + str ( int ( time . time ( ) ) ) ) full_filename = "/%s/%s" % ( bucket_name , filename ) filehandle = cloudstorage . open ( full_filename , mode = "w" ) with output_writers . GCSRecordsPool ( filehandle , ctx = ctx ) as pool : for key_record in key_records : pool . append ( key_record [ 1 ] ) logging . debug ( "Finalizing" ) filehandle . close ( ) entity = _OutputFile ( key_name = full_filename , parent = _OutputFile . get_root_key ( ctx . mapreduce_id ) ) entity . put ( )
Map function sorting records.

  Converts records to KeyValue protos, sorts them by key and writes them
  into new GCS file. Creates _OutputFile entity to record resulting
  file name.

  Args:
    records: list of records which are serialized KeyValue protos.