 
 (defmacro with-opts
   [schema opts & body]
-  (println opts)
   (let [full-name (name
                    (or (:name opts)
                        (do

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



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

(defn sfn
  ^SerializableFunction [f]
  (ClojureSerializableFn. f))

(defn with-keys
  ([f options ^PCollection pcoll]
   (let [opts (assoc options :label :with-keys)]
     (-> pcoll
         (.apply (with-opts base-schema opts
                   (WithKeys/of (sfn f)))))))
  ([f pcoll] (with-keys f {} pcoll)))

(defn make-pipeline
  [str-args]
  (let [builder (PipelineOptionsFactory/fromArgs
                 (into-array String str-args))
        options (.create builder)
        pipeline (Pipeline/create options)
        coder-registry (.getCoderRegistry pipeline)]
    (doto coder-registry
      (.registerCoder clojure.lang.IPersistentCollection (make-transit-coder))
      (.registerCoder clojure.lang.Keyword (make-transit-coder)))
    pipeline))

(defn load-text-file
  [from options ^Pipeline p]
