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


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


  [& args]
  (let [p (make-pipeline args)
        final (->> p
                   (generate-input [{:key :a :val 10} {:key :b :val 5} {:key :a :val 42}] {:name "gengen"})
                   (with-keys :key {:name :group-by-key})
                   (dmap (fn [kv] (vector (.getKey kv) (.getValue kv))) )
                   (to-edn {:name "edn"})
                   (write-text-file "tee"))]

