Chapter 5. Big Data

	"More is different."
	--Philip Warren Anderson

In the previous chapters, we've used regression techniques to fit models to the data. In Chapter 3, Correlation, for example, we built a linear model that used ordinary least squares and the normal equation to fit a straight line through the athletes' heights and log weights. In Chapter 4, Classification, we used Incanter's optimize namespace to minimize the logistic cost function and build a classifier of Titanic's passengers. In this chapter, we'll apply similar analysis in a way that's suitable for much larger quantities of data.

We'll be working with a relatively modest dataset of only 100,000 records. This isn't big data (at 100 MB, it will fit comfortably in the memory of one machine), but it's large enough to demonstrate the common techniques of large-scale data processing. Using Hadoop (the popular framework for distributed computation) as its case study, this chapter will focus on how to scale algorithms to very large...

Downloading the code and data

This chapter makes use of data on individual income by the zip code provided by the U.S. Internal Revenue Service (IRS). The data contains selected income and tax items classified by state, zip code, and income classes.

It's 100 MB in size and can be downloaded from http://www.irs.gov/pub/irs-soi/12zpallagi.csv to the example code's data directory. Since the file contains the IRS Statistics of Income (SoI), we've renamed the file to soi.csv for the examples.

Note

The example code for this chapter is available from the Packt Publishing's website or https://github.com/clojuredatascience/ch5-big-data.

As usual, a script has been provided to download and rename the data for you. It can be run on the command line from within the project directory with:

script/download-data.sh

If you run this, the file will be downloaded and renamed automatically.

Inspecting the data

Once you've downloaded the data, take a look at the column headings in the first line of the file. One way...

The reducers library

The count operation we implemented previously is a sequential algorithm. Each line is processed one at a time until the sequence is exhausted. But there is nothing about the operation that demands that it must be done in this way.

We could split the number of lines into two sequences (ideally of roughly equal length) and reduce over each sequence independently. When we're done, we would just add together the total number of lines from each sequence to get the total number of lines in the file:

If each Reduce ran on its own processing unit, then the two count operations would run in parallel. All the other things being equal, the algorithm would run twice as fast. This is one of the aims of the clojure.core.reducers library—to bring the benefit of parallelism to algorithms implemented on a single machine by taking advantage of multiple cores.

Parallel folds with reducers

The parallel implementation of reduce implemented by the reducers library is called fold. To make use...

Mathematical folds with Tesser

We should now understand how to use folds to calculate parallel implementations of simple algorithms. Hopefully, we should also have some appreciation for the ingenuity required to find efficient solutions that will perform the minimum number of iterations over the data.

Fortunately, the Clojure library Tesser (https://github.com/aphyr/tesser) includes implementations for common mathematical folds, including the mean, standard deviation, and covariance. To see how to use Tesser, let's consider the covariance of two fields from the IRS dataset: the salaries and wages, A00200, the unemployment compensation, A02300.

Calculating covariance with Tesser

We encountered covariance in Chapter 3, Correlation, as a measure of how two sequences of data vary together. The formula is reproduced as follows:

A covariance fold is included in tesser.math. In the following code, we'll include tesser.math as m and tesser.core as t:

(defn ex-5-17 []
  (let [data (into [] (load-data...

Multiple regression with gradient descent

When we ran multiple linear regression in Chapter 3, Correlation, we used the normal equation and matrices to quickly arrive at the coefficients for a multiple linear regression model. The normal equation is repeated as follows:

The normal equation uses matrix algebra to very quickly and efficiently arrive at the least squares estimates. Where all data fits in memory, this is a very convenient and concise equation. Where the data exceeds the memory available to a single machine however, the calculation becomes unwieldy. The reason for this is matrix inversion. The calculation of is not something that can be accomplished on a fold over the data—each cell in the output matrix depends on many others in the input matrix. These complex relationships require that the matrix be processed in a nonsequential way.

An alternative approach to solve linear regression problems, and many other related machine learning problems, is a technique called gradient descent...

Scaling gradient descent with Hadoop

The length of time each iteration of batch gradient descent takes to run is determined by the size of your data and by how many processors your computer has. Although several chunks of data are processed in parallel, the dataset is large and the processors are finite. We've achieved a speed gain by performing calculations in parallel, but if we double the size of the dataset, the runtime will double as well.

Hadoop is one of several systems that has emerged in the last decade which aims to parallelize work that exceeds the capabilities of a single machine. Rather than running code across multiple processors, Hadoop takes care of running a calculation across many servers. In fact, Hadoop clusters can, and some do, consist of many thousands of servers.

Hadoop consists of two primary subsystems— the Hadoop Distributed File System (HDFS)—and the job processing system, MapReduce. HDFS stores files in chunks. A given file may be composed of many chunks and chunks...

Stochastic gradient descent

The method we've just seen of calculating gradient descent is often called batch gradient descent, because each update to the coefficients happens inside an iteration over all the data in a single batch. With very large amounts of data, each iteration can be time-consuming and waiting for convergence could take a very long time.

An alternative method of gradient descent is called stochastic gradient descent or SGD. In this method, the estimates of the coefficients are continually updated as the input data is processed. The update method for stochastic gradient descent looks like this:

In fact, this is identical to batch gradient descent. The difference in application is purely that expression is calculated over a mini-batch—a random smaller subset of the overall data. The mini-batch size should be large enough to represent a fair sample of the input records—for our data, a reasonable mini-batch size might be about 250.

Stochastic gradient descent arrives at the...

Summary

In this chapter, we learned some of the fundamental techniques of distributed data processing and saw how the functions used locally for data processing, map and reduce, are powerful ways of processing even very large quantities of data. We learned how Hadoop can scale unbounded by the capabilities of any single server by running functions on smaller subsets of the data whose outputs are themselves combined to finally produce a result. Once you understand the tradeoffs, this "divide and conquer" approach toward processing data is a simple and very general way of analyzing data on a large scale.

We saw both the power and limitations of simple folds to process data using both Clojure's reducers and Tesser. We've also begun exploring how Parkour exposes more of Hadoop's underlying capabilities.

In the next chapter, we'll see how to use Hadoop and Parkour to address a particular machine learning challenge—clustering a large volume of text documents.