The Trouble with Hashing

If you use Spark's HashingTF, you avoid needing an external store of key/values that complicates the code but there are downsides. Collisions aside, the matrix it generates is overly wide and sparse. This can cause problems for some machine learning algorithms.

There are a number of clever mathematical dimensionality reduction techniques but since HashingTF makes the matrix unnecessarily sparse, you can compact it trivially with:

import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.{udf, col}

def toIndices(i: Int)(r: Row): Array[Int] = r.getAs[SparseVector](0).indices

def compacting(replacements: Map[Int, Int])(v: SparseVector): SparseVector = { 
  val indexVals  = v.indices.zip(v.values)
  val newIndices = indexVals.map { case (i, x) => (replacements(i), x) }.sortBy(_._1)
  new SparseVector(replacements.size, newIndices.map(_._1), newIndices.map(_._2))
}

def compactingUdf(replacements: Map[Int, Int]): UserDefinedFunction = {
  val compactingFn = compacting(replacements) _
  udf(compactingFn)
}

(You can find this code in my GitHub repository).

Now, you can compact your matrix. First, collect the non-zero indices:

val toIndicesFn: Row => TraversableOnce[Int] = toIndices(0)
val allIndices = df.select(YOUR_COLUMN_NAME).flatMap(toIndicesFn).distinct().collect().zipWithIndex.toMap

and then replace the old indices

val cUdf        = compactingUdf(allIndices)
val compactedDF = df.withColumn(YOUR_COLUMN_NAME, cUdf(col(YOUR_COLUMN_NAME)))

Now your matrix is considerably more narrow and less sparse.

For example, using this technique decreases the time taken by Spark's RandomForestClassifier from over an hour to a few minutes. The results were still comparable in my example (23% vs 25%) but the sparse matrix actually reduces the number of nodes from 1000 to 821. With the compacted matrix, I can increase the number of trees to nearly 3000 before seeing the same warning.

Good Hash

Since the feature hashing code that comes with Spark is based on a 32-bit hash giving us lots of collisions, we went for a 64-bit implementation. But what makes a good hash algorithm?

Rather than roll my own, I used this code. But how do I make sure it's good? This Stack Exchange answer gave me some clues by using the chi-squared test. Here, we basically compare with some fairly simple maths, our expected answer to our actual answer to indicate whether there is an unusually high number of collisions that our algorithm generates.

"The chi-squared test of goodness of fit is used to test the hypothesis that the distribution of a categorical variable within a population follows a specific pattern of proportions, while the alternative hypothesis is that the distribution of the variable follows some other pattern." [1]

The Chi-Square Test

"The values of column and rows totals are called marginals because they are on the margin of the table... The numbers within the table ... are called joint frequencies...

"If the two variables are not related, we would expect that the frequency of each cell would be the product of its marginals, divided by the sample size." [1]

This is because the probabilities of A and B are:

P(A ∩ B) = P(A) P(B) iff A ⊥ B

In other words, given a sample size of N, the probability of A is

P(A) = N_A / N

and

P(B) = N_B / N

so

P(A ∩ B) = N_B N_A / N²

so the expected frequency would be

N P(A ∩ B) = N_B N_A / N

exactly as [1] says.

The Chi-Square distribution then looks like this:

χ = Σ_i,j=1 (O_ij - E_ij)²/ E_ij

where i and j are being summed over all the rows and columns.

We can then perform the chi-squared test. This involves looking up the critical value in a table (or calculating it) from the degrees of freedom and the observed chi-squared value. "Basically, if the chi-square you calculated was bigger than the critical value in the table, then the data did not fit the model, which means you have to reject the null hypothesis." (from here)

The Results

Using this new hashing algorithm for our feature hashing resulted in no collisions while hashing all words in the corpus of text where previously we were getting about 70 million collisions with the 32-bit version.

Consequently, the number of records we then had to compare dropped by about 20 million (or about 2% of the total).

Unfortunately, although I am able to execute a chi-square test on the 32-bit algorithm, the 64-bit algorithm has such a large possible number of hash values, it's not practically possible to test it in such a manner.

[1] Statistics in a Nutshell, Sarah Boslaugh

Feature Hashing

Here's a cool little trick called feature hashing. The idea is that you want to turn some corpus of text into feature vectors without resorting to an external repository for the indices in those vectors. Looking things up in an external repository can be slow and complicates the code base.

