               (parse-int ?count :> ?cnt) 
               (o/sum ?cnt :> ?total))))
 
+(defn roll-up-by-minute 
+  ""
+  [input-directory output-directory]
+  (let [data-point (metrics input-directory)
+        output (hfs-delimited output-directory :sinkmode :replace :delimiter ",")]
+       (c/?<- output 
+              [?year ?month ?day ?hour ?minute ?rpt] 
+              (data-point ?year ?month ?day ?hour ?minute ?entity ?action ?count) 
+              (agg-perf-data ?entity ?action ?count :> ?rpt))))
+
 (defn -main
   ""
   [& args]


(defn parse-int [s]
   (Integer. (re-find  #"\d+" s )))

(defn parse-data-line
  ""

          (parse-data-line ?line :> ?year ?month ?day ?hour ?minute ?entity ?action ?count)
          (:distinct false))))

(defn roll-up-by-minute-entity-action 
  ""
  [input-directory output-directory]
  (let [data-point (metrics input-directory)
