   [context] (src/reducer src/next-key src/keykeyvals context))
 
 (defn wrap-sink
-  "Return new tuple sink which wraps keys and values as the types
-`ckey` and `cval` respectively, which should be compatible with the
-key and value type of `sink`.  Where they are not compatible, the type
-of the `sink` will be used instead.  Returns a new tuple sink which
-wraps any sunk keys and values which are not already of the correct
-type then sinks them to `sink`."
+  "Return new tuple sink which wraps keys and values as the types `ckey` and
+`cval` respectively, which should be compatible with the key and value type of
+`sink`.  Where they are not compatible, the type of the `sink` will be used
+instead.  Returns a new tuple sink which wraps any sunk keys and values which
+are not already of the correct type then sinks them to `sink`."
   ([sink] (snk/wrap-sink nil nil sink))
   ([ckey cval sink] (snk/wrap-sink ckey cval sink)))
 
 (defn sink-as
-  "Return new tuple collection which has values sinked as `kind`,
-which may be one of `:keys`, `:vals`, or `:keyvals`."
+  "Return new tuple collection which has values sinked as `kind`, which may be
+one of `:keys`, `:vals`, or `:keyvals`."
   [kind sink] (vary-meta sink assoc ::snk/tuples-as kind))
 
 (defn sink

  TaskAttemptID (taid [taid] taid))

(defn tac
  "Return a new TaskAttemptContext instance using provided
configuration `conf` and task attempt ID `taid`."
  {:tag `TaskAttemptContext}
  [conf id] (tac* (conf/ig conf) (taid id)))

    (Class/forName (format "parkour.hadoop.Reducers$_%d" i))))

(defn reducer!
  "Allocate and return a new parkour reducer class for `job` as invoking `var`.
The `var` will be called during task-setup with the job Configuration and any
provided `args` (which must be EDN-serializable).  It should return a function
of one argument, which will be invoked with the task context, and should perform
the desired content of the reduce task."
  [^Job job var & args]
  (apply reducer!* :reduce job var args))

(defn combiner!
  "As per `reducer!`, but allocate and configure for the Hadoop combine step,
which may impact e.g. output types."
  [^Job job var & args]
  (apply reducer!* :combine job var args))

(defn partitioner!
  "Allocate and return a new parkour partitioner class for `job` as invoking
`var`.  The `var` will be called during task-setup with the job Configuration
and any provided `args` (which must be EDN-serializable).  It should return a
function of three arguments: a map-output key, a map-output value, and an
integral reduce-task count.  That function will called for each map-output
tuple, and should return an integral value mod the reduce-task count.  Should be
primitive-hinted as OOLL."
  [^Job job var & args]
  (conf/assoc! job
