          (.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]

           [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

      (.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)]
