            [com.google.cloud.dataflow.sdk.io TextIO$Read TextIO$Write]
            [com.google.cloud.dataflow.sdk.transforms
             DoFn DoFn$Context ParDo DoFnTester Create
-            SerializableFunction WithKeys]
+            SerializableFunction WithKeys GroupByKey]
            [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 make-pipeline
  [str-args]
  (let [builder (PipelineOptionsFactory/fromArgs

    (decode [^InputStream in ^Coder$Context context]
      (require '[cognitect.transit :as transit])
      (let [rdr (transit/reader in :msgpack)]
        (transit/read rdr)))
    (verifyDeterministic [] nil)
    (consistentWithEquals [] true)))

(defmacro with-opts
  [schema opts & body]
