             [taoensso.nippy :refer [freeze thaw]]
             [clojure.pprint]))
 
+(defn- add-val [flowdef path id val]
+  (update-in flowdef path (partial merge {id val})))
+
+(defn- cfields [fields]
+  (Fields. (into-array (map str fields))))
+
+(defn- cascading-field [name-or-number]
+  (if (number? name-or-number)
+    (int name-or-number)
+    (str name-or-number)))
+
 (defmulti get-tap-fn
           identity)
 

          "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]

      (add-val [:pipe-to-sink] (first ancestors) id)
      (add-val [:sinks] id ((get-tap-fn storage) location opts))))

(defmethod command->flowdef :code
  [{:keys [expr args pipe field-projections]} flowdef]
  {:pre [expr args]}
  (let [{:keys [init func]} expr
        fields (if field-projections
                 (cfields (map #(cascading-field (:alias %)) field-projections))
                 Fields/UNKNOWN)
        cogroup-opts (get-in flowdef [:cogroup-opts pipe])]
    (if-not (nil? cogroup-opts)
