 
 (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/create options)))
+        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]

  [& 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"))]


(ns datasplash.core
  (:require [taoensso.nippy :as nippy :refer [thaw freeze]]
            [cognitect.transit :as transit])
  (:import [datasplash.fns ClojureDoFn ClojureSerializableFn]
           [java.io InputStream OutputStream]
           [java.util UUID]
           [com.google.cloud.dataflow.sdk.options PipelineOptionsFactory]
           [com.google.cloud.dataflow.sdk Pipeline]
           [com.google.cloud.dataflow.sdk.io TextIO$Read TextIO$Write]
           [com.google.cloud.dataflow.sdk.transforms
            DoFn DoFn$Context ParDo DoFnTester Create
            SerializableFunction WithKeys]
           [com.google.cloud.dataflow.sdk.values PCollection]
           [com.google.cloud.dataflow.sdk.coders ByteArrayCoder StringUtf8Coder CustomCoder Coder$Context]
           [datasplash.vals ClojureVal]
