    {:side-inputs {:docstr "Adds a map of PCollectionViews as side inputs to the underlying ParDo Transform. They can be accessed there by key in the datasplash.dv/*side-inputs* var."
                   :action (fn [transform inputs]
                             (.withSideInputs transform (map val (sort-by key inputs))))}
-    :without-coercion-to-clj {:docstr "Avoids coercing Dataflow types to Clojure, like KV. Coercion will happe, by default"}}))
+    :without-coercion-to-clj {:docstr "Avoids coercing Dataflow types to Clojure, like KV. Coercion will happen by default"}}))
 
 (defn map-op
   ([transform label coder]

  ([specs join-fn] (join-by {} specs join-fn)))

(defn ddistinct
  ([options ^PCollection pcoll]
   (let [opts (assoc options :label :distinct)]
     (-> pcoll

             (from-edn options))))))

(defn read-edn-file
  {:doc (with-opts-docstr "Reads a PCollection of edn strings from disk or Google Storage, with records separated by newlines.
See https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/TextIO.Read.html

  Example:
    (read-edn-file \"gs://target/path\" pcoll)"
          base-schema text-reader-schema)
   :added "0.1.0"}
  ([from options ^Pipeline p]
   (let [opts (assoc options :label (str "read-edn-file-from-"
                                         (clean-filename from)))]
     (-> p
         (.apply (with-opts (merge base-schema text-reader-schema) opts
                   (read-edn-file-transform from opts))))))
  ([from p] (read-edn-file from {} p)))

