   [data f name]
   (println (format "Cleaning data with %s..." name))
   (let [cdata (time (doall (f data)))]
-    (println "Number of rows:" (count cdata))
+    (print-nitems cdata)
     (newline)
     cdata))
 

  [data]
  (println "Flatting and cleaning data...")
  (let [fcdata (time (doall (remove nil? (reduce into data))))]
    (println "Number of rows:" (count fcdata))
    (newline)
    fcdata))


                        :to   max-point
                        :step rollup
                        :series by-path})))]
    (print-nitems (:series ndata))
    (newline)
    ndata))


(defn new-norm
  "New normalization."
  [paths data rollup to name]
  (println (format "Running new normalization (%s)..." name))
  (let [ndata
        (time (doall (let [min-point  (:time (first data))
                           max-point  (-> to (quot rollup) (* rollup))
                           points (range min-point (inc max-point) rollup)
                           dmap (reduce (fn [m e]
                                          (let [path (hash (:path e))
                                                time (:time e)
                                                metric (:metric e)]
                                            (assoc m path (assoc (get m path {}) time metric))))
                                        {} data)
                           metrics (pmap (fn [path]
                                           (let [h-path (hash path)
                                                 serie (get dmap h-path)]
                                             (map #(get serie % nil) points)))
                                         paths)
                           series (zipmap paths metrics)]
                       {:from min-point
                        :to   max-point
                        :step rollup
                        :series series})))]
    (print-nitems (:series ndata))
    (newline)))


;;------------------------------------------------------------------------------
;; Benchmark
