   [coll]
   (f/map coll (f/fn [[cnt k]] [k cnt])))
 
-(defn count-by-key
+(defn count-by-key-sorted
   "For a key, value collection, count the key, returning key, count ordered by count.
   Is a 'transformation' producing an RDD, is not an 'action'.
   The Spark count-by-key is an action, so not massively useful for large datasets."

          (f/fn [a b] (+ a b))))
      false)))

;; The parts of the analysis are divided into all-time, per year, month and day. This is because each pipline involves (potentially) re-reading the input stream
;; and there seems to be a bug or something that crops up when rereading the stream multiple times. Still unsolved: http://stackoverflow.com/questions/27403732/kryoexception-buffer-overflow-with-very-small-input
;; This does involve re-generating the stream each time, and not caching it, which isn't optimal. 

        doi-first-date (f/reduce-by-key doi-date (f/fn [a b] (util/min-date a b)))

        ; doi -> count
        doi-count (count-by-key-sorted doi-date)
        
        ; domain -> count
        domain-count (count-by-key-sorted domain-date)
        
        ; domain and subdomain, domain -> count
        ; including both subdomain and domain is necessary for the consumer of this dataset.
        subdomain-count (count-by-key-sorted subdomain-doi)]
      
  
    (.saveAsTextFile (f/map doi-first-date format-kv) (str output-location "/ever-doi-first-date"))
