A sample code demonstrating how to output advanced datatypes from a map-reduce job using mrjob.py

CustomerTotalExpenses.py

from mrjob.job import MRJob
import mrjob.protocol

from decimal import Decimal

class MRCustomerTotalExpenses(MRJob):
	INTERNAL_PROTOCOL = mrjob.protocol.PickleProtocol
	OUTPUT_PROTOCOL = mrjob.protocol.ReprProtocol

	def mapper(self, key, line):
		customerID, itemID, amount = line.split(',')
		yield customerID, Decimal(amount)

	def reducer(self, customerID, amounts):
		yield customerID, sum(amounts)

if __name__ == "__main__":
	MRCustomerTotalExpenses.run()