TemplateTap + thrift + Hfs-seqfile / hfs-lzo-thrift

gistfile1.clj

;; thrifters

(defn thrift-order-item-line_price_net
  [id val]
  (DataUnit/order_item_property
    (OrderItemProperty.
      (OrderItemID/id id)
      (OrderItemPropertyValue/line_price_net val)
      (. (new Date) getTime))))

(defn thrift-order-item-line_price_inc
  [id val]
  (DataUnit/order_item_property
    (OrderItemProperty.
      (OrderItemID/id id)
      (OrderItemPropertyValue/line_price_inc val)
      (. (new Date) getTime))))

;; convert one record of the form id, field-value1, field-value2
;; to 2 records field-name1, thrift-value1 and field-name2, thrift-value2
;; this is done to facilitate templating in sink tap based on fields names

(defmapcatop thriftcat [id p1 p2]
  [["line_price_inc" (thrift-order-item-line_price_inc id p1)]
   ["line_price_net" (thrift-order-item-line_price_net id p2)]])

;; fetch MySQL data from source-tap (dbmigrate) and dump to TemplateTap
;; one file is generated per fetched field
;; so with one source query of N fields we can generate N files

;; Using hfs-seqfile : works fine

(defn dump-order-item-prices []
  (let [fields ["id" "line_price_inc" "line_price_net"]
        cvars  (vec (map #(str "?" % "") fields))]
    (?<- (hfs-seqfile "/data/OrderItemProperty" :sinkmode :replace 
                                                :sink-template "%s/" 
                                                :templatefields ["?prop"] 
                                                :outfields ["?thrift-object"])
         [?prop ?thrift-object]
         ((source-tap fields) :>> cvars )
         (thriftcat :<< cvars :> ?prop ?thrift-object )
         (:distinct false))))

;; Using hfs-lzo-thrift : it breaks

(defn dump-order-item-prices []
  (let [fields ["id" "line_price_inc" "line_price_net"]
        cvars  (vec (map #(str "?" % "") fields))]
    (?<- (hfs-lzo-thrift "/data/OrderItemProperty" DataUnit
                                                :sinkmode :replace 
                                                :sink-template "%s/" 
                                                :templatefields ["?prop"] 
                                                :outfields ["?thrift-object"])
         [?prop ?thrift-object]
         ((source-tap fields) :>> cvars )
         (thriftcat :<< cvars :> ?prop ?thrift-object )
         (:distinct false))))