 
 (defn cogroup-transform
   ([options]
-   (let [opts (-> options
-                  (assoc :label :raw-cogroup)
-                  (dissoc :name))]
+   (let [opts (assoc options :label :raw-cogroup)]
      (proxy [PTransform] []
        (apply [^KeyedPCollectionTuple pcolltuple]
          (let [ordered-tags (->> pcolltuple

(defn -main
  [& args]
  (let [p (make-pipeline args)
        first-in (->> p
                      (generate-input [{:key :a :foo 12} {:key :b :foo 5} {:key :a :foo 42} {:key :c :val "never"}] {:name "gengen1"}))
        second-in (->> p
                       (generate-input [{:key :a :val 10} {:key :b :val 5} {:key :a :val 42}] {:name "gengen"}))
        final (->> (cogroup-by {:name :join} [[first-in :key] [second-in :key :required]] (fn [[k res]] (apply merge res)))
                   (write-edn-file "tessst.edn" {:name :output-edn}))
        ;; first-in (->> p
        ;;               (generate-input [{:key :a :foo 12} {:key :b :foo 5} {:key :a :foo 42}] {:name "gengen1"})
        ;;               (dgroup-by :key {:name :group-by-fi}))

    (.output c result)))

(def to-edn (partial (map-op identity :to-edn (StringUtf8Coder/of)) to-edn*))
(def from-edn (partial dmap #(edn/read-string %)))

(defn sfn
  ^SerializableFunction [f]
