 (defn commands->flow
   "Transforms a series of commands into a Cascading flow"
   [commands]
-  (let [flowdef (reduce (fn [def cmd] (command->flowdef cmd def)) {} commands)]
+  (let [flowdef (reduce (fn [def cmd] (command->flowdef cmd def)) {} commands)
+        {:keys [pipe-to-source sources pipe-to-sink sinks pipes]} flowdef
+        sources-map (into {} (map (fn [[p s]] [(str p) (sources s)]) pipe-to-source))
+        sinks-map (into {} (map (fn [[p s]] [(str p) (sinks s)]) pipe-to-sink))
+        tail-pipes (into-array Pipe (map #(pipes %) (keys pipe-to-sink)))]
+
+    (println sources-map)
+    (println sinks-map)
+    (println tail-pipes)
+    (.connect (HadoopFlowConnector.) sources-map sinks-map tail-pipes)
     flowdef))
 

  (if (contains? (:sources flowdef) (first ancestors))
    (let [pipe (Pipe. (str id))]
      (-> flowdef
          (update-in [:source-to-pipe] (partial merge {(first ancestors) pipe}))
          (update-in [:pipes] (partial merge {id pipe}))))
    (throw (Exception. "not implemented"))))


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

(defmethod command->flowdef :generate
           [{:keys [id ancestors projections opts]} flowdef]
