(ns dataset.optmised-by-chris
  (:require
    [tech.v3.io :as io]
    [tech.v3.datatype.functional :as dfn]
    [tech.v3.dataset :as ds]
    [tech.v3.dataset.reductions :as ds-reduce]
    [tech.v3.dataset.utils :as utils]
    [tech.v3.libs.arrow :as arrow]
    [tech.v3.libs.parquet :as parquet]
    [com.climate.claypoole :as cp])
  (:import [java.util Map HashMap Set HashSet Map$Entry List
            ArrayList]
           [java.util.concurrent ConcurrentHashMap]
           [java.util.function Function Consumer BiConsumer]
           [java.util.stream Stream]
           [tech.v3.datatype PrimitiveList IndexReduction])
  (:gen-class))


(set! *warn-on-reflection* true)
(set! *unchecked-math* :warn-on-boxed)


(utils/set-slf4j-log-level :info)


(def files (->> (io/file "/data/performance-benchmark-data")
                file-seq
                (filter #(.endsWith ^String (.toString ^Object %) "parquet"))
                (map #(.toString ^Object %))))


;;This is a thing that geni gets you - spark dataframe systems detect the dependencies
;;and don't load data they do not need to load.
(def required-columns ["member-id" "price" "quantity" "date" "brand-id" "style-id"])


;;Because we are only loading the columns we need, it is easy to load the entire dataset
;;into memory.
(defn load-combined-parquet
  []
Use exactly 12 threads to load the parquet files .
  (cp/pmap 12 #(ds/->dataset % {:column-whitelist required-columns})
           files))


;;Writing the dataset out to a store where we can access everything even if it is
;;larger than available memory is also possible.
(defn write-combined-arrow!
  []
  (arrow/write-dataset-seq-to-stream! (load-combined-parquet)
                                      "../benchmark-data.arrow"))


;;Loading this entire dataset then takes milliseconds regardless of size.
(defn load-combined-arrow
  []
  (arrow/stream->dataset-seq-inplace "../benchmark-data.arrow"))


(defn dataset-seq->group-by-aggregate
  [ds-seq]
  (-> (->> (map #(assoc % "sales" (dfn/* (% "price") (% "quantity"))) ds-seq)
           ;;group-by-column-agg returns a new dataset with the columns named after
           ;;the keys in the aggregation map.
           (ds-reduce/group-by-column-agg
            "member-id"
            {:total-spend (ds-reduce/sum "sales")
             :avg-basket-size (ds-reduce/mean "sales")
             :avg-price (ds-reduce/mean "price")
             :n-transactions (ds-reduce/row-count)
             ;;Counting distinct when you know the the values fit in
             ;;int32 space is far faster than counting distinct of arbitrary
values due to  's excellent implementation .
             :n-visits (ds-reduce/count-distinct "date" :int32)
             :n-brands (ds-reduce/count-distinct "brand-id" :int32)
             :n-styles (ds-reduce/count-distinct "style-id" :int32)}))
      (ds/write! "output.nippy")))


(comment

  (time (def ignored (dataset-seq->group-by-aggregate
                      (load-combined-parquet))))

  (time (def ignored (dataset-seq->group-by-aggregate
                      (take 1 (load-combined-parquet)))))

  (time (def ignored (dataset-seq->group-by-aggregate
                      (load-combined-arrow))))

32

  (time (def ignored (dataset-seq->group-by-aggregate
                      (take 1 (load-combined-arrow)))))

  )

This is a thing that geni gets you - spark dataframe systems detect the dependencies
and don't load data they do not need to load.
Because we are only loading the columns we need, it is easy to load the entire dataset
into memory.
Writing the dataset out to a store where we can access everything even if it is
larger than available memory is also possible.
Loading this entire dataset then takes milliseconds regardless of size.
group-by-column-agg returns a new dataset with the columns named after
the keys in the aggregation map.
Counting distinct when you know the the values fit in
int32 space is far faster than counting distinct of arbitrary
(ns dataset.optmised-by-chris
  (:require
    [tech.v3.io :as io]
    [tech.v3.datatype.functional :as dfn]
    [tech.v3.dataset :as ds]
    [tech.v3.dataset.reductions :as ds-reduce]
    [tech.v3.dataset.utils :as utils]
    [tech.v3.libs.arrow :as arrow]
    [tech.v3.libs.parquet :as parquet]
    [com.climate.claypoole :as cp])
  (:import [java.util Map HashMap Set HashSet Map$Entry List
            ArrayList]
           [java.util.concurrent ConcurrentHashMap]
           [java.util.function Function Consumer BiConsumer]
           [java.util.stream Stream]
           [tech.v3.datatype PrimitiveList IndexReduction])
  (:gen-class))


(set! *warn-on-reflection* true)
(set! *unchecked-math* :warn-on-boxed)


(utils/set-slf4j-log-level :info)


(def files (->> (io/file "/data/performance-benchmark-data")
                file-seq
                (filter #(.endsWith ^String (.toString ^Object %) "parquet"))
                (map #(.toString ^Object %))))


(def required-columns ["member-id" "price" "quantity" "date" "brand-id" "style-id"])


(defn load-combined-parquet
  []
Use exactly 12 threads to load the parquet files .
  (cp/pmap 12 #(ds/->dataset % {:column-whitelist required-columns})
           files))


(defn write-combined-arrow!
  []
  (arrow/write-dataset-seq-to-stream! (load-combined-parquet)
                                      "../benchmark-data.arrow"))


(defn load-combined-arrow
  []
  (arrow/stream->dataset-seq-inplace "../benchmark-data.arrow"))


(defn dataset-seq->group-by-aggregate
  [ds-seq]
  (-> (->> (map #(assoc % "sales" (dfn/* (% "price") (% "quantity"))) ds-seq)
           (ds-reduce/group-by-column-agg
            "member-id"
            {:total-spend (ds-reduce/sum "sales")
             :avg-basket-size (ds-reduce/mean "sales")
             :avg-price (ds-reduce/mean "price")
             :n-transactions (ds-reduce/row-count)
values due to  's excellent implementation .
             :n-visits (ds-reduce/count-distinct "date" :int32)
             :n-brands (ds-reduce/count-distinct "brand-id" :int32)
             :n-styles (ds-reduce/count-distinct "style-id" :int32)}))
      (ds/write! "output.nippy")))


(comment

  (time (def ignored (dataset-seq->group-by-aggregate
                      (load-combined-parquet))))

  (time (def ignored (dataset-seq->group-by-aggregate
                      (take 1 (load-combined-parquet)))))

  (time (def ignored (dataset-seq->group-by-aggregate
                      (load-combined-arrow))))

32

  (time (def ignored (dataset-seq->group-by-aggregate
                      (take 1 (load-combined-arrow)))))

  )
