       (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)
-    (int name-or-number)
-    (str name-or-number)))
-
 (defmethod command->flowdef :code
   [{:keys [expr args pipe field-projections]} flowdef]
   {:pre [expr args]}
   (let [{:keys [init func]} expr
         fields (if field-projections
-                 (Fields. (into-array (map #(cascading-field (:alias %)) field-projections)))
+                 (cfields (map #(cascading-field (:alias %)) field-projections))
                  Fields/UNKNOWN)
         cogroup-opts (get-in flowdef [:cogroup-opts pipe])]
     (if-not (nil? cogroup-opts)

                                                                                     p (Pipe. (str (nth fields (* i 2))) p)
                                                                                     p (Each. p (Identity.))]
                                                                                 p)) ancestors))
                                       (into-array (map #(let [f (Fields. (into-array (map str %)))]
                                                          (when-not join-nils (.setComparator f (str (first %)) (NullNotEquivalentComparator.)))
                                                          f) keys))
                                       (Fields. (into-array (map str fields)))
                                       joiner))
        (add-val [:cogroup-opts] id {:group-type :join
                                     :all-args   (true? (:all-args opts))}))))

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


(defmethod command->flowdef :load
  [{:keys [id location storage fields opts]} flowdef]
