I've written a patch with an alternative interface for the mapper and reducer functions. The Dumbo framework automatically detects which interface is being used by the function, and uses the function appropriately.

The reason I'm doing it is so that I can integrate my existing Python ETL framework into Dumbo. The ETL framework is designed around a Filters/Pipes Pipeline model. The framework is designed so that the input filter of the pipeline must get an iterator for the input filter to iterate on; the framework is not designed to provide a function for something else to call. For instance, the framework has an etl.file.DelimitedInput input filter, which takes a file object as a parameter. The filter then iterates over the file. Another example is etl.db.Query input filter: it takes a db connection and sql text as parameters, and the filter iterates over the db cursor (obviously, this input filter would not be used in Hadoop, because I don't want any db connectivity on the Hadoop mapreduce nodes).

So, the alternative interface for mappers and reducers is that it is passed an iterator as a parameter. Then, functions would be written like:

def mapper(data):
    # Initialization goes here
    for key, value in data:
        for word in value.split(): 
            yield word,1
    # Cleanup goes here

def reducer(data):
    # Initialization goes here
    for key, values in data:
        yield key, sum(values)
    # Cleanup goes here

There are also some other advantages to this interface:

• No need for mapclose/redclose options: you can define initialization and cleanup inside your mapper/reducer function.
• Slight performance benefit: not calling a function for every row of input, and one less for-loop.

Here's an an example of a portion of an ETL framework script:

pl = Pipeline(
 DelimitedInput(sys.stdin, Schema(sf.replace(encoding_errors='replace') for sf in hub1_combined_page.schema),
  keep=keep, drop=drop, rename=rename)
 |
 etl.transform.TwosComplement('network_ip')
 |
 etl.transform.IPString('ip_address=client_ip_addr')
 |
 etl.transform.StringCleaner()
 |
 GenerateNetworkRev()
 |
 etl.db.DatabaseLoaderFile(sys.stdout, 'netezza', load_page_events.schema,
  schema_filter=lambda s: (sf.replace(encoding_errors='replace') for sf in s),
  insert_filters=extra_filters)
).do()

This is doing:

• DelimitedInput: Take a tab-delimited file, with the 113 fields described in hub1_combined_page.schema. If the input data does not match the defined types in the schema, convert to the schema's types, using a conversion cache on each column. Use the codecs module's "replace" error handler for unicode decoding errors. Before converting types of fields, keep a list of fields, drop certain fields, and rename certain fields.
• TwosComplement: For the network_ip field, convert the unsigned 32-bit int field to signed 32-bit, so it'd fit in a 32-bit field in the database, since the database only supports signed values.
• IPString: Convert an unsigned 32-bit int to dotted-quad string notation.
• StringCleaner: For all fields in the schema which is of type unicode string, do unicode string cleaning, which includes things like transforming to Unicode Normalized Form C and dealing with unassigned codepoints, control characters, surrogate pairs, and Byte Order Marks.
• GenerateNetworkRev: Add the network_rev field to the data.
• DatabaseLoaderFile: Write out the data in netezza database bulk loader file format, using the table schema defined in load_page_events.schema.

I see integrating it into Dumbo something like this:

def mapper(data):
    pl = Pipeline(
     DumboInput(data, Schema(sf.replace(encoding_errors='replace') for sf in hub1_combined_page.schema),
      keep=keep, drop=drop, rename=rename)
     |
     etl.transform.TwosComplement('network_ip')
     |
     etl.transform.IPString('ip_address=client_ip_addr')
     |
     etl.transform.StringCleaner()
     |
     GenerateNetworkRev()
     |
     etl.db.DatabaseLoaderGenerator('netezza', load_page_events.schema,
      schema_filter=lambda s: (sf.replace(encoding_errors='replace') for sf in s))
     |
     DumboKeyGenerator('site_id')
    )
    pl.setup()
    return iter(pl)

DumboKeyGenerator will extract site_id from the input row, assign that as the key, and yield key, row.

I'll submit the patch in a few minutes.