       (let [rdr (transit/reader in :msgpack)]
         (transit/read rdr)))))
 
+(defmacro with-opts
+  [schema opts & body]
+  `(let [pcoll# (do ~@body)]
+     (reduce
+      (fn [pc# [k# f#]]
+        (if-let [v# (get ~opts k#)]
+          (f# pc# v#)
+          pc#))
+      pcoll# ~schema)))
+
+(def base-schema
+  {:named (fn [pcoll n] (.setName pcoll n))
+   :coder (fn [pcoll coder] (.setCoder pcoll coder))})
+
 (defn map-op
   ([transform coder]
    (fn
-     [f ^PCollection pcoll]
-     (-> pcoll
-         (.apply (ParDo/of (dofn (transform f))))
-         (.setCoder coder))))
+     [f opts ^PCollection pcoll]
+     (with-opts base-schema opts
+       (-> pcoll
+           (.apply (ParDo/of (dofn (transform f))))
+           (.setCoder coder)))))
   ([transform]
    (map-op transform (make-transit-coder))))
 

(def dfilter (map-op filter-fn))

(defn generate-input
  [coll p]
  (-> p
      (.apply (Create/of (seq coll)))
      (.setCoder (make-transit-coder))))

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

        options (.create builder)]
    (Pipeline/create options)))

(defn load-text-file
  [from opts ^Pipeline p]
  (-> p
      (.apply (TextIO$Read/from from))
      (.setCoder (StringUtf8Coder/of))))

(defn write-text-file
  ([to opts ^PCollection pcoll]
   (-> pcoll
       (.apply (-> (TextIO$Write/to to)))))
  ([to pcoll] (write-text-file to {} pcoll)))

(comment
  (compile 'datasplash.core))
