 
 (defmethod job-fn :source
   [node conf jname]
-  (fn [& args] (-> node :source first)))
+  (constantly (:source node)))
 
 (defmethod job-fn :default
   [node conf jname]

  "Update Hadoop `job` with configuration `step`, returning the mutated job."
  pgconf/configure!)

(defn ^:private mux-step
  [step] #(mux/add-subconf % (configure! (mr/job %) step)))

(defn ^:private source-config
  "Configure a job for a provided source vector."
  [^Job job source]
  (if (= 1 (count source))
    (configure! job source)
    (configure! job (mapv mux-step source))))

(declare node-config)

(defn ^:private subnode-config

  "Configure a job according to the state of a job graph node."
  [^Job job node]
  (doto job
    (configure! (:source node))
    (subnodes-config node (:subnodes node))
    (pgt/mapper! node)
    (pgt/combiner! node)
