 (defn run-standard-sql-query
   [str-args]
   ;; the DirectPipelineRunner doesn't support standardSql yet
-  (let [p (ds/make-pipeline StandardSQLOptions str-args {:runner "DataflowPipelineRunner"})
+  (let [p (ds/make-pipeline StandardSQLOptions str-args {:runner "DataflowRunner"})
         {:keys [input output]} (ds/get-pipeline-options p)
         query "SELECT * from `bigquery-public-data.samples.shakespeare` LIMIT 100"
         results (->> p

;; Pub/Sub ;;
;;;;;;;;;;;;;

;; Run using: lein run pub-sub --project=[your google cloud project] --stagingLocation=gs://[your-bucket]/jars
;; You must create the my-subscription and my-transformed-subscription subscriptions, and the my-transformed-topic topics
;; before you run this

(defoptions PubSubOptions
  {:project {:default "yourproject"
             :type String}
   :stagingLocation {:default "gs://yourbucket"
                     :type String}})

(defn stream-interactions-from-pubsub
 [pipeline read-topic write-transformed-topic]

(defn run-pub-sub
  [str-args]
  (let [pipeline (ds/make-pipeline
                  PubSubOptions
                  str-args
                  {:runner "DataflowRunner"
                   :streaming true})
        {:keys [pubsubProject]} (ds/get-pipeline-options pipeline)
        read-topic (format "projects/%s/topics/my-topic" pubsubProject)
        write-transformed-topic (format "projects/%s/topics/my-transformed-topic" pubsubProject)
        read-transformed-subscription (format "projects/%s/subscriptions/my-transformed-subscription" pubsubProject)]
    (stream-interactions-from-pubsub pipeline read-topic write-transformed-topic)
    (stream-forwarded-interactions-from-pubsub pipeline read-transformed-subscription)))

