Bucket Sorting with Maps

For most sorting needs, general purpose algorithms like quicksort or merge sort will fit the bill quite nicely and are often provided as part of a language's standard library. Other times, however, other sorting algorithms can greatly outperform the generic algorithms. One case where the general-purpose algorithms is when there are for more records than unique values to sort by; e.g. if you're sorting a customer list by state or country. In those cases, we can reach for bucket sort, which is stable and can sort a collection in linear time, potentially with no comparisons.

Basically, the way that bucket sort works is that we split the things to be sorted into progressively smaller buckets until either the input is sorted to the granularity that we want or the bucket is small enough that an algorithm like insertion sort will be faster. As an example, if you're sorting by dates you could break it down into buckets by year, since every date in 1990 will be earlier than every date in 1991. Since no comparisons are needed to split the input into buckets, it can be done in linear time. If, then, a particular year had a lot of entries, we can split it up by month and then by day of the month, if need-be.

When we have discrete values, like in the states example above, we can take the straight-forward approach and just use the values as buckets. It's certainly possible to implement a custom way of mapping keys to buckets, but it's easier to use the map implementation that comes with your programming language. In Java, that leaves us with a decision to make between using a TreeMap or a HashMap. At first blush, it looks like TreeMap would be the obvious choice, since we could just do the inserts and then iterate over its values. If we look closer at the expected runtime, though, we see that the hash map implementation should be faster, since each insert involves a log num-keys tree search.

Here's an example implementation using a hash map.


The KeyExtractor argument is just a simple interface that, given an item, returns the key for that item.
Given that the whole point of this exercise was to beat the performance of a generic sort, we should see how it performs. To test that, I generated a 10,000 record list with 50 distinct keys and then sorted a copy of it using Collections.sort and the bucket sort 10,001 times to give the optimizer a chance to run. Then I ran the sorts individually 100 times, capturing the fastest & slowest times and the total time (all in nanoseconds). The results are summarized below:

Run	Min time	Max time	Total time
Collections.sort	1,589,705	3,290,003	167,857,942
Bucket sort	191,537	243,981	20,913,723

The full source code is available on git at https://github.com/adbrowning/blog-code/blob/master/src/com/adbrowning/sort/HashMapSort.java

Stream Madness

People often forget, or don't realize the ramifications of, the fact that Java's I/O streams and readers/writers are decorators. Sure, they're reminded of it every time they type InputStream in = new BufferedInputStream(new FileInputStream("/some/path"));, and that usually leads to grumbling about how verbose Java is and how stupid it is to have to type all of that and, while we're at it, why didn't they just make streams buffered to begin with?

This apparent eccentricity of Java (and many other languages), actually allows for some really neat tricks like transparently decompressing data and converting from one character encoding to another (e.g. with InputStreamReader). The really slick trick, though, is that you can completely change the input or add additional input. For example, you can inject data into a stream based on some value in the stream, as in the following code:

package streams;

import java.io.*;

public class ImportingStream extends InputStream {
  public static final int RECORD_SEPARATOR = 0x1E;

  private InputStream wrapped;
  private InputStream injecting;
 
  public ImportingStream(InputStream wrapped) {
    this.wrapped = wrapped;
  }
 
  @Override
  public int read() throws IOException {
    int retVal = -1;
    if(injecting != null) {
      retVal = injecting.read();
      if(retVal == -1) {
        /* done with the injected stream, null it out & call myself
         *  recursively */
        injecting.close();
        injecting = null;
        return read();
      }
    } else {
      retVal = wrapped.read();
      if(retVal == RECORD_SEPARATOR) {
        // starting a new injection of data
        ByteArrayOutputStream sink = new ByteArrayOutputStream();
        for(int read = wrapped.read(); -1 != read; read = wrapped.read()) {
          if(read == RECORD_SEPARATOR) {
             break;
          }
          sink.write(read);
        }
        
        injecting = new BufferedInputStream(
           new FileInputStream(new String(sink.toByteArray())));
         return read();
       }
    }
    return retVal;
  }
 
  public static void main(String[] args) throws IOException {
    ByteArrayOutputStream sink = new ByteArrayOutputStream();
    sink.write("Some header text ".getBytes());
    sink.write(RECORD_SEPARATOR);
    sink.write("/tmp/inserted_text.txt".getBytes());
    sink.write(RECORD_SEPARATOR);
    sink.write(" and this is some footer text".getBytes("utf8"));
    InputStream in = new ImportingStream(
      new ByteArrayInputStream(sink.toByteArray()));
    sink.reset();
    for(int read = in.read(); -1 != read; read = in.read()) {
      sink.write(read);
    }
    System.out.println(new String(sink.toByteArray()));
  }
}

