 (ds/cogroup [{:type :required} {:type :optional}] {:name :my-cogroup} [pcoll1 pcoll2])
 
 ```
-
 See https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/join/CoGroupByKey and for a more idiomatic approach to joins [[cogroup-by]] and [[join-by]]"
-          base-schema)
+          named-schema)
    :added "0.1.0"}
   ([options specs reduce-fn]
-   (let [safe-opts (dissoc options :name)
-         pcolls (for [[idx [pcoll f {:keys [drop-nil?]}]] (map-indexed vector specs)]
-                  (let [op (with-keys f (assoc safe-opts :name (str "group-by-" idx)) pcoll)]
-                    (if drop-nil?
-                      (dfilter (fn [^KV kv] (not (nil? (.getKey kv))))
-                               {:name (str "drop-nil-" idx)
-                                :without-coercion-to-clj true}
-                               op)
-                      op)))
-         safe-specs (into []
-                          (doall
-                           (for [[pcoll key-fn spec] specs]
-                             (cond
-                               (nil? spec) {:type :optional}
-                               (keyword? spec) {:type spec}
-                               (map? spec) spec
-                               :else (throw (ex-info "Invalid spec for cogroup" {:specs specs}))))))
-         grouped-coll (cogroup safe-specs safe-opts pcolls)]
-     (if reduce-fn
-       (dmap reduce-fn safe-opts grouped-coll)
-       grouped-coll)))
+   (let [group-specs (make-group-specs specs)
+         cogroup-by-tr (cogroup-by-transform options reduce-fn) ]
+     (apply-transform group-specs cogroup-by-tr named-schema options)))
   ([options specs] (cogroup-by options specs nil)))
 
 (defn join-by
   ([options specs join-fn]
-   (->> (cogroup-by options specs)
-        (dmapcat (fn [^KV kv]
-                   (let [results (.getValue kv)
-                         results-ok (map #(if (empty? %) [nil] %) results)
-                         raw-res (apply combo/cartesian-product results-ok)
-                         res (map (fn [prod] (apply join-fn prod)) raw-res)]
-                     res)) {:without-coercion-to-clj true })))
+   (let [nam (or (:name options) "join-by")]
+     (pt->>
+      (or (:name options) "join-by")
+      (cogroup-by (assoc options :name (str nam "-cogroup-by")) specs)
+      (dmapcat (fn [^KV kv]
+                 (let [results (.getValue kv)
+                       results-ok (map #(if (empty? %) [nil] %) results)
+                       raw-res (apply combo/cartesian-product results-ok)
+                       res (map (fn [prod] (apply join-fn prod)) raw-res)]
+                   res))
+               {:name (let [root-name (or (:name options) "join-by")]
+                        (str root-name "-cartesian-product"))
+                :without-coercion-to-clj true }))))
   ([specs join-fn] (join-by {} specs join-fn)))
 
 (defn ddistinct

          base-schema text-writer-schema)
   :added "0.1.0"}
  ([to options ^PCollection pcoll]
   (let [opts (assoc options :label (str "write-edn-file-to-" (clean-filename to))
                     :coder nil)]
     (apply-transform pcoll (write-edn-file-transform to opts)
                      named-schema opts)))
  ([to pcoll] (write-edn-file to {} pcoll)))

(def json-reader-schema
  {:key-fn {:docstr "Selects a policy for parsing map keys. If true, keywordizes the keys. If given a fn, uses it to transform each map key."}
   :return-type {:docstr "Allows passing in a function to specify what kind of types to return."}})

(defn- read-json-file-transform
  [from {:keys [key-fn return-type] :as options}]
  (let [safe-opts (dissoc options :name :key-fn :return-type)]
    (ptransform
     :read-json-file
     [p]
     (->> p
          (read-text-file from (dissoc safe-opts :coder))
          (dmap (fn [l]
                  (cond
                    (and key-fn return-type) (json/decode l key-fn return-type)
                    key-fn (json/decode l key-fn)
                    return-type (json/decode l nil return-type)
                    :else (json/decode l)))
                safe-opts)))))

(defn read-json-file
  {:doc (with-opts-docstr "Reads a PCollection of JSON strings from disk or Google Storage, with records separated by newlines.


                          (merge named-schema text-reader-schema) opts))))
  ([from p] (read-text-file from {} p)))

(defn read-edn-file
  {:doc (with-opts-docstr "Reads a PCollection of edn strings from disk or Google Storage, with records separated by newlines.

