   [& args]
   (let [p (make-pipeline args)
         final (->> p
-                   (generate-input (range 10000) {:named "gengen"})
-                   (dmap inc {:named "map"})
-                   (dfilter even? {:named "filter"})
-                   (to-edn {:named "edn"})
+                   (generate-input (range 10000) {:name "gengen"})
+                   (dmap inc {:name "map"})
+                   (dfilter even? {:name "filter"})
+                   (to-edn {:name "edn"})
                    (write-text-file "tee"))]
 
     (.run p)))


(defmacro with-opts
  [schema opts & body]
  `(let [transform# (do ~@body)]
     (reduce
      (fn [f# [k# apply#]]
        (if-let [v# (get ~opts k#)]
          (apply# f# v#)
          f#))
      transform# ~schema)))

(def base-schema
  {:name (fn [transform n] (.withName transform n))})

(defn map-op
  ([transform coder]
   (fn
     [f opts ^PCollection pcoll]
     (-> pcoll
         (.apply (with-opts base-schema opts
                   (ParDo/of (dofn (transform f)))))
         (.setCoder (or (:coder opts) coder)))))
  ([transform]
   (map-op transform (make-transit-coder))))

(def dmap (map-op map-fn))
(def dmapcat (map-op mapcat-fn))
(def dfilter (map-op filter-fn))

(defn generate-input
  [coll opts ^Pipeline p]
  (-> p
      (.apply (with-opts base-schema opts
                (Create/of (seq coll))))
      (.setCoder (or (:coder opts) (make-transit-coder)))))

(defn- to-edn*
  [^DoFn$Context c]

            [cognitect.transit :as transit])
  (:import [datasplash.fns ClojureDoFn]
           [java.io InputStream OutputStream]
           [java.util UUID]
           [com.google.cloud.dataflow.sdk.options PipelineOptionsFactory]
           [com.google.cloud.dataflow.sdk Pipeline]
           [com.google.cloud.dataflow.sdk.io TextIO$Read TextIO$Write]
