       (.registerCoder clojure.lang.Keyword (make-transit-coder)))
     pipeline))
 
+;;;;;;;;;;;;;
+;; Text IO ;;
+;;;;;;;;;;;;;
+
 (defn write-text-file
   ([to options ^PCollection pcoll]
    (let [opts (assoc options :label :write-text-file)]

           [java.util UUID]
           [com.google.cloud.dataflow.sdk.options PipelineOptionsFactory]
           [com.google.cloud.dataflow.sdk Pipeline]
           [com.google.cloud.dataflow.sdk.io TextIO$Read TextIO$Write]
           [com.google.cloud.dataflow.sdk.transforms
            DoFn DoFn$Context DoFn$ProcessContext ParDo DoFnTester Create PTransform
            SerializableFunction WithKeys GroupByKey RemoveDuplicates

         (.apply (with-opts base-schema opts
                   (write-edn-file-transform to opts))))))
  ([to pcoll] (write-edn-file to {} pcoll)))

;;;;;;;;;;;;;;;;;;
;; Big Query IO ;;
;;;;;;;;;;;;;;;;;;

(defn read-big-query-table-raw
  ([from options p]
   (let [opts (assoc options :label :read-big-query-table-raw)]
     (-> p
         (cond-> (instance? Pipeline p) (PBegin/in))
         (.apply (with-opts base-schema opts
                   (BigQueryIO$Read/from from))))))
  ([from p] (read-big-query-table-raw from {} p)))

(defn table-row->edn
  [^TableRow row]
  (let [keyset (.keySet row)]
    (persistent!
     (reduce
      (fn [acc k]
        (assoc! acc (keyword k) (.get row k)))
      (transient {}) keyset))))

(defn- read-big-query-table-transform
  [from options]
  (let [safe-opts (dissoc options :name)]
    (proxy [PTransform] []
      (apply [p]
        (->> p
             (read-big-query-table-raw from options)
             (dmap table-row->edn options))))))

(defn read-big-query-table
  ([from options ^Pipeline p]
   (let [opts (assoc options :label :read-big-query-table)]
     (-> p
         (.apply (with-opts base-schema opts
                   (read-big-query-table-transform from opts))))))
  ([from p] (read-big-query-table from {} p)))

(defn make-keyed-pcollection-tuple
  [pcolls]
