Using HyperLogLog in Cascalog Supporting tagline

Recently I have been working on using the HyperLogLog cardinality estimation algorithm with Cascalog. As I perused the web for examples of how other people were doing this I was able to find a couple of approaches to doing it.

This post is an off-shoot of an excellent article on this along with some good discussion about performance. Before you continue reading this I would encourage you to go over there and at least skim the article. They did a really nice job of explaining things, so I won't repeat most of it here.

In the article the authors did a very nice comparison of two different ways of using the HyperLogLog object from the library. Basically you can either use one object and "offer" values to it or create a new object with your value then merge it with another object that has values already in it. Not totally unexpectedly the performance is much better if you use one object and add values to it.

However, this doesn't take into account some of the optimization strategies that one needs to use when building MapReduce jobs. One of the things that I have learned while working with Cascalog and MapReduce over the last year or so is that it is very important to do as much work in the map tasks as possible. In other words, you really want to minimize the amount of data that needs to pass from map task to reduce task.

So what does this mean in the world of Cascalog? I have interpreted this to mean preferring defparallelagg to the other custom aggregators when doing aggregation. The problem is that to do some of the aggregation map-side we don't get a chance to hold on to any state like we do with the defaggregateop. So our defparallelagg is forced to create a new HLL object for each one of the tuple values and then merge that object into another.

So our parallel aggregator will look like this:

(c/defparallelagg agg-hyperloglog
  :init-var #'hll-construct
  :combine-var #'hll-merge)

The way this works is that the function we've defined as :init-var will get called once for each tuple. In this case the hll-construct function is creating a HyperLogLog object and offering the tuple value to it then returning the HLL object. Then the function defined as the :combine-var will get called with the results of the calls to the :init-var functions as well as other combines until there is only one value left. In this case, the hll-merge function gets two HyperLogLog objects (either returned from hll-construct or a previous call to hll-merge) and simply merges them.

So what's going to happen in our query is that each ?uuid value will be put into a HLL object and then successively merged into other HLL objects with one or more ?uuid values in them. This will happen first on each map task then those single HLL objects will be sent from the map tasks to the reduce tasks where they will all be combined (merged). And remember that this combining of HLL's is happening based on the aggregation grouping in the query (in this case we're grouping by city, year and day).

Here are hll-construct and hll-merge

(defn hll-construct
  "Construct a HyperLogLog and offer the given init-value. 12 bits gives a little less than 2% standard error."
  [init-value]
  (doto (HyperLogLog. 12) (.offer init-value)))

(defn hll-merge
  "Merge two or more HyperLogLog's together."
  [^HyperLogLog first-hlog & hlogs]
  (.merge ^HyperLogLog first-hlog (into-array HyperLogLog hlogs)))

You can find all of the code that I used in my Github repository. Most of the code is directly from the original blog post at Screen6. The code for the parallel implementation was adapted from a post on the Cascalog Google group by Jeroen van Dijk

(for [i (range 1 11)] (write-ov-chipkaart-accesslog-file (str "/tmp/chipkaart-" i) 1000000))

Published