           base-schema)
    :added "0.1.0"}
   ([coll options ^Pipeline p]
-   (let [opts (assoc options :label :generate-input)
+   (let [opts (merge {:coder (make-nippy-coder)}
+                     (assoc options :label :generate-input))
          ptrans (Create/of (seq coll))]
      (apply-transform p ptrans base-schema opts)))
   ([coll p] (generate-input coll {} p)))


(defn apply-transform
  [pcoll ^PTransform transform schema
   {:keys [coder coll-name] :as options
    :or {coder (make-nippy-coder)}}]
  (let [nam (some-> options (:name) (name))
        clean-opts (dissoc options :name :coder :coll-name)
        configured-transform (with-opts schema clean-opts transform)


(def write-edn-file-by (partial write-file-by write-edn-file))
(def write-text-file-by (partial write-file-by write-text-file))

;;;;;;;;;;;
;; Joins ;;
;;;;;;;;;;;

(defn make-keyed-pcollection-tuple
  [pcolls]
  (let [empty-kpct (KeyedPCollectionTuple/empty (.getPipeline (first pcolls)))]
    (reduce
     (fn [coll-tuple [idx pcoll]]
       (let [tag (TupleTag. (str idx))
             new-coll-tuple (.and coll-tuple tag pcoll)]
         new-coll-tuple))
     empty-kpct (map-indexed (fn [idx v] [idx v]) pcolls))))

(defn cogroup-transform
  ([options]
