 
 (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))))))
+   (fn make-map-op
+     ([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)))))
+     ([f pcoll] (make-map-op f {} pcoll))))
   ([transform label]
    (map-op transform label (make-transit-coder))))
 


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

(defn make-pipeline
  [str-args]
  (let [builder (PipelineOptionsFactory/fromArgs
                 (into-array String str-args))
        options (.create builder)]
    (Pipeline/create options)))

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


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