service.clj

(def service
  (-> (new-service)
      (input-topic "events")
      (string-starts-with? [:context :url] "https://pyroclast.io")
      (f/= :event :page-view)
      (parse-datetime :timestamp "YYY-MM-dd'T'HH:mm:ss.SSSZ")
      (aggregate-together
       [{:name "hourly-page-views"
         :aggregate (a/count (fixed-windows-of 1 :hour :timestamp))}]
       :user-id)))