Instead, you avoid this through hashing the words to an integer. Spark uses the HashingTF class for this.

However, the Spark code is very dependent on all your words being able to fit into positive Ints. This is probably OK for most corpus of text but when you start using n-grams ("Groups of words in a split are known as n-grams." [1]), then the numbers start to approach what an Int can hold.

But what's more worrying is that hashing leads to the possibility of hash collisions. How many you expect can be calculated from the formula here. Basically, given M possible slots and N words that need fitting into them, the probability of a word going into a particular slot, z, is 1/M. Therefore, the probability of avoiding a slot is:

p(miss z) = 1 - 1 = M - 1 = Y
                M     M

The probability that we miss z for all N words as we add them to the slots is Y^N. So, obviously, the probability that the slot receives at least one word is 1-Y^N.

This is a formula for the probability of a slot being filled but what is the expected number of collisions? Here, we note a nice trick outlined here.

The expectation of a sum is the sum of the expectations. This theorem holds "always." The random variables you are summing need not be independent.

Now, we introduce a function that is 1 if slot z contains one or more words and 0 otherwise. Note: we're not interested in the number of words in the slot. We're only interested if the slot contains something.

We call this function for slot z, f_z.

So, the expected total number of slots that hold at least one word is:

E(X) = E(f₀) + E(f₁) + E(f₂) + ... + E(f_M) 

The expected value of anything is the sum of all possible values multiplied by their probability. So, the expected value of an individual f for any of the M slots is:

E(f_z) = 0 * p(f_z=0) +  1 * p(f_z=1) = p(f_z=1)

(Note that the expected value need not be an integer nor indeed sensible. The expected number when throwing a 6-sided die is 3.5.)

Given M slots, the expected number of slots that contain something is:

E(X) = M (1-Y^N)

So the total number of collisions is the difference between this and N as the Quora link says.

For a spike, we used the MurmerHash3 hashing algorithm over 820 million n-grams and the number of slots that had more than one word was 70 million (close to the 74 million this equation predicts when M = 2^32 and N = 820 million).

So, although it's a nice idea, Spark's current implementation is not for us. We'll look at rolling-our-own using Longs for which we could reasonably expect no collisions (using the above equation with M = 2^64).

[1] Real World Machine Learning.

Cuckoo Hashing

I was taking a look at a library called Kryo that is being used on a project I'm working on. It promises to be a much faster way of de/serializing objects.

Trying to discover it's secrets, I came across a class called com.esotericsoftware.kryo.util.ObjectMap. It's an implementation of a map that claims to be very fast for get, containsKey and remove although it maybe slower for put.

The way it does this is by using a technique called Cuckoo hashing. Here, we have an array of keys and an array of values. Upon insertion, we calculate a hash for our key and see if the value slot is free. If it is, we insert the value and we're done. If not, we may repeat this process for N more times using a different hash. Finally, failing this, we check a fixed size of memory called the stash. I'll return to this in a moment.

If we have failed to find any space for our mapping to go, an old key/value mapping is evicted to make way for our new key/value pair. This is where there reference to cuckoos comes from (cuckoo chicks eject their foster siblings from the nest).

The old key/value pair is put into the stash I mentioned earlier. This is a fixed-size area of memory. Only when this is exhausted do the arrays resize.

Incidentally, because the stash is fixed-size, this algorithm qualifies as an in-place algorithm.

Running tests using both Kryo's ObjectMap and Java's HashMap each initialized with space for 10 000 slots and running the experiment 10 000 times gave these results (all times in milliseconds).

	Kryo ObjectMap			Java HashMap
Number of Elements	Put	Get	Remove	Put	Get	Remove
2000	636	96	126	379	295	353
4000	1372	401	704	716	569	709
8000	3384	1763	1628	2184	1269	1585

True to their word, the author appears to have implemented a very fast map when it come to get and remove. The only caveat is that as the data structure approaches any where near capacity, it slows down.

As an aside, the stash is searched using a technique called "linear probing" or "open addressing". Here "we just add the key to an [array] index but in linear probing, we put it in position i if that's free and if not look at position i+1, i+2... It works well if the size of the array is significantly bigger than the number of keys" [1].

[1] Prof Bob Sedgewick, Coursera.