1 биллион челлендж

;; using manual looping and mutable hashmaps

(defn looping-stats-mutable!
  "Returns the map of station stats keyed by station names.
  Reads a semicolon-separated file with measurement records eagerly
  line by line, parses each line into a measurement (station name and
  measured value) and updates the corresponding station's stats using
  the value. Uses mutable hashmaps to store total and station stats."
  [filename]
  (let [rdr (io/reader filename)]
    (loop [stats (java.util.HashMap.)]
      (if-let [line (.readLine ^java.io.BufferedReader rdr)]
        (let [[key string-value] (s/split line scsv-split-regex)
              value              (parse-double string-value)
              {:keys [count minimum maximum average] :as key-stats}
              (or (.get ^java.util.HashMap stats key)
                  (doto (java.util.HashMap.)
                    (.put :count 0)
                    (.put :maximum value)
                    (.put :minimum value)
                    (.put :average value)))
              updated-key-stats
              (doto ^java.util.HashMap key-stats
                (.put :count (inc count))
                (.put :minimum (min minimum value))
                (.put :maximum (max maximum value))
                (.put :average (moving-average average count value)))]
          (recur (doto stats (.put key updated-key-stats))))
        stats))))

(comment

  ;; ~50% better than baseline (and ~30% better than the transducing variant)

  ;; ~0.7 sec
  (time (format-stats (looping-stats-mutable! "dev/resources/measurements1M.txt")))

  ;; ~750 sec (~12 min)
  (time (format-stats (looping-stats-mutable! "dev/resources/measurements.txt")))

)

(ns ubrc.simple
  (:require
   [clojure.string :as str]))

(defn merge-two-stations [station1 station2]
  {:min (min (:min station1) (:min station2))
   :max (max (:max station1) (:max station2))
   :n (+ (:n station1) (:n station2))
   :t (/ (+ (* (:n station1) (:t station1)) (* (:n station2) (:t station2)))
         (+ (:n station1) (:n station2)))})

(defn update-station [station temperature]
  (let [n (:n station)
        n1 (inc n)
        average-temperature (:t station)
        new-average (/ (+ temperature (* n average-temperature)) n1)]
    {:min (min (:min station) temperature)
     :max (max (:max station) temperature)
     :n n1 :t new-average}))

(defn new-station [temperature]
  {:min temperature
   :max temperature
   :n 1 :t temperature})

(defn get-new-station [station temperature]
  (if station (update-station station temperature)
               (new-station temperature)))

;; stations example    -- {:Test {:min -4, :max 32, :n 2, :t 2.2} :Milan {:min 2, :max 22, :n 1, :t 8.7}}
;; new-station example -- {:Test 3.3}
(defn add-station [stations new-station]
  (let [name (first (keys new-station))
        temperature (get new-station name)
        station (get stations name)
        new-station (get-new-station station temperature)]
    (assoc stations name new-station)))

