         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})
+                   (group-by-key)
                    (dmap (fn [kv] (vector (.getKey kv) (.getValue kv))) )
                    (to-edn {:name "edn"})
                    (write-text-file "tee"))]

           [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]

                   (WithKeys/of (sfn f)))))))
  ([f pcoll] (with-keys f {} pcoll)))

(defn group-by-key
  ([options ^PCollection pcoll]
   (let [opts (assoc options :label :group-by-keys)]
     (-> pcoll
         (.apply (GroupByKey/create)))))
  ([pcoll] (group-by-key {} pcoll)))

(defn make-pipeline
  [str-args]
  (let [builder (PipelineOptionsFactory/fromArgs
