gszeliga

(defn process! [path fname npar]
  (let [lines         (mult (stream-lines-from path))
        aggregate-tap (tap lines (chan 1024))
        count-tap     (tap lines (chan 1024))]
    (<!! (async/reduce 
          (fn [_ v] (println v)) 
          "" 
          (async/merge [(aggregate-field fname aggregate-tap npar)
                        (count-lines count-tap npar)])))))

gszeliga

(defn aggregate-field [fname channel npar]
  (let [as-value-fn       (extract-field-fn fname channel)
        fvalue-chan       (pipe channel (chan 1024 (map as-value-fn)))
        fvalue-aggregator (for [_ (range npar)]
                            (w-aggregate fvalue-chan))]
    (go
      (let [[sum count] (<! (async/reduce 
                             #(apply map + [%1 %2]) 
                             [0.0 0] 
                             (async/merge fvalue-aggregator)))]

gszeliga

(defn w-aggregate [channel]
  (go-loop [c 0 s 0.0]
    (if-some [n (<! channel)]
      (recur (inc c) (+ n s))
      [s c])))

gszeliga

(defn extract-field-fn [fname channel]
  (let [headers   (<!! channel)
        field-idx (->> (split headers  #",") 
                       (map-indexed vector) 
                       (filter (fn [[_ v]] (= v fname))) 
                       first 
                       first)]
    #(java.lang.Double/parseDouble (nth (split % #",") field-idx))))

gszeliga

(defn stream-lines-from [path]                                                                                                                                                                 
  (let [c (chan 1024)]                                                                                                                                                                         
    (go                                                                                                                                                                                        
      (with-open [rdr (io/reader path)]                                                                                                                                                        
        (doseq [line (line-seq rdr)]                                                                                                                                                           
          (>! c line)))                                         

gszeliga

(comment ; Fun with transducers, v2

  ;; Still haven't found a brief + approachable overview of Clojure 1.7's new
  ;; transducers in the particular way I would have preferred myself - so here goes:

  ;;;; Definitions
  ;; Looking at the `reduce` docstring, we can define a 'reducing-fn' as:
  (fn reducing-fn ([]) ([accumulation next-input])) -> new-accumulation
  ;; (The `[]` arity is actually optional; it's only used when calling
  ;; `reduce` w/o an init-accumulator).

gszeliga

What follows is a technical test for this job offer at CARTO: https://boards.greenhouse.io/cartodb/jobs/705852#.WSvORxOGPUI

Build the following and make it run as fast as you possibly can using Python 3 (vanilla). The faster it runs, the more you will impress us!

Your code should:

• Download this 2.2GB file: https://s3.amazonaws.com/carto-1000x/data/yellow_tripdata_2016-01.csv
• Count the lines in the file
• Calculate the average value of the tip_amount field.

gszeliga

Atom Clojure Setup

This describes how I setup Atom for an ideal Clojure development workflow. This fixes indentation on newlines, handles parentheses, etc. The keybinding settings for enter (in keymap.cson) are important to get proper newlines with indentation at the right level. There are other helpers in init.coffee and keymap.cson that are useful for cutting, copying, pasting, deleting, and indenting Lisp expressions.

Install Atom

Download Atom

The Atom documentation is excellent. It's highly worth reading the flight manual.

gszeliga

    public static <T> CompletableFuture<Stream<T>> sequence(Stream<CompletableFuture<T>> source) {
        return source
                .reduce(completedFuture(Stream.empty()),
                        (fl, fo) -> fl.thenCombine(fo, (s, o) -> concat(s, Stream.of(o))),
                        (fa, fb) -> fa.thenCombine(fb, Stream::concat));
    }

gszeliga

package com.gszeliga.tictattoe.handlers;

import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.Random;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;