Another interesting way that this can be used is to switch the method that you use to read the data mid-stream. One example where that might come in handy is parsing XML documents with embedded base-64 encoded binary; e.g. if you have an XML file with an embedded PDF that you want to index for searching in Lucene and you want the PDF's contents to be included inline. You could use a multi-step process where you split the document into sections, run Tika against the binary segments and then re-merge them, or you could just run Tika inline (though you'd probably want to use a co-processes/thread for running Tika in and either an exchanger or pipes, just to keep the code a little simpler). A vastly simpler, if more contrived, use-case for dealing with embedded data would be sending a large chunk of highly compressible data through a system that can only accept ASCII text and requires specific headers & footers on data. Here's example code for an InputStream for just such an occasion.

package streams;

import java.io.*;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import org.apache.commons.codec.binary.Base64InputStream;
import org.apache.commons.codec.binary.Base64OutputStream;

public class MultipartStream extends InputStream {
  public static final int RECORD_SEPARATOR = 0x1E;
  private InputStream input;
  private boolean inCompressed = false;
  private GZIPInputStream gzIn;

  public MultipartStream(InputStream in) {
    input = in;
  }

  public static void main(String[] args) throws IOException {
    ByteArrayOutputStream sink = new ByteArrayOutputStream();
    sink.write("This is some header text!\n".getBytes("utf8"));
    sink.write((byte) RECORD_SEPARATOR);
    OutputStream temp = new GZIPOutputStream(
      new Base64OutputStream(sink));
    byte[] toWrite = "And this is some compressed body " + 
      "data! \n".getBytes("utf8");
    for(int i = 0; i < 10; ++i) {
      temp.write(toWrite);
    }
    temp.close();
    sink.write(RECORD_SEPARATOR);
    sink.write("\nAnd this is some footer text\n".getBytes("utf8"));
    sink.close();
    System.out.println("Sink:\n" + new String(sink.toByteArray(), "utf8"));
    System.out.println("-----------------------------------------------");
    final MultipartStream inputStream = new MultipartStream(
      new ByteArrayInputStream(sink.toByteArray()));
    Reader in = new InputStreamReader(inputStream, "utf8");
    char[] chars = new char[4096];
    int numRead = 0;
    for(int read = in.read(chars); -1 != read; read = in.read(chars)) {
      numRead += read;
      System.out.print(new String(chars, 0, read));
    }
    System.out.println("Read " + numRead + " characters");
  }

  @Override
  public int read() throws IOException {
    int read = inCompressed ? gzIn.read() : input.read();
    if(inCompressed && read == -1) {
      inCompressed = false;
      read = input.read();
      gzIn.close();
      gzIn = null;
    } else if(read == RECORD_SEPARATOR) {
      inCompressed = true;
      gzIn = new GZIPInputStream(
        new Base64InputStream(
          new RecordSeparatedInputStream(input)));
      return read;
    }
    return read;
  }
 
  private static class RecordSeparatedInputStream extends InputStream {
    private InputStream in;
    public RecordSeparatedInputStream(InputStream in) {
      this.in = in;
    }

    @Override
    public int read() throws IOException {
      int retVal = in.read();
      if(retVal == RECORD_SEPARATOR) {
        retVal = -1;
      }
      return retVal;
    }
  }
}

Of course, in practice you'd probably want to implement FilterInputStream for all of these cases to make it clear to users of the code that it's meant to wrap and modify another InputStream. The reason that I didn't implement that here is that it also requires overriding the read methods taking a byte array to use the no-arg read method.

Custom input streams aren't always the best solution to a problem, since they do add a translation between source data and what the final processor actually sees, but they do provide another avenue for dealing with inconvenient data. Hopefully this has been informative enough to add another tool to your development toolbox, or at least given you an increased appreciation for the capabilities of input streams.

RIP dmr

Dennis Ritchie has passed away, and the world is a lesser place for it.

#include <stdio.h>

int main(void) {
   printf("Goodbye, dmr, and thanks for everything\n");
   return 0;
}

High-Level Hadoop

At a high level, map reduce converts an input record to one or more output records, and then combines those output records into the final output records. The way that it does this can seem like magic, since so much of it is handled by the framework. Hopefully, this will demystify some of what's going on.

You start by configuring the job with a Configuration. The configuration tells Hadoop things like the location of your input, where to put your output, what class implements your mapper and reducer, and how the input and output are formatted.

The input formatter and its record reader take the input file (or files) and split them into records that are passed to the map class.¹

For each record that the record reader produces, map gets called once to produce an output record. Those records get stored in memory for awhile, and then written to disk when memory runs low or the map-phase is finished.²

Once the mappers are done, their output gets sorted by its key, so that the records group together by key. Then those records are split up by key and sent off to reducers (all of the values for any given key will go to the same reducer, though).

