                    (RemoveDuplicates/create))))))
   ([pcoll] (ddistinct {} pcoll)))
 
+(def scoped-ops-schema
+  {:scope {:docstr "Scope given to the combinating operation. One of (:globally :per-key)."}})
+
 (defn sample
+  {:doc (with-opts-docstr
+          "Takes samples of the elements in a PCollection, or samples of the values associated with each key in a PCollection of KVs.
+See https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/Sample.html
+
+  Example:
+    (ds/sample 100 {:scope :per-key} pcoll)"
+          base-schema scoped-ops-schema)
+   :added "0.1.0"}
   ([size {:keys [scope] :as options} ^PCollection pcoll]
    (let [opts (assoc options :label (keyword "sample-" (if scope (name scope) "any")))]
      (-> pcoll

  ([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)))

