diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..9c9f9fb
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,2 @@
+[run]
+omit: tinymr/_backport_heapq.py
diff --git a/README.rst b/README.rst
index 853c011..d653294 100644
--- a/README.rst
+++ b/README.rst
@@ -105,7 +105,7 @@ implementation with parallelized map and reduce phases will be added.
 
             yield key, sum(values)
 
-        def final_reducer(self, pairs):
+        def output(self, pairs):
 
             """
             Normally this phase is where the final dataset is written to disk,
@@ -186,52 +186,24 @@ that appear below match the ``word`` only because a ``sort`` key was not given.
 Words that appear in the input text on multiple lines have multiple
 ``(word, count)`` pairs.  A ``count`` of ``2`` would indicate a word that
 appeared twice on a single line, but our input data does not have this
-condition.
+condition.  Truncated output below.  The dictionary values are lists containing
+tuples to allow for a sort key, which is explained elsewhere.
 
 .. code-block:: python
 
     {
-        'use': [('use', 1)],
-        'new': [('new', 1)],
-        'above': [('above', 1)],
-        'redistributions': [('redistributions', 1)],
-        'source': [('source', 1), ('source', 1)],
-        'without': [('without', 1)],
-        'notice': [('notice', 1)],
-        'redistribution': [('redistribution', 1)],
-        'bsd': [('bsd', 1)],
-        'that': [('that', 1)],
-        'permitted': [('permitted', 1)],
-        'forms': [('forms', 1)],
-        'rights': [('rights', 1)],
-        'must': [('must', 1)],
-        'list': [('list', 1)],
-        'are': [('are', 1), ('are', 1)],
-        'with': [('with', 1)],
-        'd': [('d', 1)],
-        'license': [('license', 1)],
-        'binary': [('binary', 1)],
-        'reserved': [('reserved', 1)],
-        'or': [('or', 1)],
-        'the': [('the', 1), ('the', 1), ('the', 1)],
-        'and': [('and', 1), ('and', 1), ('and', 1)],
-        'all': [('all', 1)],
-        'met': [('met', 1)],
-        'this': [('this', 1)],
-        'provided': [('provided', 1)],
-        'of': [('of', 1), ('of', 1)],
-        'c': [('c', 1)],
-        'wurster': [('wurster', 1)],
-        'code': [('code', 1)],
-        'disclaimer': [('disclaimer', 1)],
-        'modification': [('modification', 1)],
-        'copyright': [('copyright', 1), ('copyright', 1)],
-        'retain': [('retain', 1)], 'kevin': [('kevin', 1)],
-        'conditions': [('conditions', 1), ('conditions', 1)],
-        'following': [('following', 1), ('following', 1)],
-        'in': [('in', 1)], '2015': [('2015', 1)]
+        '2015': [(1,)]
+        'above': [(1,)]
+        'all': [(1,)]
+        'and': [(1,), (1,), (1,)]
+        'are': [(1,), (1,)]
+        'binary': [(1,)]
+        'bsd': [(1,)]
+        'c': [(1,)]
+        'code': [(1,)]
     }
 
+
 **Reduce**
 
 Sum ``count`` for each ``word``.
@@ -241,62 +213,31 @@ Sum ``count`` for each ``word``.
     # The ``reducer()`` receives a key and an iterator of values
     key = 'the'
     values = (1, 1, 1)
-    yield key, sum(values)
+    def reducer(key, values):
+        yield key, sum(values)
 
 **Partition**
 
 The reducer does not _have_ to produces the same key it was given, so the data
 is partitioned by key again, which is superfluous for this wordcount example.
 Again the keys are kept in case the data is sorted and only match ``word``
-because an optional ``sort`` key was not given.
+because an optional ``sort`` key was not given.  Truncated output below.
 
 .. code-block:: python
 
     {
-        'following': [('following', 2)],
-        '2015': [('2015', 1)],
-        'reserved': [('reserved', 1)],
-        'permitted': [('permitted', 1)],
-        'forms': [('forms', 1)],
-        'are': [('are', 2)],
-        'license': [('license', 1)],
-        'c': [('c', 1)],
-        'kevin': [('kevin', 1)],
-        'without': [('without', 1)],
-        'redistribution': [('redistribution', 1)],
-        'copyright': [('copyright', 2)],
-        'met': [('met', 1)],
-        'use': [('use', 1)],
-        'the': [('the', 3)],
-        'rights': [('rights', 1)],
-        'that': [('that', 1)],
-        'or': [('or', 1)],
-        'this': [('this', 1)],
-        'with': [('with', 1)],
-        'source': [('source', 2)],
-        'new': [('new', 1)],
-        'binary': [('binary', 1)],
-        'wurster': [('wurster', 1)],
-        'list': [('list', 1)],
-        'must': [('must', 1)],
-        'of': [('of', 2)],
-        'retain': [('retain', 1)],
-        'modification': [('modification', 1)],
-        'and': [('and', 3)],
-        'above': [('above', 1)],
-        'all': [('all', 1)],
-        'redistributions': [('redistributions', 1)],
-        'bsd': [('bsd', 1)],
-        'in': [('in', 1)],
-        'conditions': [('conditions', 2)],
-        'disclaimer': [('disclaimer', 1)],
-        'd': [('d', 1)],
-        'code': [('code', 1)],
-        'provided': [('provided', 1)],
-        'notice': [('notice', 1)]
+        '2015': [(1,)]
+        'above': [(1,)]
+        'all': [(1,)]
+        'and': [(3,)]
+        'are': [(2,)]
+        'binary': [(1,)]
+        'bsd': [(1,)]
+        'c': [(1,)]
+        'code': [(1,)]
     }
 
-**Final Reduce**
+**Output**
 
 The default implementation is to return ``(key, iter(values))`` pairs from the
 ``final_reducer()``, which would look something like:
diff --git a/tinymr/__init__.py b/tinymr/__init__.py
index 5ccf28f..f283135 100644
--- a/tinymr/__init__.py
+++ b/tinymr/__init__.py
@@ -5,6 +5,10 @@ Heavily inspired by Spotify's Luigi framework - github.com/Spotify/Luigi
 """
 
 
+import logging
+logging.basicConfig()
+
+
 __version__ = '0.1'
 __author__ = 'Kevin Wurster'
 __email__ = 'wursterk@gmail.com'
