Sinking to HBase

hbasesink.clj

(def sample-tweets
  [["1234567890" (str "[\"" "http://t.co/09hWanj8" "\",\"" "http://t.co/lozKUbnY" "\"]")]
   ["1029384756" (str "[\"" "http://t.co/1O8YAA21" "\",\"" "http://t.co/ONYmi6gK" "\"]")]])

;; tap/hbase
;; (defn hbase [table-name key-field column-family & value-fields]
;;   (let [scheme (HBaseScheme. (w/fields key-field) column-family (w/fields value-fields))]
;;     (HBaseTap. table-name scheme)))

;; This works
(let [short-urls-from-tweets (<- [?short]
                                 (sample-tweets _ ?url-string)
                                 (parse-urls ?url-string :> ?short))
      existing-short-urls (<- [?short-url]
                              ((tap/hbase "short_urls" "id" "rel" "url") ?short-url-bytes _)
                              (as-string ?short-url-bytes :> ?short-url))]
  (?<- (stdout)
       [?short-url ?long-url]
       (short-urls-from-tweets ?short-url)
       ((c/negate existing-short-urls) ?short-url)
       (url-contents ?short-url :> ?long-url _)))

;; This fails
(let [short-urls-from-tweets (<- [?short]
                                 (sample-tweets _ ?url-string)
                                 (parse-urls ?url-string :> ?short))
      existing-short-urls (<- [?short-url]
                              ((tap/hbase "short_urls" "id" "rel" "url") ?short-url-bytes _)
                              (as-string ?short-url-bytes :> ?short-url))]
  (?<- (tap/hbase "short_urls" "id" "rel" "url")
       [?short-url ?long-url]
       (short-urls-from-tweets ?short-url)
       ((c/negate existing-short-urls) ?short-url)
       (url-contents ?short-url :> ?long-url _)))

; [HBaseScheme[['id', 'ur...][sun.reflect.GeneratedConstructorAccessor59.newInstance(Unknown Source)] unable to resolve scheme sink selector: [{1}:'id'], with incoming: [{2}:'?short-url', '?long-url']