casc-cb-abv.clj

(ns cascacb.test
  (:use cascalog.api)
  (:require [cascalog.ops :as c]
            [cbdrawer.client :as cb]
            [cbdrawer.transcoders :refer [json-transcoder]]
            [cascacb.tap :refer [cbtap]]))

;; Enhance the "beer-sample" dataset that comes with couchbase.
(defn update-abv-averages []
  (cb/set-transcoder! json-transcoder)
  (let [cli (cb/client "beer-sample" "" "http://127.0.0.1:8091")
        in (cbtap "beer-sample" "" "http://127.0.0.1:8091")
        ;;Find all docs with type "beer" and abv > 1.0,
        ;;group on brewery_id and return the average abv
        ;;of each group
        brewery-abv-avg (<- [?brewery ?avgabv]
                            (in _ ?doc)
                            (get ?doc :type :> "beer")
                            (get ?doc :brewery_id :> ?brewery)
                            (get ?doc :abv :> ?abv)
                            (< 1.0 ?abv)
                            (c/avg ?abv :> ?avgabv))
        top-brewery-abvs (<- [?brewery ?avgabv]
                             (brewery-abv-avg ?brew0 ?avg0)
                             (:sort ?avg0) (:reverse true)
                             (c/limit [10] ?brew0 ?avg0 :> ?brewery ?avgabv))]
    ;;Update each brewery doc with its abv
    (doseq [[brewery_id abv] (first (??- brewery-abv-avg))]
      (cb/cas! cli brewery_id assoc :average_abv abv))
    ;;Update a document listing the top 10 breweries by abv
    (cb/force! cli :top_brewery_abvs
              {:type "generated"
               :top_breweries (first (??- top-brewery-abvs))})
    ;;Give the client some time to shut down
    (cb/shutdown cli 10 java.util.concurrent.TimeUnit/SECONDS)))