diff --git a/tinymr/_backport_heapq.py b/tinymr/_backport_heapq.py
new file mode 100644
index 0000000..a574351
--- /dev/null
+++ b/tinymr/_backport_heapq.py
@@ -0,0 +1,620 @@
+# encoding: utf-8
+
+
+# This module was copied from the cpython source code and maintains its
+# original license.  Modifications were limited to the changes required to
+# support Python 2 + 3 and are provided without any guarantee.  For example,
+# `yield from generator` calls were replaced with:
+#
+#   for value in generator:
+#       yield value
+#
+# and `iterator.__next__` were replaced with:
+#
+#   getattr(iterator, '__next__', getattr(iterator, 'next'))
+#
+# Python 3's `heapq.merge()` accepts a keyfunc for determining how to sort a
+# given object.  Code was copied from:
+# https://github.com/python/cpython/tree/15572204799fb8506645dc1c448135f4e4ffde00
+
+
+"""Heap queue algorithm (a.k.a. priority queue).
+Heaps are arrays for which a[k] <= a[2*k+1] and a[k] <= a[2*k+2] for
+all k, counting elements from 0.  For the sake of comparison,
+non-existing elements are considered to be infinite.  The interesting
+property of a heap is that a[0] is always its smallest element.
+Usage:
+heap = []            # creates an empty heap
+heappush(heap, item) # pushes a new item on the heap
+item = heappop(heap) # pops the smallest item from the heap
+item = heap[0]       # smallest item on the heap without popping it
+heapify(x)           # transforms list into a heap, in-place, in linear time
+item = heapreplace(heap, item) # pops and returns smallest item, and adds
+                               # new item; the heap size is unchanged
+Our API differs from textbook heap algorithms as follows:
+- We use 0-based indexing.  This makes the relationship between the
+  index for a node and the indexes for its children slightly less
+  obvious, but is more suitable since Python uses 0-based indexing.
+- Our heappop() method returns the smallest item, not the largest.
+These two make it possible to view the heap as a regular Python list
+without surprises: heap[0] is the smallest item, and heap.sort()
+maintains the heap invariant!
+"""
+
+# Original code by Kevin O'Connor, augmented by Tim Peters and Raymond Hettinger
+
+__about__ = """Heap queues
+[explanation by François Pinard]
+Heaps are arrays for which a[k] <= a[2*k+1] and a[k] <= a[2*k+2] for
+all k, counting elements from 0.  For the sake of comparison,
+non-existing elements are considered to be infinite.  The interesting
+property of a heap is that a[0] is always its smallest element.
+The strange invariant above is meant to be an efficient memory
+representation for a tournament.  The numbers below are `k', not a[k]:
+                                   0
+                  1                                 2
+          3               4                5               6
+      7       8       9       10      11      12      13      14
+    15 16   17 18   19 20   21 22   23 24   25 26   27 28   29 30
+In the tree above, each cell `k' is topping `2*k+1' and `2*k+2'.  In
+an usual binary tournament we see in sports, each cell is the winner
+over the two cells it tops, and we can trace the winner down the tree
+to see all opponents s/he had.  However, in many computer applications
+of such tournaments, we do not need to trace the history of a winner.
+To be more memory efficient, when a winner is promoted, we try to
+replace it by something else at a lower level, and the rule becomes
+that a cell and the two cells it tops contain three different items,
+but the top cell "wins" over the two topped cells.
+If this heap invariant is protected at all time, index 0 is clearly
+the overall winner.  The simplest algorithmic way to remove it and
+find the "next" winner is to move some loser (let's say cell 30 in the
+diagram above) into the 0 position, and then percolate this new 0 down
+the tree, exchanging values, until the invariant is re-established.
+This is clearly logarithmic on the total number of items in the tree.
+By iterating over all items, you get an O(n ln n) sort.
+A nice feature of this sort is that you can efficiently insert new
+items while the sort is going on, provided that the inserted items are
+not "better" than the last 0'th element you extracted.  This is
+especially useful in simulation contexts, where the tree holds all
+incoming events, and the "win" condition means the smallest scheduled
+time.  When an event schedule other events for execution, they are
+scheduled into the future, so they can easily go into the heap.  So, a
+heap is a good structure for implementing schedulers (this is what I
+used for my MIDI sequencer :-).
+Various structures for implementing schedulers have been extensively
+studied, and heaps are good for this, as they are reasonably speedy,
+the speed is almost constant, and the worst case is not much different
+than the average case.  However, there are other representations which
+are more efficient overall, yet the worst cases might be terrible.
+Heaps are also very useful in big disk sorts.  You most probably all
+know that a big sort implies producing "runs" (which are pre-sorted
+sequences, which size is usually related to the amount of CPU memory),
+followed by a merging passes for these runs, which merging is often
+very cleverly organised[1].  It is very important that the initial
+sort produces the longest runs possible.  Tournaments are a good way
+to that.  If, using all the memory available to hold a tournament, you
+replace and percolate items that happen to fit the current run, you'll
+produce runs which are twice the size of the memory for random input,
+and much better for input fuzzily ordered.
+Moreover, if you output the 0'th item on disk and get an input which
+may not fit in the current tournament (because the value "wins" over
+the last output value), it cannot fit in the heap, so the size of the
+heap decreases.  The freed memory could be cleverly reused immediately
+for progressively building a second heap, which grows at exactly the
+same rate the first heap is melting.  When the first heap completely
+vanishes, you switch heaps and start a new run.  Clever and quite
+effective!
+In a word, heaps are useful memory structures to know.  I use them in
+a few applications, and I think it is good to keep a `heap' module
+around. :-)
+--------------------
+[1] The disk balancing algorithms which are current, nowadays, are
+more annoying than clever, and this is a consequence of the seeking
+capabilities of the disks.  On devices which cannot seek, like big
+tape drives, the story was quite different, and one had to be very
+clever to ensure (far in advance) that each tape movement will be the
+most effective possible (that is, will best participate at
+"progressing" the merge).  Some tapes were even able to read
+backwards, and this was also used to avoid the rewinding time.
+Believe me, real good tape sorts were quite spectacular to watch!
+From all times, sorting has always been a Great Art! :-)
+"""
+
+
+import functools
+
+
+__all__ = ['heappush', 'heappop', 'heapify', 'heapreplace', 'merge',
+           'nlargest', 'nsmallest', 'heappushpop']
+
+
+builtin_next = next
+
+
+def heappush(heap, item):
+    """Push item onto heap, maintaining the heap invariant."""
+    heap.append(item)
+    _siftdown(heap, 0, len(heap)-1)
+
+
+def heappop(heap):
+    """Pop the smallest item off the heap, maintaining the heap invariant."""
+    lastelt = heap.pop()    # raises appropriate IndexError if heap is empty
+    if heap:
+        returnitem = heap[0]
+        heap[0] = lastelt
+        _siftup(heap, 0)
+        return returnitem
+    return lastelt
+
+
+def heapreplace(heap, item):
+    """Pop and return the current smallest value, and add the new item.
+    This is more efficient than heappop() followed by heappush(), and can be
+    more appropriate when using a fixed-size heap.  Note that the value
+    returned may be larger than item!  That constrains reasonable uses of
+    this routine unless written as part of a conditional replacement:
+        if item > heap[0]:
+            item = heapreplace(heap, item)
+    """
+    returnitem = heap[0]    # raises appropriate IndexError if heap is empty
+    heap[0] = item
+    _siftup(heap, 0)
+    return returnitem
+
+
+def heappushpop(heap, item):
+    """Fast version of a heappush followed by a heappop."""
+    if heap and heap[0] < item:
+        item, heap[0] = heap[0], item
+        _siftup(heap, 0)
+    return item
+
+
+def heapify(x):
+    """Transform list into a heap, in-place, in O(len(x)) time."""
+    n = len(x)
+    # Transform bottom-up.  The largest index there's any point to looking at
+    # is the largest with a child index in-range, so must have 2*i + 1 < n,
+    # or i < (n-1)/2.  If n is even = 2*j, this is (2*j-1)/2 = j-1/2 so
+    # j-1 is the largest, which is n//2 - 1.  If n is odd = 2*j+1, this is
+    # (2*j+1-1)/2 = j so j-1 is the largest, and that's again n//2-1.
+    for i in reversed(range(n//2)):
+        _siftup(x, i)
+
+
+def _heappop_max(heap):
+    """Maxheap version of a heappop."""
+    lastelt = heap.pop()    # raises appropriate IndexError if heap is empty
+    if heap:
+        returnitem = heap[0]
+        heap[0] = lastelt
+        _siftup_max(heap, 0)
+        return returnitem
+    return lastelt
+
+
+def _heapreplace_max(heap, item):
+    """Maxheap version of a heappop followed by a heappush."""
+    returnitem = heap[0]    # raises appropriate IndexError if heap is empty
+    heap[0] = item
+    _siftup_max(heap, 0)
+    return returnitem
+
+
+def _heapify_max(x):
+    """Transform list into a maxheap, in-place, in O(len(x)) time."""
+    n = len(x)
+    for i in reversed(range(n//2)):
+        _siftup_max(x, i)
+
+
+# 'heap' is a heap at all indices >= startpos, except possibly for pos.  pos
+# is the index of a leaf with a possibly out-of-order value.  Restore the
+# heap invariant.
+def _siftdown(heap, startpos, pos):
+    newitem = heap[pos]
+    # Follow the path to the root, moving parents down until finding a place
+    # newitem fits.
+    while pos > startpos:
+        parentpos = (pos - 1) >> 1
+        parent = heap[parentpos]
+        if newitem < parent:
+            heap[pos] = parent
+            pos = parentpos
+            continue
+        break
+    heap[pos] = newitem
+
+
+# The child indices of heap index pos are already heaps, and we want to make
+# a heap at index pos too.  We do this by bubbling the smaller child of
+# pos up (and so on with that child's children, etc) until hitting a leaf,
+# then using _siftdown to move the oddball originally at index pos into place.
+#
+# We *could* break out of the loop as soon as we find a pos where newitem <=
+# both its children, but turns out that's not a good idea, and despite that
+# many books write the algorithm that way.  During a heap pop, the last array
+# element is sifted in, and that tends to be large, so that comparing it
+# against values starting from the root usually doesn't pay (= usually doesn't
+# get us out of the loop early).  See Knuth, Volume 3, where this is
+# explained and quantified in an exercise.
+#
+# Cutting the # of comparisons is important, since these routines have no
+# way to extract "the priority" from an array element, so that intelligence
+# is likely to be hiding in custom comparison methods, or in array elements
+# storing (priority, record) tuples.  Comparisons are thus potentially
+# expensive.
+#
+# On random arrays of length 1000, making this change cut the number of
+# comparisons made by heapify() a little, and those made by exhaustive
+# heappop() a lot, in accord with theory.  Here are typical results from 3
+# runs (3 just to demonstrate how small the variance is):
+#
+# Compares needed by heapify     Compares needed by 1000 heappops
+# --------------------------     --------------------------------
+# 1837 cut to 1663               14996 cut to 8680
+# 1855 cut to 1659               14966 cut to 8678
+# 1847 cut to 1660               15024 cut to 8703
+#
+# Building the heap by using heappush() 1000 times instead required
+# 2198, 2148, and 2219 compares:  heapify() is more efficient, when
+# you can use it.
+#
+# The total compares needed by list.sort() on the same lists were 8627,
+# 8627, and 8632 (this should be compared to the sum of heapify() and
+# heappop() compares):  list.sort() is (unsurprisingly!) more efficient
+# for sorting.
+
+
+def _siftup(heap, pos):
+    endpos = len(heap)
+    startpos = pos
+    newitem = heap[pos]
+    # Bubble up the smaller child until hitting a leaf.
+    childpos = 2*pos + 1    # leftmost child position
+    while childpos < endpos:
+        # Set childpos to index of smaller child.
+        rightpos = childpos + 1
+        if rightpos < endpos and not heap[childpos] < heap[rightpos]:
+            childpos = rightpos
+        # Move the smaller child up.
+        heap[pos] = heap[childpos]
+        pos = childpos
+        childpos = 2*pos + 1
+    # The leaf at pos is empty now.  Put newitem there, and bubble it up
+    # to its final resting place (by sifting its parents down).
+    heap[pos] = newitem
+    _siftdown(heap, startpos, pos)
+
+
+def _siftdown_max(heap, startpos, pos):
+    'Maxheap variant of _siftdown'
+    newitem = heap[pos]
+    # Follow the path to the root, moving parents down until finding a place
+    # newitem fits.
+    while pos > startpos:
+        parentpos = (pos - 1) >> 1
+        parent = heap[parentpos]
+        if parent < newitem:
+            heap[pos] = parent
+            pos = parentpos
+            continue
+        break
+    heap[pos] = newitem
+
+
+def _siftup_max(heap, pos):
+    'Maxheap variant of _siftup'
+    endpos = len(heap)
+    startpos = pos
+    newitem = heap[pos]
+    # Bubble up the larger child until hitting a leaf.
+    childpos = 2*pos + 1    # leftmost child position
+    while childpos < endpos:
+        # Set childpos to index of larger child.
+        rightpos = childpos + 1
+        if rightpos < endpos and not heap[rightpos] < heap[childpos]:
+            childpos = rightpos
+        # Move the larger child up.
+        heap[pos] = heap[childpos]
+        pos = childpos
+        childpos = 2*pos + 1
+    # The leaf at pos is empty now.  Put newitem there, and bubble it up
+    # to its final resting place (by sifting its parents down).
+    heap[pos] = newitem
+    _siftdown_max(heap, startpos, pos)
+
+
+def merge(*iterables, **kwargs):
+    '''Merge multiple sorted inputs into a single sorted output.
+    Similar to sorted(itertools.chain(*iterables)) but returns a generator,
+    does not pull the data into memory all at once, and assumes that each of
+    the input streams is already sorted (smallest to largest).
+    >>> list(merge([1,3,5,7], [0,2,4,8], [5,10,15,20], [], [25]))
+    [0, 1, 2, 3, 4, 5, 5, 7, 8, 10, 15, 20, 25]
+    If *key* is not None, applies a key function to each element to determine
+    its sort order.
+    >>> list(merge(['dog', 'horse'], ['cat', 'fish', 'kangaroo'], key=len))
+    ['dog', 'cat', 'fish', 'horse', 'kangaroo']
+    '''
+
+    key = kwargs.pop('key', None)
+    reverse = kwargs.pop('reverse', False)
+
+    assert not kwargs, "Unrecognized kwargs: {}".format(kwargs)
+
+    h = []
+    h_append = h.append
+
+    if reverse:
+        _heapify = _heapify_max
+        _heappop = _heappop_max
+        _heapreplace = _heapreplace_max
+        direction = -1
+    else:
+        _heapify = heapify
+        _heappop = heappop
+        _heapreplace = heapreplace
+        direction = 1
+
+    if key is None:
+        for order, it in enumerate(map(iter, iterables)):
+            try:
+                next = functools.partial(builtin_next, it)
+                # next = getattr(it, '__next__', getattr(it, 'next'))
+                h_append([next(), order * direction, next])
+            except StopIteration:
+                pass
+        _heapify(h)
+        while len(h) > 1:
+            try:
+                while True:
+                    value, order, next = s = h[0]
+                    yield value
+                    s[0] = next()           # raises StopIteration when exhausted
+                    _heapreplace(h, s)      # restore heap condition
+            except StopIteration:
+                _heappop(h)                 # remove empty iterator
+        if h:
+            # fast case when only a single iterator remains
+            value, order, next = h[0]
+            yield value
+            while True:
+                yield next()
+        return
+
+    for order, it in enumerate(map(iter, iterables)):
+        try:
+            next = functools.partial(builtin_next, it)
+            # next = getattr(it, '__next__', getattr(it, 'next'))
+            value = next()
+            h_append([key(value), order * direction, value, next])
+        except StopIteration:
+            pass
+    _heapify(h)
+    while len(h) > 1:
+        try:
+            while True:
+                key_value, order, value, next = s = h[0]
+                yield value
+                value = next()
+                s[0] = key(value)
+                s[2] = value
+                _heapreplace(h, s)
+        except StopIteration:
+            _heappop(h)
+    if h:
+        key_value, order, value, next = h[0]
+        yield value
+        while True:
+            yield next()
+
+
+# Algorithm notes for nlargest() and nsmallest()
+# ==============================================
+#
+# Make a single pass over the data while keeping the k most extreme values
+# in a heap.  Memory consumption is limited to keeping k values in a list.
+#
+# Measured performance for random inputs:
+#
+#                                   number of comparisons
+#    n inputs     k-extreme values  (average of 5 trials)   % more than min()
+# -------------   ----------------  ---------------------   -----------------
+#      1,000           100                  3,317               231.7%
+#     10,000           100                 14,046                40.5%
+#    100,000           100                105,749                 5.7%
+#  1,000,000           100              1,007,751                 0.8%
+# 10,000,000           100             10,009,401                 0.1%
+#
+# Theoretical number of comparisons for k smallest of n random inputs:
+#
+# Step   Comparisons                  Action
+# ----   --------------------------   ---------------------------
+#  1     1.66 * k                     heapify the first k-inputs
+#  2     n - k                        compare remaining elements to top of heap
+#  3     k * (1 + lg2(k)) * ln(n/k)   replace the topmost value on the heap
+#  4     k * lg2(k) - (k/2)           final sort of the k most extreme values
+#
+# Combining and simplifying for a rough estimate gives:
+#
+#        comparisons = n + k * (log(k, 2) * log(n/k) + log(k, 2) + log(n/k))
+#
+# Computing the number of comparisons for step 3:
+# -----------------------------------------------
+# * For the i-th new value from the iterable, the probability of being in the
+#   k most extreme values is k/i.  For example, the probability of the 101st
+#   value seen being in the 100 most extreme values is 100/101.
+# * If the value is a new extreme value, the cost of inserting it into the
+#   heap is 1 + log(k, 2).
+# * The probability times the cost gives:
+#            (k/i) * (1 + log(k, 2))
+# * Summing across the remaining n-k elements gives:
+#            sum((k/i) * (1 + log(k, 2)) for i in range(k+1, n+1))
+# * This reduces to:
+#            (H(n) - H(k)) * k * (1 + log(k, 2))
+# * Where H(n) is the n-th harmonic number estimated by:
+#            gamma = 0.5772156649
+#            H(n) = log(n, e) + gamma + 1 / (2 * n)
+#   http://en.wikipedia.org/wiki/Harmonic_series_(mathematics)#Rate_of_divergence
+# * Substituting the H(n) formula:
+#            comparisons = k * (1 + log(k, 2)) * (log(n/k, e) + (1/n - 1/k) / 2)
+#
+# Worst-case for step 3:
+# ----------------------
+# In the worst case, the input data is reversed sorted so that every new element
+# must be inserted in the heap:
+#
+#             comparisons = 1.66 * k + log(k, 2) * (n - k)
+#
+# Alternative Algorithms
+# ----------------------
+# Other algorithms were not used because they:
+# 1) Took much more auxiliary memory,
+# 2) Made multiple passes over the data.
+# 3) Made more comparisons in common cases (small k, large n, semi-random input).
+# See the more detailed comparison of approach at:
+# http://code.activestate.com/recipes/577573-compare-algorithms-for-heapqsmallest
+
+def nsmallest(n, iterable, key=None):
+    """Find the n smallest elements in a dataset.
+    Equivalent to:  sorted(iterable, key=key)[:n]
+    """
+
+    # Short-cut for n==1 is to use min()
+    if n == 1:
+        it = iter(iterable)
+        sentinel = object()
+        if key is None:
+            result = min(it, default=sentinel)
+        else:
+            result = min(it, default=sentinel, key=key)
+        return [] if result is sentinel else [result]
+
+    # When n>=size, it's faster to use sorted()
+    try:
+        size = len(iterable)
+    except (TypeError, AttributeError):
+        pass
+    else:
+        if n >= size:
+            return sorted(iterable, key=key)[:n]
+
+    # When key is none, use simpler decoration
+    if key is None:
+        it = iter(iterable)
+        # put the range(n) first so that zip() doesn't
+        # consume one too many elements from the iterator
+        result = [(elem, i) for i, elem in zip(range(n), it)]
+        if not result:
+            return result
+        _heapify_max(result)
+        top = result[0][0]
+        order = n
+        _heapreplace = _heapreplace_max
+        for elem in it:
+            if elem < top:
+                _heapreplace(result, (elem, order))
+                top = result[0][0]
+                order += 1
+        result.sort()
+        return [r[0] for r in result]
+
+    # General case, slowest method
+    it = iter(iterable)
+    result = [(key(elem), i, elem) for i, elem in zip(range(n), it)]
+    if not result:
+        return result
+    _heapify_max(result)
+    top = result[0][0]
+    order = n
+    _heapreplace = _heapreplace_max
+    for elem in it:
+        k = key(elem)
+        if k < top:
+            _heapreplace(result, (k, order, elem))
+            top = result[0][0]
+            order += 1
+    result.sort()
+    return [r[2] for r in result]
+
+
+def nlargest(n, iterable, key=None):
+    """Find the n largest elements in a dataset.
+    Equivalent to:  sorted(iterable, key=key, reverse=True)[:n]
+    """
+
+    # Short-cut for n==1 is to use max()
+    if n == 1:
+        it = iter(iterable)
+        sentinel = object()
+        if key is None:
+            result = max(it, default=sentinel)
+        else:
+            result = max(it, default=sentinel, key=key)
+        return [] if result is sentinel else [result]
+
+    # When n>=size, it's faster to use sorted()
+    try:
+        size = len(iterable)
+    except (TypeError, AttributeError):
+        pass
+    else:
+        if n >= size:
+            return sorted(iterable, key=key, reverse=True)[:n]
+
+    # When key is none, use simpler decoration
+    if key is None:
+        it = iter(iterable)
+        result = [(elem, i) for i, elem in zip(range(0, -n, -1), it)]
+        if not result:
+            return result
+        heapify(result)
+        top = result[0][0]
+        order = -n
+        _heapreplace = heapreplace
+        for elem in it:
+            if top < elem:
+                _heapreplace(result, (elem, order))
+                top = result[0][0]
+                order -= 1
+        result.sort(reverse=True)
+        return [r[0] for r in result]
+
+    # General case, slowest method
+    it = iter(iterable)
+    result = [(key(elem), i, elem) for i, elem in zip(range(0, -n, -1), it)]
+    if not result:
+        return result
+    heapify(result)
+    top = result[0][0]
+    order = -n
+    _heapreplace = heapreplace
+    for elem in it:
+        k = key(elem)
+        if top < k:
+            _heapreplace(result, (k, order, elem))
+            top = result[0][0]
+            order -= 1
+    result.sort(reverse=True)
+    return [r[2] for r in result]
+
+
+# If available, use C implementation
+try:
+    from _heapq import *
+except ImportError:
+    pass
+try:
+    from _heapq import _heapreplace_max
+except ImportError:
+    pass
+try:
+    from _heapq import _heapify_max
+except ImportError:
+    pass
+try:
+    from _heapq import _heappop_max
+except ImportError:
+    pass
diff --git a/tinymr/_mrtools.py b/tinymr/_mrtools.py
new file mode 100644
index 0000000..1a200db
--- /dev/null
+++ b/tinymr/_mrtools.py
@@ -0,0 +1,100 @@
+"""
+Helpers for MapReduce implementations.
+"""
+
+
+from collections import namedtuple
+
+import six
+
+from tinymr.tools import sorter
+
+
+def strip_sort_key(kv_stream):
+
+    """
+    Given a stream of `(key, [(sort, data), (sort, data)])` with sort key
+    intact, remove the key from the values.
+
+    Example:
+
+        [
+            ('key1', [(10, data1), (3, data25)]),
+            ('key2', [(200, data100), (250, data67))
+        ]
+
+    Produces:
+
+        [
+            ('key1', [data1, data25]),
+            ('key2', [data100, data67)
+        ]
+
+    Parameters
+    ----------
+    kv_stream : dict or iter
+        Dictionary like `{key: [(sort, data)]}` or a stream of tuples like
+        `(key, [(sort, data])`.
+
+    Yields
+    ------
+    tuple
+        `(key, [data, data, ...])`
+    """
+
+    kv_stream = six.iteritems(kv_stream) if isinstance(kv_stream, dict) else kv_stream
+    return ((k, tuple(i[-1] for i in v)) for k, v in kv_stream)
+
+
+def sort_partitioned_values(kv_stream):
+
+    """
+    Given a stream of `(key, [(sort, data), (sort, data)])` sort the values
+    for every key.
+
+    Example:
+
+        [
+            ('key1', [(10, data), (3, data)]),
+            ('key2', [(200, data), (250, data))
+        ]
+
+    Produces:
+
+        [
+            ('key1', [(3, data), (10, data)]),
+            ('key2', [(200, data), (250, data))
+        ]
+
+    Parameters
+    ----------
+    kv_stream : dict or iter
+        Dictionary like `{key: [(sort, data]}` or a stream of tuples like
+        `(key, [(sort, data])`.
+
+    Yields
+    ------
+    tuple
+        `(key, [(sort, data), (sort, data), ...])`
+    """
+
+    kv_stream = six.iteritems(kv_stream) if isinstance(kv_stream, dict) else kv_stream
+    return ((k, sorter(v, key=lambda x: x[0])) for k, v in kv_stream)
+
+
+class ReduceJob(namedtuple('ReduceJob', ['reducer', 'sort', 'jobs', 'chunksize'])):
+
+    """
+    Describes a reduce job.  Makes keeping track of multiple reducers easier.
+
+    Parameters
+    ----------
+    reducer : callable
+        Does the reducing.  Has a signature like `reducer(key, iter(values))`.
+    sort : bool
+        Determines if the partitioned values should be sorted.
+    jobs : int
+        Number of jobs to run in parallel.
+    chunksize : int
+        Amount of data to pass to one `job`.
+    """
diff --git a/tinymr/base.py b/tinymr/base.py
index 5c5c07f..eba8905 100644
--- a/tinymr/base.py
+++ b/tinymr/base.py
@@ -3,30 +3,30 @@ Base classes.  Subclass away!
 """
 
 
-from collections import defaultdict
-from contextlib import contextmanager
 from itertools import chain
 
 import six
 
+from tinymr import _mrtools
+from tinymr import errors
+from tinymr import tools
 
-class MRBase(object):
+
+class BaseMapReduce(object):
 
     """
     Base class for various MapReduce implementations.  Not all methods are
     used by every implementation.
     """
 
-    _closed = False
-
     def __enter__(self):
 
         """
         See `__exit__` for context manager usage.
         """
 
-        if self.closed:
-            raise IOError("MapReduce task is closed - cannot reuse.")
+        # if self.closed:
+        #     raise IOError("MapReduce task is closed - cannot reuse.")
 
         return self
 
@@ -44,10 +44,109 @@ class MRBase(object):
         self.close()
 
     @property
-    def sort_map(self):
+    def jobs(self):
 
         """
-        Sort the output from the map phase before the combine phase.
+        Default number of jobs to run in parallel for each phase.
+
+        Returns
+        -------
+        int
+        """
+
+        return 1
+
+    @property
+    def map_jobs(self):
+
+        """
+        Number of jobs to run in parallel during the map phase.  Defaults
+        to `jobs` property.
+
+        Returns
+        -------
+        int
+        """
+
+        return self.jobs
+
+    @property
+    def sort_jobs(self):
+
+        """
+        Number of jobs to run in parallel during the sort phases.  Defaults
+        to `jobs` property.
+
+        Returns
+        -------
+        int
+        """
+
+        return self.jobs
+
+    @property
+    def reduce_jobs(self):
+
+        """
+        Number of jobs to run in parallel during the reduce phase.  If `None`,
+        defaults to `jobs` property.
+
+        Returns
+        -------
+        int
+        """
+
+        return self.jobs
+
+    @property
+    def chunksize(self):
+
+        """
+        Amount of data to process in each `job`.
+
+        Returns
+        -------
+        int
+        """
+
+        return 1
+
+    @property
+    def map_chunksize(self):
+
+        """
+        Amount of data to process in each `job` during the map phase.
+        Defaults to `chunksize`.
+        """
+
+        return self.chunksize
+
+    @property
+    def sort_chunksize(self):
+
+        """
+        Amount of data to process in each `job` during the sort phase.
+        Defaults to `chunksize`.
+        """
+
+        return self.chunksize
+
+    @property
+    def reduce_chunksize(self):
+
+        """
+        Amount of data to process in each `job` during the reduce phase.
+        Defaults to `chunksize`.
+        """
+
+        return self.chunksize
+
+    @property
+    def sort(self):
+
+        """
+        Disable all sorting phases.  Setting individual properties overrides
+        this setting for individual phases.
 
         Returns
         -------
@@ -57,56 +156,59 @@ class MRBase(object):
         return True
 
     @property
-    def sort_combine(self):
+    def sort_map(self):
 
         """
-        Sort the output from the combine phase before the partition phase.
+        Sort the output from the map phase before the combine phase.
 
         Returns
         -------
         bool
         """
 
-        return True
+        return self.sort
 
     @property
-    def sort_final_reduce(self):
+    def sort_combine(self):
 
         """
-        Pass data to the `final_reducer()` sorted by key.
+        Sort the output from the combine phase before the partition phase.
 
         Returns
         -------
         bool
         """
 
-        return True
+        return self.sort
 
     @property
-    def sort_reduce(self):
+    def sort_output(self):
 
         """
-        Sort the output from the `reducer()` phase before the `final_reducer().
+        Pass data to `output()` sorted by key.
 
         Returns
         -------
         bool
         """
 
-        return True
+        return self.sort
 
     @property
-    def closed(self):
+    def sort_reduce(self):
 
         """
-        Indicates whether or not the MapReduce task is closed.
+        Sort the output from each `reducer()` before executing the next or
+        before being passed to `output()`.
+
+        Define one property per reducer, so `reducer2()` would be `sort_reduce2`.
 
         Returns
         -------
         bool
         """
 
-        return self._closed
+        return self.sort
 
     def close(self):
 
@@ -158,7 +260,10 @@ class MRBase(object):
         See `reducer()` for more information.
         """
 
-        raise NotImplementedError
+        # Not required so we raise a special exception that we can catch later
+        # Raising NotImplementedError also causes linters and code inspectors
+        # to prompt the user to implement this method when it is not required.
+        raise errors._CombinerNotImplemented
 
     def init_reduce(self):
 
@@ -189,7 +294,7 @@ class MRBase(object):
 
         raise NotImplementedError
 
-    def final_reducer(self, pairs):
+    def output(self, pairs):
 
         """
         Receives `(key, value)` pairs from each reducer.  The output of this
@@ -212,89 +317,94 @@ class MRBase(object):
 
         return ((key, tuple(values)) for key, values in pairs)
 
-    @contextmanager
-    def _partition(self, psd_stream):
+    @property
+    def _reduce_jobs(self):
+
+        reducers = tools.sorter(filter(
+            lambda x: not x.startswith('_') and 'reducer' in x,
+            dir(self)))
+
+        for r in reducers:
+            yield _mrtools.ReduceJob(
+                reducer=getattr(self, r),
+                sort=getattr(self, 'sort_{}'.format(r.replace('reducer', 'reduce'))),
+                jobs=getattr(self, '{}_jobs'.format(r.replace('reducer', 'reduce'))),
+                chunksize=getattr(self, '{}_jobs'.format(r.replace('reducer', 'reduce'))))
+
+    def _map_combine_partition(self, stream):
 
         """
-        Context manager to partition data and destroy it when finished.
+        Run `mapper()`, partition, `combiner()` (if implemented) and partition
+        on a chunk of input data.  Data may be sorted between each phase
+        according to the control properties.
+
+        Produces a dictionary of partitioned data with sort keys intact.
 
         Parameters
         ----------
-        psd_stream : iter
-            Produces `(partition, sort, data)` tuples.
-        sort_key : bool, optional
-            Some MapReduce implementations don't benefit from sorting, and
-            therefore do not pass around a sort key.  Set to `False` in this
-            case.
+        stream : iter
+            Input data passed to the MapReduce task.
 
         Returns
         -------
-        defaultdict
-            Keys are partition keys and values are lists of `(sort, data)` tuples.
+        dict
+            {key: [(sort, data), (sort, data), ...]}
         """
 
-        partitioned = defaultdict(list)
-
-        try:
-
-            for kv_data in psd_stream:
-                partitioned[kv_data[0]].append(kv_data)
-
-            yield partitioned
-
-        finally:
-            partitioned = None
-
-    def _sorter(self, key_values, fake=False):
-
-        """
-        Produces sorted data without any keys.
+        # Map, partition, and convert back to a `(key, [v, a, l, u, e, s])` stream
+        mapped = chain(*(self.mapper(item) for item in stream))
+        map_partitioned = tools.partition(mapped)
+        map_partitioned = six.iteritems(map_partitioned)
 
-        Parameters
-        ----------
-        data : iter
-            Produces tuples from the map phase.
-        fake : bool, optional
-            Don't do the sort - just strip off the data key.
+        if self.sort_map:
+            map_partitioned = _mrtools.sort_partitioned_values(map_partitioned)
 
-        Yields
-        ------
-        iter
-            Sorted data without keys.
-        """
+        try:
 
-        for key, values in key_values:
-            values = iter(values) if fake else sorted(values, key=lambda x: x[-2])
-            yield key, (v[-1] for v in values)
+            # The generators within this method get weird and don't break
+            # properly when wrapped in this try/except
+            # Instead we just kinda probe the `combiner()` to see if it exists
+            # and hope it doesn't do any setup.
+            self.combiner(None, None)
+            has_combiner = True
 
-    def _map(self, stream):
+        except errors.CombinerNotImplemented:
+            has_combiner = False
 
-        """
-        Apply `mapper()` across the input stream.
-        """
+        if has_combiner:
+            map_partitioned = _mrtools.strip_sort_key(map_partitioned)
+            combined = chain(*(self.combiner(k, v) for k, v in map_partitioned))
+            combine_partitioned = tools.partition(combined)
+            combine_partitioned = six.iteritems(combine_partitioned)
 
-        return chain(*(self.mapper(item) for item in stream))
+        # If we don't have a combiner then we don't need to partition either
+        # because we're already dealing with partitioned output from the
+        # map phase
+        else:
+            combine_partitioned = map_partitioned
 
-    def _reduce(self, kv_stream):
+        # If we don't have a combiner or if we're not sorting, then whatever
+        # we got from the mapper is good enough
+        if has_combiner and self.sort_combine:
+            combine_partitioned = _mrtools.sort_partitioned_values(combine_partitioned)
 
-        """
-        Apply the `reducer()` across a stream of `(key, values)` pairs.
-        """
+        return dict(combine_partitioned)
 
-        return chain(*(self.reducer(key, values) for key, values in kv_stream))
+    def _reduce_partition(self, stream, reducer, sort):
 
-    def _combine(self, kv_stream):
+        reduced = chain(*(reducer(k, v) for k, v in stream))
+        partitioned = tools.partition(reduced)
+        partitioned = six.iteritems(partitioned)
 
-        """
-        Apply the `combiner()` across a stream of `(key, values)` pairs.
-        """
+        if sort:
+            partitioned = _mrtools.sort_partitioned_values(partitioned)
 
-        return chain(*(self.combiner(key, values) for key, values in kv_stream))
+        return tuple(partitioned)
 
-    def _final_reducer_sorter(self, kv_stream):
+    def _output_sorter(self, kv_stream):
 
         """
-        Sort data by key before it enters the `final_reducer()`.
+        Sort data by key before it enters `output()`.
 
         Parameters
         ----------
@@ -306,4 +416,4 @@ class MRBase(object):
         tuple
         """
 
-        return ((k, v) for k, v in sorted(kv_stream, key=lambda x: x[0]))
+        return ((k, v) for k, v in tools.sorter(kv_stream, key=lambda x: x[0]))
diff --git a/tinymr/errors.py b/tinymr/errors.py
new file mode 100644
index 0000000..ee546cc
--- /dev/null
+++ b/tinymr/errors.py
@@ -0,0 +1,38 @@
+"""
+tinymr specific exceptions.
+"""
+
+
+class MRException(Exception):
+
+    """
+    Base exception for tinymr.
+    """
+
+
+class UnorderableKeys(MRException):
+
+    """
+    Encountered keys during a sort operation that could not be ordered.  This
+    could mean that some keys are `str`, some are `int`, some are `None`, etc.
+    """
+
+
+class CombinerNotImplemented(MRException):
+
+    """
+    MapReduce task does not implement a `combiner()`.
+    """
+
+
+class ClosedTask(MRException):
+
+    """
+    Cannot re-use closed MapReduce tasks.
+    """
+
+
+# Instantiated exceptions to make sure we get a clear message
+_UnorderableKeys = UnorderableKeys(UnorderableKeys.__doc__.strip())
+_CombinerNotImplemented = CombinerNotImplemented(CombinerNotImplemented.__doc__.strip())
+_ClosedTask = ClosedTask(ClosedTask.__doc__.strip())
diff --git a/tinymr/memory.py b/tinymr/memory.py
index a136a8e..74245f2 100644
--- a/tinymr/memory.py
+++ b/tinymr/memory.py
@@ -3,81 +3,47 @@ In-memory MapReduce - for those weird use cases ...
 """
 
 
-from itertools import chain
-import multiprocessing as mp
-
-import six
-
-import tinymr as mr
-import tinymr.base
-import tinymr.tools
-
-
-# class MRParallelNoSort(mr.base.MRBase):
-#
-#     @property
-#     def jobs(self):
-#
-#         """
-#         Number of tasks to execute in parallel.
-#         """
-#
-#         return 1
-#
-#     @property
-#     def map_size(self):
-#
-#         """
-#         Number of items from the input data stream to hand to each mapper.
-#         """
-#
-#         return 1
-#
-#     def __call__(self, stream):
-#
-#         stream = mr.tools.slicer(stream, self.map_size)
-#
-#         combined = chain(
-#             *mp.Pool(self.jobs).imap_unordered(self._map_partition_combine, stream))
-#
-#         with self._partition_no_sort(combined) as partitioned:
-#             partitioned = tuple(six.iteritems(partitioned))
-#
-#         reduced = tuple(mp.Pool(self.jobs).imap_unordered(self._imap_reducer, partitioned))
-#
-#         with self._partition_no_sort(reduced) as partitioned:
-#             return self.final_reducer(six.iteritems(partitioned))
-#
-#     def _imap_reducer(self, pair):
-#
-#         """
-#         Adapter to integrate `reducer()` into the `imap_unordered()` API.
-#         """
-#
-#         return tuple(self.reducer(*pair))
-
-
-class MRSerial(mr.base.MRBase):
-
-    """
-    For MapReduce operations that don't benefit from sorting or parallelism.
-
-    The `mapper()` and `reducer()` must yield 2 element tuples.  The first
-    element is used for partitioning and the second is data.
-    """
+import functools
+import logging
+
+from tinymr import _mrtools
+from tinymr import base
+from tinymr import tools
+from tinymr.tools import runner
+
+
+logger = logging.getLogger('tinymr')
+logger.setLevel(logging.DEBUG)
+
+
+class MapReduce(base.BaseMapReduce):
 
     def __call__(self, stream):
 
-        with self._partition(self._map(stream)) as partitioned:
+        sliced = tools.slicer(stream, self.map_chunksize)
+
+        # Map, partition, combine, partition
+        with runner(self._map_combine_partition, sliced, self.map_jobs) as mcp:
+            partitioned = tools.merge_partitions(*mcp, sort=self.sort_combine)
+
+        self.init_reduce()
+
+        # Run all partition jobs
+        reducer_input = partitioned
+        for rj in self._reduce_jobs:
+
+            func = functools.partial(
+                self._reduce_partition, reducer=rj.reducer, sort=rj.sort)
 
-            sorted_data = self._sorter(six.iteritems(partitioned), fake=self.sort_map)
+            reducer_input = _mrtools.strip_sort_key(reducer_input)
+            sliced = tools.slicer(reducer_input, rj.chunksize)
 
-        with self._partition(self._reduce(sorted_data)) as partitioned:
+            with runner(func, sliced, rj.jobs) as reduced:
+                partitioned = tools.merge_partitions(*reduced, sort=rj.sort)
 
-            sorted_data = self._sorter(six.iteritems(partitioned), fake=self.sort_reduce)
+        partitioned = _mrtools.strip_sort_key(partitioned)
 
-            if self.sort_final_reduce:
-                sorted_data = self._final_reducer_sorter(sorted_data)
+        if self.sort_output:
+            partitioned = self._output_sorter(partitioned)
 
-            self.init_reduce()
-            return self.final_reducer(sorted_data)
+        return self.output(partitioned)
diff --git a/tinymr/tools.py b/tinymr/tools.py
index 422faa3..eddc9e5 100644
--- a/tinymr/tools.py
+++ b/tinymr/tools.py
@@ -3,12 +3,37 @@ Tools for building MapReduce implementations.
 """
 
 
-from collections import OrderedDict
+from collections import defaultdict
 import itertools as it
 import multiprocessing as mp
 
+import six
 from six.moves import zip
 
+from tinymr._backport_heapq import merge as heapq_merge
+from tinymr import errors
+
+
+# Make instance methods pickle-able in Python 2
+# Instance methods are not available as a type, so we have to create a tiny
+# class so we can grab an instance method
+# We then register our improved _reduce_method() with copy_reg so pickle knows
+# what to do.
+if six.PY2:  # pragma: no cover
+    import copy_reg
+
+    class _I:
+        def m(self):
+            pass
+
+    def _reduce_method(m):
+        if m.im_self is None:
+            return getattr, (m.im_class, m.im_func.func_name)
+        else:
+            return getattr, (m.im_self, m.im_func.func_name)
+
+    copy_reg.pickle(type(_I().m), _reduce_method)
+
 
 def slicer(iterable, chunksize):
 
@@ -46,53 +71,73 @@ def slicer(iterable, chunksize):
             raise StopIteration
 
 
-def runner(func, iterable, jobs):
+class runner(object):
 
     """
     The `multiprocessing` module can be difficult to debug and introduces some
     overhead that isn't needed when only running one job.  Use a generator in
     this case instead.
 
-    Parameters
-    ----------
-    func : callable
-        Callable object to map across `iterable`.
-    iterable : iter
-        Data to process.
-    jobs : int
+    Wrapped in a class to make the context syntax optional.
     """
 
-    if jobs < 1:
-        raise ValueError("jobs must be >= 1, not: {}".format(jobs))
-    elif jobs == 1:
-        return (func(i) for i in iterable)
-    else:
-        return mp.Pool(jobs).imap_unordered(func, iterable)
+    def __init__(self, func, iterable, jobs):
+
+        """
+        Parameters
+        ----------
+        func : callable
+            Callable object to map across `iterable`.
+        iterable : iter
+            Data to process.
+        jobs : int
+            Number of `multiprocessing` jobs.
+        """
+
+        self._func = func
+        self._iterable = iterable
+        self._jobs = jobs
+        self._closed = False
+
+        if jobs < 1:
+            raise ValueError("jobs must be >= 1, not: {}".format(jobs))
+        elif jobs == 1:
+            self._pool = None
+            self._proc = (func(i) for i in iterable)
+        else:
+            self._pool = mp.Pool(jobs)
+            self._proc = self._pool.imap_unordered(func, iterable)
+
+    def __enter__(self):
+        return self
 
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
 
-class DefaultOrderedDict(OrderedDict):
+    def __repr__(self):
+        return "{cname}(func={func}, iterable={iterable}, jobs={jobs})".format(
+            cname=self.__class__.__name__,
+            func=repr(self._func),
+            iterable=repr(self._iterable),
+            jobs=self._jobs)
 
-    def __init__(self, default_factory, *args, **kwargs):
+    def __iter__(self):
+        return self._proc
 
-        if not callable(default_factory):
-            raise TypeError("default_factory must be callable")
+    def __next__(self):
+        return next(self._proc)
 
-        super(DefaultOrderedDict, self).__init__(*args, **kwargs)
-        self.default_factory = default_factory
+    next = __next__
 
-    def __missing__(self, key):
-        v = self.default_factory()
-        super(DefaultOrderedDict, self).__setitem__(key, v)
-        return v
+    def close(self):
 
-    def __repr__(self):
-        return "{cname}({df}, {dr})".format(
-            cname=self.__class__.__name__,
-            df=self.default_factory,
-            dr=super(DefaultOrderedDict, self).__repr__())
+        """
+        Close the `multiprocessing` pool if we're using it.
+        """
 
-    def copy(self):
-        return self.__class__(self.default_factory, self)
+        if self._pool is not None:
+            self._pool.close()
+        self._closed = True
 
 
 def mapkey(key, values):
@@ -121,4 +166,205 @@ def mapkey(key, values):
     iter
     """
 
-    return zip(it.cycle([key]), values)
+    return zip(it.repeat(key), values)
+
+
+def sorter(*args, **kwargs):
+
+    """
+    Wrapper for the builtin `sorted()` that produces a better error when
+    unorderable types are encountered.
+
+    Instead of:
+
+        >>> sorted(['1', 1])
+        Traceback (most recent call last):
+          File "<stdin>", line 1, in <module>
+        TypeError: unorderable types: int() < str()
+
+    we get a `tinymr.errors.UnorderableKeys` exception.
+
+    Python 2 is much more forgiving of unorderable types so the example above
+    does not raise an exception.
+
+    Parameters
+    ----------
+    *args : *args
+        Positional arguments for `sorted()`.
+    **kwargs : **kwargs
+        Keyword arguments for `sorted()`.
+
+    Raises
+    ------
+    tinymr.errors.UnorderableKeys
+
+    Returns
+    -------
+    list
+        Output from `sorted()`.
+    """
+
+    try:
+        return sorted(*args, **kwargs)
+    except TypeError as e:
+        if 'unorderable' in str(e):
+            raise errors._UnorderableKeys
+        else:
+            raise e
+
+
+def partition(key_values):
+
+    """
+    Given a stream of `(key, value)` tuples, group them by key into a dict.
+    Equivalent to the code below, but faster:
+
+        >>> from itertools import groupby
+        >>> {k: list(v) for k, v in groupby(key_values, key=lambda x: x[0])}
+
+    Example:
+
+        >>> data = [('key1', 1), ('key1', 2), ('key2', None)]
+        >>> partition(data)
+        {
+            'key1': [('key1', 1), ('key1', 2)],
+            'key2': [('key2', None)]
+        }
+
+    Parameters
+    ----------
+    key_values : iter
+        Tuples - typically `(key, value)`, although only the first key is
+
+    Returns
+    -------
+    dict
+    """
+
+    out = defaultdict(list)
+    for data in key_values:
+        out[data[0]].append(data[1:])
+
+    return dict(out)
+
+
+class Orderable(object):
+
+    """
+    Make any object orderable.
+    """
+
+    __slots__ = ['_obj', '_lt', '_le', '_gt', '_ge', '_eq']
+
+    def __init__(self, obj, lt=True, le=True, gt=False, ge=False, eq=False):
+
+        """
+        Default parameters make the object sort as less than or equal to.
+
+        Parameters
+        ----------
+        obj : object
+            The object being made orderable.
+        lt : bool, optional
+            Set `__lt__()` evaluation.
+        le : bool, optional
+            Set `__le__()` evaluation.
+        gt : bool, optional
+            Set `__gt__()` evaluation.
+        ge : bool, optional
+            Set `__ge__()` evaluation.
+        eq : bool or None, optional
+            Set `__eq__()` evaluation.  Set to `None` to enable a real
+            equality check.
+        """
+
+        self._obj = obj
+        self._lt = lt
+        self._le = le
+        self._gt = gt
+        self._ge = ge
+        self._eq = eq
+
+    @property
+    def obj(self):
+
+        """
+        Handle to the object being made orderable.
+        """
+
+        return self._obj
+
+    def __lt__(self, other):
+        return self._lt
+
+    def __le__(self, other):
+        return self._le
+
+    def __gt__(self, other):
+        return self._gt
+
+    def __ge__(self, other):
+        return self._ge
+
+    def __eq__(self, other):
+        if self._eq is None:
+            return isinstance(other, self.__class__) and other.obj == self.obj
+        else:
+            return self._eq
+
+
+class _OrderableNone(Orderable):
+
+    """
+    Like `None` but orderable.
+    """
+
+    def __init__(self):
+
+        """
+        Use the instantiated `OrderableNone` variable.
+        """
+
+        super(_OrderableNone, self).__init__(None, eq=None)
+
+
+# Instantiate so we can make it more None-like
+OrderableNone = _OrderableNone()
+
+
+def merge_partitions(*partitions, **kwargs):
+
+    """
+    Merge data from multiple `partition()` operations into one dictionary.
+
+    Parameters
+    ----------
+    partitions : *args
+        Dictionaries from `partition()`.
+    sort : bool, optional
+        Sort partitioned data as it is merged.  Uses `heapq.merge()` so within
+        each partition's key, all values must be sorted smallest to largest.
+
+    Returns
+    -------
+    dict
+        {key: [values]}
+    """
+
+    sort = kwargs.pop('sort', False)
+    assert not kwargs, "Unrecognized kwargs: {}".format(kwargs)
+
+    partitions = (six.iteritems(ptn) if isinstance(ptn, dict) else ptn for ptn in partitions)
+
+    out = defaultdict(list)
+
+    if not sort:
+        for ptn in partitions:
+            for key, values in ptn:
+                out[key].extend(values)
+    else:
+        for ptn in partitions:
+            for key, values in ptn:
+                out[key] = tuple(heapq_merge(out[key], values, key=lambda x: x[0]))
+
+    return dict(out)

Parallel mapping with a chunksize
The inputer iterable can be read in chunks and shoved off into `mapper()` subprocesses would be extremely helpful.
diff --git a/tests/conftest.py b/tests/conftest.py
index c9ec77e..569196d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -39,7 +39,7 @@ def tiny_text_wc_output():
 @pytest.fixture(scope='function')
 def mr_wordcount_memory_no_sort():
 
-    class WordCount(mr.memory.MRSerial):
+    class WordCount(mr.memory.MapReduce):
 
         def mapper(self, item):
             for word in item.split():
@@ -48,7 +48,7 @@ def mr_wordcount_memory_no_sort():
         def reducer(self, key, values):
             yield key, sum(values)
 
-        def final_reducer(self, pairs):
+        def output(self, pairs):
             return {k: tuple(v)[0] for k, v in pairs}
 
     return WordCount
diff --git a/tests/test_base.py b/tests/test_base.py
index 0ce44ee..7468888 100644
--- a/tests/test_base.py
+++ b/tests/test_base.py
@@ -6,14 +6,15 @@ Unittests for tinymr.base
 import pytest
 
 from tinymr import base
+from tinymr import errors
 
 
 def test_not_implemented_methods():
 
-    mr = base.MRBase()
+    mr = base.BaseMapReduce()
     with pytest.raises(NotImplementedError):
         mr.mapper(None)
-    with pytest.raises(NotImplementedError):
+    with pytest.raises(errors.CombinerNotImplemented):
         mr.combiner(None, None)
     with pytest.raises(NotImplementedError):
         mr.reducer(None, None)
@@ -21,59 +22,68 @@ def test_not_implemented_methods():
 
 def test_default_settings():
 
-    mr = base.MRBase()
+    mr = base.BaseMapReduce()
     assert mr.sort_map
     assert mr.sort_combine
     assert mr.sort_reduce
-    assert mr.sort_final_reduce
+    assert mr.sort_output
 
 
 def test_default_methods():
 
-    mr = base.MRBase()
+    mr = base.BaseMapReduce()
     expected = [(i, tuple(range(i))) for i in range(1, 10)]
-    assert list(mr.final_reducer(expected)) == expected
-
-
-def test_context_manager():
-
-    class MR(base.MRBase):
-
-        def __init__(self):
-            self._closed = False
-
-        def close(self):
-            self._closed = True
-
-    with MR() as mr:
-        assert not mr.closed
-    assert mr.closed
-
-
-def test_no_context_manager():
-
-    class MR(base.MRBase):
-
-        def close(self):
-            self._closed = True
-
-    mr = MR()
-    assert not mr.closed
-    mr.close()
-    assert mr.closed
-    assert not MR._closed
-    assert not MR().closed
-
-
-def test_cant_reuse_tasks():
-
-    class MR(base.MRBase):
-        pass
-
-    with MR() as mr:
-        pass
-
-    assert mr.closed
-    with pytest.raises(IOError):
-        with mr as c:
-            pass
+    assert list(mr.output(expected)) == expected
+
+
+# def test_context_manager():
+#
+#     class MapReduce(base.BaseMapReduce):
+#
+#         def __init__(self):
+#             self._closed = False
+#
+#         def close(self):
+#             self._closed = True
+#
+#     with MapReduce() as mr:
+#         assert not mr.closed
+#     assert mr.closed
+
+
+# def test_no_context_manager():
+#
+#     class MapReduce(base.BaseMapReduce):
+#
+#         def close(self):
+#             self._closed = True
+#
+#     mr = MapReduce()
+#     assert not mr.closed
+#     mr.close()
+#     assert mr.closed
+#     assert not MapReduce._closed
+#     assert not MapReduce().closed
+
+
+# def test_cannot_reuse_tasks():
+#
+#     class MapReduce(base.BaseMapReduce):
+#         pass
+#
+#     with MapReduce() as mr:
+#         pass
+#
+#     # assert mr.closed
+#     with pytest.raises(IOError):
+#         with mr as c:
+#             pass
+
+
+# def test_runtime_validate():
+#
+#     class MapReduce(base.BaseMapReduce):
+#         closed = True
+#
+#     with pytest.raises(errors.ClosedTask):
+#         MapReduce()._runtime_validate()
diff --git a/tests/test_memory.py b/tests/test_memory.py
index 39314a0..59078cd 100644
--- a/tests/test_memory.py
+++ b/tests/test_memory.py
@@ -7,14 +7,14 @@ import tinymr as mr
 import tinymr.memory
 
 
-def test_MRSerial_no_sort(tiny_text, tiny_text_wc_output, mr_wordcount_memory_no_sort):
-
-    with mr_wordcount_memory_no_sort() as wc:
-        actual = wc(tiny_text.splitlines())
-    assert actual == tiny_text_wc_output
+# def test_MapReduce_no_sort(tiny_text, tiny_text_wc_output, mr_wordcount_memory_no_sort):
+#
+#     with mr_wordcount_memory_no_sort() as wc:
+#         actual = wc(tiny_text.splitlines())
+#     assert actual == tiny_text_wc_output
 
 
-def test_MRSerial_init_reduce(tiny_text, tiny_text_wc_output, mr_wordcount_memory_no_sort):
+def test_MapReduce_init_reduce(tiny_text, tiny_text_wc_output, mr_wordcount_memory_no_sort):
 
     class WCInitReduce(mr_wordcount_memory_no_sort):
 
@@ -23,3 +23,175 @@ def test_MRSerial_init_reduce(tiny_text, tiny_text_wc_output, mr_wordcount_memor
 
     with WCInitReduce() as wc:
         actual = wc(tiny_text.splitlines())
+
+
+def test_MapReduce_sort():
+
+    """
+    Make sure enabling sorting actually sorts.
+    """
+
+    text = [
+        'key2 sort2 data2',
+        'key2 sort1 data1',
+        'key3 sort2 data2',
+        'key3 sort1 data1',
+        'key1 sort2 data2',
+        'key1 sort1 data1'
+    ]
+
+    class WordCount(mr.memory.MapReduce):
+
+        # Make sure everything gets sent to a single map + combine
+        chunksize = 10
+
+        def mapper(self, item):
+            yield item.split()
+
+        def combiner(self, key, values):
+
+            d1, d2 = list(values)
+            assert [d1, d2] == ['data1', 'data2']
+
+            yield key, 'sort2', d2
+            yield key, 'sort1', d1
+
+        def reducer(self, key, values):
+
+            d1, d2 = list(values)
+            assert [d1, d2] == ['data1', 'data2']
+
+            yield key, 'sort2', d2
+            yield key, 'sort1', d1
+
+    wc = WordCount()
+
+    for attr in ('jobs', 'map_jobs', 'sort_jobs', 'reduce_jobs'):
+        assert getattr(wc, attr) == 1
+    for attr in ('sort', 'sort_map', 'sort_combine', 'sort_reduce', 'sort_output'):
+        assert getattr(wc, attr)
+    for attr in ('chunksize', 'map_chunksize', 'reduce_chunksize', 'sort_chunksize'):
+        assert getattr(wc, attr) == 10
+
+    assert tuple(wc(text)) == (
+        ('key1', ('data1', 'data2')),
+        ('key2', ('data1', 'data2')),
+        ('key3', ('data1', 'data2')))
+
+
+def test_MapReduce_no_sort():
+
+    """
+    Make sure that disabling sorting actually disables sorting.
+    """
+
+    text = [
+        'key2 sort2 data2',
+        'key2 sort1 data1',
+        'key3 sort2 data2',
+        'key3 sort1 data1',
+        'key1 sort2 data2',
+        'key1 sort1 data1'
+    ]
+
+    class WordCount(mr.memory.MapReduce):
+
+        # Make sure everything gets sent to a single map + combine
+        chunksize = 10
+        sort = False
+
+        def mapper(self, item):
+            yield item.split()
+
+        def combiner(self, key, values):
+
+            d2, d1 = list(values)
+            assert [d2, d1] == ['data2', 'data1']
+
+            yield key, 'sort2', d2
+            yield key, 'sort1', d1
+
+        def _final_reducer_sorter(self, kv_stream):
+            raise Exception("Shouldn't hit this.")
+
+        def reducer(self, key, values):
+
+            d2, d1 = list(values)
+            assert [d2, d1] == ['data2', 'data1']
+
+            yield key, 'sort2', d2
+            yield key, 'sort1', d1
+
+    wc = WordCount()
+    for attr in ('jobs', 'map_jobs', 'sort_jobs', 'reduce_jobs'):
+        assert getattr(wc, attr) == 1
+    for attr in ('sort', 'sort_map', 'sort_combine', 'sort_reduce', 'sort_output'):
+        assert getattr(wc, attr) is False
+    for attr in ('chunksize', 'map_chunksize', 'reduce_chunksize', 'sort_chunksize'):
+        assert getattr(wc, attr) == 10
+
+    # Can't really check key order here, so we're just going to
+    assert dict(wc(text)) == {
+        'key2': ('data2', 'data1'),
+        'key3': ('data2', 'data1'),
+        'key1': ('data2', 'data1')}
+
+
+class _WCParallelSort(mr.memory.MapReduce):
+
+    """
+    Define out here so we can pickle it in multiprocessing
+    """
+
+    # Make sure everything gets sent to a single map + combine
+    chunksize = 10
+    jobs = 4
+
+    def mapper(self, item):
+        yield item.split()
+
+    def combiner(self, key, values):
+
+        d1, d2 = list(values)
+        assert [d1, d2] == ['data1', 'data2']
+
+        yield key, 'sort2', d2
+        yield key, 'sort1', d1
+
+    def reducer(self, key, values):
+
+        d1, d2 = list(values)
+        assert [d1, d2] == ['data1', 'data2']
+
+        yield key, 'sort2', d2
+        yield key, 'sort1', d1
+
+
+def test_MapReduce_parallel_sort():
+
+    """
+    Process in parallel with sorting.
+    """
+
+    text = [
+        'key2 sort2 data2',
+        'key2 sort1 data1',
+        'key3 sort2 data2',
+        'key3 sort1 data1',
+        'key1 sort2 data2',
+        'key1 sort1 data1'
+    ]
+
+    wc = _WCParallelSort()
+
+    for attr in ('jobs', 'map_jobs', 'sort_jobs', 'reduce_jobs'):
+        assert getattr(wc, attr) == 4
+    for attr in ('sort', 'sort_map', 'sort_combine', 'sort_reduce', 'sort_output'):
+        assert getattr(wc, attr)
+    for attr in ('chunksize', 'map_chunksize', 'reduce_chunksize', 'sort_chunksize'):
+        assert getattr(wc, attr) == 10
+
+    assert tuple(wc(text)) == (
+        ('key1', ('data1', 'data2')),
+        ('key2', ('data1', 'data2')),
+        ('key3', ('data1', 'data2')))
diff --git a/tests/test_tools.py b/tests/test_tools.py
index 7d17b63..4b66402 100644
--- a/tests/test_tools.py
+++ b/tests/test_tools.py
@@ -3,13 +3,14 @@ Unittests for tinymr.tools
 """
 
 
-import pickle
+from collections import defaultdict
 from multiprocessing.pool import IMapUnorderedIterator
 from types import GeneratorType
 
 import pytest
 import six
 
+from tinymr import errors
 from tinymr import tools
 
 
@@ -45,104 +46,179 @@ def _func(v):
     return v + 1
 
 
-def test_runner():
+def test_runner_1job():
 
     input = list(range(10))
     expected = tuple(i + 1 for i in input)
 
     j1 = tools.runner(_func, input, 1)
-    assert isinstance(j1, GeneratorType)
+    assert isinstance(j1, tools.runner)
+    assert isinstance(iter(j1), GeneratorType)
     assert tuple(j1) == expected
 
-    j2 = tools.runner(_func, input, 2)
-    assert isinstance(j2, IMapUnorderedIterator)
-    assert tuple(sorted(j2)) == expected
 
+def test_runner_2job():
+
+    input = list(range(10))
+    expected = tuple(i + 1 for i in input)
+
+    # Also tests context manager
+    with tools.runner(_func, input, 2) as j2:
+        assert not j2._closed
+        assert isinstance(j2, tools.runner)
+        assert isinstance(iter(j2), IMapUnorderedIterator)
+        assert tuple(sorted(j2)) == expected
+    assert j2._closed
+
+
+def test_runner_next():
+
+    input = list(range(10))
+    expected = list(i + 1 for i in input)
+
+    r = tools.runner(_func, input, 1)
+    assert next(r) == _func(input[0])
+
+    # Multiple jobs - have to pretty much run the whole thing and sort to compare
+    results = []
+    with tools.runner(_func, input, 2) as proc:
+        for i in input:
+            results.append(next(proc))
+
+    assert sorted(results) == expected
+
+
+def test_runner_attrs_and_exceptions():
+
+    # repr
+    r = tools.runner(_func, range(10), 2)
+    assert repr(r).startswith(r.__class__.__name__)
+    assert 'jobs=2' in repr(r)
+    assert 'iterable={}'.format(repr(range(10))) in repr(r)
+
+    # Bad values
     with pytest.raises(ValueError):
         tools.runner(None, None, -1)
 
 
-class TestDefaultUnorderedDict:
+def test_mapkey():
+
+    actual = tools.mapkey('key', range(5))
+    expected = [('key', 0), ('key', 1), ('key', 2), ('key', 3), ('key', 4)]
+
+    assert not isinstance(actual, (list, tuple))  # Make sure we get an iterator
+    assert list(actual) == expected
 
-    def setup_method(self, method):
-        self.d = tools.DefaultOrderedDict(list)
 
-    def test_repr(self):
-        assert self.d.__class__.__name__ in repr(self.d)
+def test_sorter():
 
-    def test_present_key(self):
-        self.d['key'] = 'value'
-        assert self.d['key'] == 'value'
+    items = [1, 6, 3, 5, 9, 10]
+    assert sorted(items) == tools.sorter(items)
 
-    def test_missing_key(self):
-        assert self.d['missing'] == []
 
-    def test_bool_true(self):
-        self.d[None] = 'word'
-        assert self.d
+# Python 2 isn't very forgiving when it comes to sorting.
+# Make sure a useful error is raised for unorderable types
+if six.PY3:
+    def test_sorter_unorderable():
+        # Unorderable types
+        with pytest.raises(errors.UnorderableKeys):
+            tools.sorter(['2', 1])
 
-    def test_bool_false(self):
-        assert not self.d
 
-    def test_not_present(self):
-        d = tools
+def test_sorter_exceptions():
 
-    def test_exceptions(self):
-        with pytest.raises(TypeError):
-            tools.DefaultOrderedDict(None)
+    if not six.PY2:
+        with pytest.raises(errors.UnorderableKeys):
+            tools.sorter(['1', 1])
 
-    def test_copy(self):
-        self.d['key1'] = 'v1'
-        self.d['key2'] = 'v2'
-        c = self.d.copy()
-        assert isinstance(c, tools.DefaultOrderedDict)
-        assert self.d['key1'] == 'v1'
-        assert self.d['key2'] == 'v2'
-        self.d['key1'] = None
-        assert c['key1'] == 'v1'
-        assert len(c) == 2
-        assert list(c.keys()) == ['key1', 'key2']
-        assert list(c.values()) == ['v1', 'v2']
-        assert c.default_factory is list
+    def _k(v):
+        raise TypeError('bad')
 
-    def test_sorted_keys(self):
+    with pytest.raises(TypeError):
+        tools.sorter([2, 1], key=_k)
 
-        """
-        Verify that keys maintain their insert position.
-        """
 
-        # Set values
-        it = list(range(10))
-        for i in it:
-            self.d[i] = i + 1
+def test_Orderable():
 
-        # Check values
-        for k, v in self.d.items():
-            assert k + 1 == v
+    on = tools.Orderable(None)
+    for v in (-1, 0, 1):
+        assert on < v
+        assert on <= v
+        assert not on > v
+        assert not on >= v
+        assert on != v
+        assert on.obj is None
 
-        # Check sorting
-        assert list(self.d.keys()) == it
-        assert sorted(self.d.keys()) == it
+    on = tools.Orderable(None, lt=False, le=False, gt=True, ge=True)
+    for v in (-1, 0, 1):
+        assert on > v
+        assert on >= v
+        assert not on < v
+        assert not on <= v
+        assert on != v
+        assert on.obj is None
 
-    def test_unsorted_keys(self):
+    # Actually perform equality test
+    on = tools.Orderable(None, eq=None)
+    assert on == on
+    assert not on is False
+    assert not on == 67
 
-        """
-        Verify that unsorted keys remain the the same unsorted order.
-        """
+    # Never equal to a type
+    on = tools.Orderable(None, eq=False)
+    assert not on == on
+    assert not on == on
+    assert not on == 'True'
+    assert not on == 21
 
-        for i in range(5):
-            self.d[i] = i + 1
-        for i in reversed(range(30, 35)):
-            self.d[i] = i + 1
+    # Always equal to any type
+    on = tools.Orderable(None, eq=True)
+    assert on == on
+    assert on == 'False'
+    assert on == 10
 
-        assert list(self.d.keys()) == [0, 1, 2, 3, 4, 34, 33, 32, 31, 30]
-        assert len(self.d.keys()) == 10
 
+def test_OrderableNone():
 
-def test_mapkey():
+    assert isinstance(tools.OrderableNone, tools._OrderableNone)
+    assert tools.OrderableNone.obj is None
 
-    actual = tools.mapkey('key', range(5))
-    expected = [('key', 0), ('key', 1), ('key', 2), ('key', 3), ('key', 4)]
 
-    assert not isinstance(actual, (list, tuple))  # Make sure we get an iterator
-    assert list(actual) == expected
+def test_partition():
+
+    data = [
+        (1, 2),
+        (1, 1),
+        (2, 1),
+        (3, 1),
+        ('ptn', 'sort', 'data')]
+
+    expected = {
+        1: [(2,), (1,)],
+        2: [(1,)],
+        3: [(1,)],
+        'ptn': [('sort', 'data')]}
+
+    ptn = tools.partition(data)
+
+    assert isinstance(ptn, dict)
+    assert not isinstance(ptn, defaultdict)
+    assert ptn == expected
+
+
+def test_merge_partitions():
+
+    dptn = {
+        1: [(1, 2), (1, 1)],
+        2: [(2, 1)],
+        3: [(3, 1)],
+        'ptn': [('ptn', 'sort', 'data')]}
+
+    expected = {
+        1: [(1, 2), (1, 1), (1, 2), (1, 1)],
+        2: [(2, 1), (2, 1)],
+        3: [(3, 1), (3, 1)],
+        'ptn': [('ptn', 'sort', 'data'), ('ptn', 'sort', 'data')]}
+
+    actual = tools.merge_partitions(dptn, dptn)
+    assert expected == actual

pip install -e .[dev]
parse_log_pytest
pytest
pytest
3.5
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
certifi==2021.5.30
charset-normalizer==2.0.12
coverage==6.2
coveralls==3.3.1
docopt==0.6.2
idna==3.10
importlib-metadata @ file:///tmp/build/80754af9/importlib-metadata_1631916693255/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
more-itertools @ file:///tmp/build/80754af9/more-itertools_1637733554872/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pluggy @ file:///tmp/build/80754af9/pluggy_1615976315926/work
py @ file:///opt/conda/conda-bld/py_1644396412707/work
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
pytest==6.2.4
pytest-cov==4.0.0
requests==2.27.1
six==1.17.0
-e git+https://github.com/geowurster/tinymr.git@a387cf72cfc2a18978b77058e1e28f532258ae49#egg=tinymr
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
tomli==1.2.3
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3==1.26.20
zipp @ file:///tmp/build/80754af9/zipp_1633618647012/work

name: tinymr
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - attrs=21.4.0=pyhd3eb1b0_0
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - importlib-metadata=4.8.1=py36h06a4308_0
  - importlib_metadata=4.8.1=hd3eb1b0_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - more-itertools=8.12.0=pyhd3eb1b0_0
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pip=21.2.2=py36h06a4308_0
  - pluggy=0.13.1=py36h06a4308_0
  - py=1.11.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pytest=6.2.4=py36h06a4308_2
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zipp=3.6.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - charset-normalizer==2.0.12
      - coverage==6.2
      - coveralls==3.3.1
      - docopt==0.6.2
      - idna==3.10
      - pytest-cov==4.0.0
      - requests==2.27.1
      - six==1.17.0
      - tomli==1.2.3
      - urllib3==1.26.20
prefix: /opt/conda/envs/tinymr

tests/test_base.py::test_not_implemented_methods tests/test_base.py::test_default_settings tests/test_base.py::test_default_methods tests/test_memory.py::test_MapReduce_init_reduce tests/test_memory.py::test_MapReduce_sort tests/test_memory.py::test_MapReduce_no_sort tests/test_memory.py::test_MapReduce_parallel_sort tests/test_tools.py::test_slicer_even tests/test_tools.py::test_slicer_odd tests/test_tools.py::test_runner_1job tests/test_tools.py::test_runner_2job tests/test_tools.py::test_runner_next tests/test_tools.py::test_runner_attrs_and_exceptions tests/test_tools.py::test_mapkey tests/test_tools.py::test_sorter tests/test_tools.py::test_Orderable tests/test_tools.py::test_OrderableNone tests/test_tools.py::test_partition tests/test_tools.py::test_merge_partitions
tests/test_tools.py::test_sorter_unorderable tests/test_tools.py::test_sorter_exceptions
README.rst tinymr/errors.py tinymr/base.py tinymr/tools.py tinymr/memory.py tinymr/__init__.py .coveragerc tinymr/_mrtools.py tinymr/_backport_heapq.py
README.rst tinymr/errors.py tinymr/base.py tinymr/tools.py tinymr/memory.py tinymr/__init__.py .coveragerc tinymr/_mrtools.py tinymr/_backport_heapq.py