     (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."
+  "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."
+  "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
+  "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

  [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`."
  ([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`."
  [kind sink] (vary-meta sink assoc ::snk/tuples-as kind))

(defn sink

  (->> Job reflect/type-reflect :members (some #(= 'getInstance (:name %)))))

(defmacro ^:private make-job
  "Macro to create a new `Job` instance, using Hadoop version appropriate
mechanism."
  [& args] `(~(if job-factory-method? `Job/getInstance `Job.) ~@args))

(defn job
  "Return new Hadoop `Job` instance, optionally initialized with configuration
`conf`."
  {:tag `Job}
  ([] (make-job (conf/ig)))
  ([conf] (make-job (conf/clone conf))))

(defn mapper!
  "Allocate and return a new parkour mapper 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 map task."
  [^Job job var & args]
  (let [i (conf/get-int job "parkour.mapper.next" 0)]
    (conf/assoc! job
