         pipes (map (fn [p k t] (if (= :required t)
                                  (Each. p (Fields. (into-array (map str k))) (FilterNull.))
                                  p))
-                   pipes keys join-types)
-        flowdef (update-in flowdef [:pipes] (partial merge {id (CoGroup. (str id)
-                                                                         (into-array Pipe pipes)
-                                                                         (into-array (map #(Fields. (into-array (map str %)))
-                                                                                          keys))
-                                                                         Fields/NONE
-                                                                         (BufferJoin.))}))
-        flowdef (update-in flowdef [:cogroup-opts] (partial merge {id {:group-type  :group
-                                                                       :group-all   is-group-all
-                                                                       :num-streams (count pipes)
-                                                                       :inner-join  (= #{:required} (into #{} join-types))}}))]
-    flowdef))
+                   pipes keys join-types)]
+    (-> flowdef
+        (add-val [:pipes] id (CoGroup. (str id)
+                                       (into-array Pipe pipes)
+                                       (into-array (map #(Fields. (into-array (map str %)))
+                                                        keys))
+                                       Fields/NONE
+                                       (BufferJoin.)))
+        (add-val [:cogroup-opts] id {:group-type  :group
+                                     :group-all   is-group-all
+                                     :num-streams (count pipes)
+                                     :inner-join  (= #{:required} (into #{} join-types))}))))
 
 (defmethod command->flowdef :join
   [{:keys [id keys fields join-types ancestors opts]} flowdef]

  [{:keys [id ancestors]} flowdef]
  {:pre [id (= 1 (count ancestors))]}
  (let [pipe ((:pipes flowdef) (first ancestors))]
    (update-in flowdef [:pipes] (partial merge {id (Unique. pipe Fields/ALL)}))))

(defmethod command->flowdef :limit
  [{:keys [id n ancestors]} flowdef]
  {:pre [id n (= 1 (count ancestors))]}
  (let [pipe ((:pipes flowdef) (first ancestors))]
    (update-in flowdef [:pipes] (partial merge {id (Each. pipe (Limit. n))}))))

(defmethod command->flowdef :sample
  [{:keys [id p ancestors]} flowdef]
  {:pre [id p (= 1 (count ancestors))]}
  (let [pipe ((:pipes flowdef) (first ancestors))]
    (update-in flowdef [:pipes] (partial merge {id (Each. pipe (Sample. p))}))))

(defmethod command->flowdef :script
  [_ flowdef]

          "Converts an individual command into the equivalent Cascading flow definition."
          (fn [{:keys [type]} flowdef] type))

(defn- add-val [flowdef path id val]
  (update-in flowdef path (partial merge {id val})))

(defmethod command->flowdef :load
  [{:keys [id location storage fields opts]} flowdef]
  {:pre [id location storage fields]}
  (let [pipe (Pipe. (str id))]
    (-> flowdef
        (add-val [:sources] id ((get-tap-fn storage) location opts))
        (add-val [:pipes] id pipe))))

(defmethod command->flowdef :store
  [{:keys [id ancestors location storage opts]} flowdef]
  {:pre [id ancestors location storage]}
  (-> flowdef
      (add-val [:pipe-to-sink] (first ancestors) id)
      (add-val [:sinks] id ((get-tap-fn storage) location opts))))

(defn- cascading-field [name-or-number]
  (if (number? name-or-number)
