 
 (def ^:private stage
   "The stage of a job graph node."
-  (comp :stage pr/arg0))
+  (fn [node & args]
+    (cond
+     (vector? node) ::vector
+     (map? node) (:stage node)
+     :else (throw (ex-info "Invalid `node`" {:node node})))))
 
 (let [id (atom 0)]
   (defn ^:private gen-id


(def remote nil)
(defmulti remote
  "Create a new remote-execution task chained following the provided
job graph `node` and implemented by function `f`, which should accept
and return reducible collections of task tuples.  When producing a
`:map`-stage node, may provide a combiner function `g`, will replace
any existing job combiner function."
  {:arglists '([node f] [node f g])}
  stage)

(defmethod remote :default
  [node f]
  (let [stage (:stage node)
        msg (str "Cannot change remote-execution task from node of stage `"
                 stage "`.")]
    (throw (ex-info msg {:node node, :f f}))))

(defmethod remote :source
  ([node f] (assoc node :stage :map, :mapper (f-list f)))

            [parkour.io.mux :as mux]
            [parkour.util
             :refer [ignore-errors returning var-str mpartial mcomp]])
  (:import [org.apache.hadoop.mapreduce Job]
           [org.apache.hadoop.mapreduce.lib.partition HashPartitioner]))

(comment