(defn parse-station [station]
  (let [arr (str/split station #";")
        name (first arr)
        t (read-string (second arr))]
    {(keyword name) t}))

(defn read-file [file-name]
  (str/split (slurp file-name) #"\n"))

(defn iter [file-name]
  (let [seq (read-file file-name)]
    (reduce add-station {} (map parse-station seq))))

(defn run [files]
  (let [dbs (pmap iter files)]
    (reduce #(merge-with merge-two-stations %1 %2)
            dbs)))

(print-result (simple/run dataFiles))

java -jar /home/ig/tmp/ubrc/target/uberjar/ubrc-0.1.0-SNAPSHOT-standalone.jar  4887.45s user 77.95s system 933% cpu 8:52.01 total

	    data = data[workerSize*workerID : last]
	    data = data[bytes.IndexByte(data, '\n')+1 : bytes.LastIndexByte(data, '\n')+1]

while (segment[size - 1] != '\n') {
    size--;
}

(ql:quickload "uiop")
(ql:quickload :parse-float)
(ql:quickload "alexandria")
(ql:quickload "lparallel")

(defpackage :blc-challenge
  (:use :cl :lparallel :lparallel.queue)
  (:export #:main))

(in-package :blc-challenge)

(defun init ()
  (setf *kernel* (make-kernel 8 :name "channel-queue-kernel")))

(init)


(declaim (optimize (speed 3) (safety 0)))

(defparameter dataDir "~/src/github.com/gunnarmorling/1brc/data")
(defparameter dataFiles (remove-if-not #'(lambda (path) (uiop:string-prefix-p "x" (file-namestring path))) (uiop:directory-files dataDir)))

(defstruct station
  (min 0.0 :type single-float)
  (max 0.0 :type single-float)
  (n 1 :type fixnum)
  (avg 0.0 :type single-float))

(defun parse-station (l)
  (let ((station (uiop:split-string l :max 2 :separator ";")))
    (cons (intern (car station))
	  (parse-float:parse-float (cadr station)))))

(defun new-avg (temperature n average-temperature)
  (/ (+ temperature (* n average-temperature)) (1+ n)))

(defun updated-station (station temperature)
  (make-station :min (min temperature (station-min station))
		:max (max temperature (station-max station))
		:n (1+ (station-n station))
		:avg (new-avg temperature (station-n station) (station-avg station))))


(defun add-station (db station)
  (let* ((station-name (car station))
	 (temperature (cdr station))
	 (old-station (gethash station-name db)))
    (setf (gethash station-name db)
	  (if old-station (updated-station old-station temperature)
	      (make-station :min temperature
			    :max temperature
			    :n 1
			    :avg temperature)))))

(defun iter (file-name)
  (let ((db (make-hash-table :size 413)))
    (with-open-file (in file-name)
	    (loop for l = (read-line in nil nil)
		  while l
		  do (add-station db (parse-station l))))
    db))

(defun merge-two-stations (db1 db2)
  (maphash #'(lambda (name station2)
		    (let* ((station1 (gethash name db1))
			   (new-station
			     (make-station :min (min (station-min station1)
						     (station-min station2))
					   :max (max (station-max station1)
						     (station-max station2))
					   :n (+ (station-n station1) (station-n station2))
					   :avg (/ (+ (* (station-n station1) (station-avg station1))
						      (* (station-n station2) (station-avg station2)))
						   (+ (station-n station1) (station-n station2))))))
		      (setf (gethash name db1) new-station)))
	   db2)
  db1)

(defun merge-stations (stations)
  (reduce #'merge-two-stations stations))

(defun print-station (name station)
  (format nil "~a=~,1F/~,1F/~,1F"
	  name (station-min station) (station-avg station) (station-max station)))

(defun print-stations (stations)
  (let* ((names (sort
		(mapcar #'string
			(alexandria:hash-table-keys stations))
		#'string-lessp))
	 (statistics (loop for name in names
			   collect (print-station name (gethash (intern name "COMMON-LISP-USER") stations)))))
    (format t "{~{~A~^ ~}}~%" statistics)))

(defun run (files)
  (merge-stations (pmapcar #'iter files)))

(defun main ()
  (print-stations (run dataFiles)))

(time (main))
;; лишние строки не показаны
Evaluation took:
  219.599 seconds of real time
  1690.001351 seconds of total run time (1675.769479 user, 14.231872 system)
  [ Real times consist of 5.983 seconds GC time, and 213.616 seconds non-GC time. ]
  [ Run times consist of 5.861 seconds GC time, and 1684.141 seconds non-GC time. ]
  769.59% CPU
  569,191,774,868 processor cycles
  51 page faults
  349,112,268,000 bytes consed

;; using multiple threads to read the file (divided to logical segments) and
;; compute stats for each segment, then combining the results

(def ^:dynamic *end-of-file*
  "Integer value returned by a file input stream when end of file is reached."  -1)

(def ^:dynamic *end-of-line*
  "Integer value that corresponds to the end of line character." 10)

(defn file-segment
  "Returns a logical segment of `file`, defined by start and end
  positions in bytes. The segment will begin at `start` and end after
  newline nearest to `start` + `initial-length` position (inclusive),
  i.e., segment always ends between lines (or at the end of file);
  `start` defaults to 0; `initial-length` defaults to the length of
  the entire file."
  ([^File file]
   (file-segment file 0))
  ([^File file start]
   (file-segment file start (.length file)))
  ([^File file start initial-length]
   (when (pos? initial-length)
     (with-open [file-input-stream (FileInputStream. file)]
       (let [file-channel         (.getChannel file-input-stream)
             end-of-line-or-file? #{*end-of-file* *end-of-line*}
             at-initial-end       (+ start initial-length)
             before-initial-end   (dec at-initial-end)]
         {:file  file
          :start start
          :end   (do
                   (.position file-channel ^int before-initial-end)
                   (let [initial-end-byte (.read file-input-stream)]
                     (if (end-of-line-or-file? initial-end-byte)
                       (min (.position file-channel) (.length file))
                       (do
                         (.position file-channel ^int at-initial-end)
                         (while (not (end-of-line-or-file? (.read file-input-stream))))
                         (min (.position file-channel) (.length file))))))})))))

(defn file-segment-seq
  "Returns a lazy sequence of at most `n` successive logical segments of
  `file`. (As segments may be expanded to align with line boundaries,
  there may be fewer of them, especially if segments are small and
  lines are long.)"
  ([^java.io.File file n]
   (file-segment-seq file 0 n))
  ([^java.io.File file start n]
   (when (> n 0)
     (let [remaining-file-length  (- (.length file) start)
           initial-segment-length (int (/ remaining-file-length n))]
       (when-let [segment (file-segment file start initial-segment-length)]
         (cons segment (lazy-seq (file-segment-seq file (:end segment) (dec n)))))))))

(defn file-segment-line-seq
  "Returns a lazy sequence of lines from the `segment`. Reading begins
  at `start` and ends after `end`. (The line that contains the end
  position will be the last read. Segment constructor guarantees that
  segment will always end at a line boundary, but manually created
  segments may not.)"
  ([{:keys [file start end]}]
   (let [file-input-stream (FileInputStream. ^File file)
         _                 (.position (.getChannel file-input-stream) ^int start)
         reader            (io/reader file-input-stream)]
     (file-segment-line-seq reader 0 (- end start))))
  ([^BufferedReader reader bytes-read bytes-max]
   (when (not (>= bytes-read bytes-max))
     (when-let [line (.readLine reader)]
       (let [line-length-bytes (inc (count (.getBytes line)))]
         (cons line (lazy-seq (file-segment-line-seq reader
                                                     (+ bytes-read line-length-bytes)
                                                     bytes-max))))))))

(defn merge-station-stats
  "Returns two station stats maps merged."
  [{cnt1 :count min1 :minimum max1 :maximum avg1 :average}
   {cnt2 :count min2 :minimum max2 :maximum avg2 :average}]
  {:count   (+ cnt1 cnt2)
   :minimum (min min1 min2)
   :maximum (max max1 max2)
   :average (/ (+ (* cnt1 avg1) (* cnt2 avg2)) (+ cnt1 cnt2))})

(defn parallel-stats
  "Returns the map of station stats keyed by station names. Reads the
  measurements data from `filename`, that is a text file containing
  semicolon-separated values. Uses multiple threads to read the
  file (divided to `n` logical segments) and computes stats for each
  segment, then combines the results. Default number of segments is the
  number of available cores + 2, same as the number of threads used by
  `pmap`."
  ([filename]
   (parallel-stats filename (+ 2 (.. Runtime getRuntime availableProcessors))))
  ([filename n]
   (let [file (io/file filename)]
     (reduce #(merge-with merge-station-stats %1 %2)
             (pmap (fn [segment]
                     (transducing-stats (file-segment-line-seq segment)))
                   (file-segment-seq file n))))))

(comment

  ;; 80% better than baseline (4.5 min vs 22 min), with 4 physical (8 logical) cores (Ryzen 5 3400G)

  ;; 1e6 lines: 0.3 sec
  (time (format-stats (parallel-stats "dev/resources/measurements1M.txt")))

  ;; 1e9 lines: 263 sec (~4.5 min)
  (time (format-stats (parallel-stats "dev/resources/measurements.txt")))

)