 (def dfilter (map-op filter-fn))
 
 (defn generate-input
-  [coll p]
-  (-> p
-      (.apply (Create/of (seq coll)))
-      (.setCoder (make-transit-coder))))
+  [coll opts ^Pipeline p]
+  (with-opts base-schema opts
+    (-> p
+        (.apply (Create/of (seq coll)))
+        (.setCoder (make-transit-coder)))))
 
 (defn- to-edn*
   [^DoFn$Context c]

(def to-edn (partial (map-op identity (StringUtf8Coder/of)) to-edn*))

(defn to-edn
  [^PCollection pcoll]
  (-> pcoll
      (.apply (ParDo/of (dofn to-edn*)))
      (.setCoder (StringUtf8Coder/of))))

(defn make-pipeline
  [str-args]

  [& 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"})
                   (write-text-file "tee"))]

    (.run p)))