The reducer then gets called with a key and all of the values for that key (as an Iterator). The reducer then does its thing with the values and writes them to its Context, which will send them to the output formatter, which deals with persisting the results.

Footnotes:

_{1. The files in HDFS are split into chunks and replicated across the cluster. For performance, Hadoop will try to run the mappers on the same machines that house the blocks of the file. Unless you're writing your own input format, this will be transparent to you.}

_{2. You can also add a combiner, which does the same thing as the reducer and, in fact, can be the same class as your reducer. Adding one can potentially improve performance dramatically by drastically reducing the amount of data that has to be sorted and distributed to reducers}

Know When to Fold 'Em

The Map & Reduce approach to dealing with large data sets, as implemented in Hadoop, has its background in functional programming, where both are used to manipulate the elements of a collection. Specifically, map is used to transform one element from the collection to some other type, for example to turn a string representation of an integer into an actual integer. Reduce, on the other hand, is used to combine all of the elements of the collection into a single result (which may, in turn, be another collection).

With just these two “primitives”, you can do a lot of powerful number crunching (or string manipulation, as the case may be), but what if you've got a lot of data coming in constantly and you need to be able to answer questions quickly? (Hadoop tends to have the throughput of a delivery truck full of hard drives and the latency of a delivery truck full of hard drives)

First, let's tackle the delivering-results-quickly part of the problem. Traditional RDBMS' handle this by building indices on the data for the fields that are most frequently searched. Fortunately, there's absolutely nothing that says that we can't do that with Map-Reduce. By building indices with map-reduce, we can then answer questions (that we know about ahead of time) about the data much more quickly. We can also calculate things like sums and averages on the data, so that answering analytics questions is just a simple value lookup.

Now, on to the constant influx of data problem. In functional programming, reduce has a “sister” function called fold that does the same thing as reduce except it takes a starting value to reduce (or fold) into. With fold, you can take the results that you calculated earlier and incorporate them into the new data (or vice versa). So, right now you may be thinking, “Well hello Mr. Fancy Pants. I've got news for you pal, Hadoop doesn't support fold, just reduce!” This is where the relationship between fold and reduce comes in (warning, LISP-like pseudocode follows); (fold startingValue list) has the same effect as (reduce (cons startingValue list)). We can take advantage of that equality by having a mapper that generates the previous run's results as input to a reducer, which combines it with the new data's results.

Not Your Type

Not too long ago, I was asked by a friend why I like Scala's type system and how it helps develop software faster (relative to dynamically typed languages, like Ruby). The short answer is that it doesn't; the slightly longer answer is that that isn't the point. An effective type system is more about making the computer help you get your code right, because throwing $125,000,000 down the drain when one person uses metric and the other English units looks bad on the next budget request.

I suspect that most detractors of type systems are doing so from the perspective of weak type systems (like Java's), which really does get in the way as much, if not more, than it helps. A more powerful type system (e.g. Haskell's or Scala's), however, can actually make it a compile error to make some subtle errors. As an example, here's some Scala to make Velocity a type defined by Meters/Seconds:


class Meters(val dist : Float) {
def /(time : Seconds) : Velocity = { new Velocity(dist/time.magnitude)}
}

class Seconds(val magnitude : Float)

class Velocity(val magnitude: Float) {
override def toString() = magnitude + " m/s"
}

class Feet(val magnitude : Float)


With that definition, you can define something of type Meters, divide it by something of type Seconds and get a Velocity. Ok, so far so good, but that doesn't seem to be that helpful... until you find that you can not multiply the Meters by a typeless unit or accidentally by Feet! Now, you may be thinking, "Hey, that's still getting in my way! I have a parameter in Feet and I don't want to bother changing it to meters!" That's where the second trick up Scala's sleeve comes in, by providing a way to implicitly convert from Feet to Meters.

By adding this little snippet of code:

implicit def feetToMeters(feet : Feet) = new Meters(feet.dist * 0.3048F)

we can now divide Feet by Seconds and get Velocity in meters / second!

So now, these two statements print the same value:

println(new Meters(1000)/new Seconds(60))
println(new Feet(3280.8399F)/new Seconds(60))




And helping avoid turning a Mars Orbiter into a Mars Crasher is just one more reason why I really like Scala.

QuickBuzz

A few of my friends use FizzBuzz to screen candidates, and that's a good thing because it's really good at weeding out the completely incompetent. The problem comes when you need someone who's more than just competent, you need someone who actually knows something. In those cases, I propose using Quick sort as a 2nd level FizzBuzz problem (or even a replacement for FizzBuzz). Quick sort is simple enough that it can be implemented in almost no time and almost no code by anyone who knows their programming language. It also has the benefit of opening the door for follow-on questions, like how do you avoid the algorithm taking n^2 time if the list is in order or what do you do if the list is too big to fit into main memory.