MapReduce for Counting Word Frequencies with MPI and GPUs

As explained above, since dictionaries work on the CPU, we count the words in parallel on the CPU using the FLoops package and the command @floop, which is a generalization of typing “Threads.@threads” before a for loop in Julia and allows the for loop to be executed in parallel. We use @floop because when we tried Threads.@threads or @sync @distributed, the code became much slower. However, @floop gave the desired effect of a parallel implementation.

We create a String array of the documents, and then iterate across that array. Then, we use the replace command to remove all punctuation (so that “dog.” and “dog” count as the same word) and we use the split command to split each document into individual words.

Then, for each document, we initialize a dictionary, and we count the words in the document and their frequencies. We maintain an array of these dictionaries. After that, we initialize a large dictionary and we iterate in parallel through each of the dictionaries that were created for each document. Unlike the small dictionaries for each document, whose key value pairs were (string, count), the large dictionary has key value pairs (string, array of counts). This means that no addition is taking place in the map stage, but rather, the same keys are being grouped with each other for a fast reduce later on.

This code is implemented in the GPU_MapReduce.jl file, since even though the reduce phase in that file is implemented on the GPU, the map phase still occurs on the CPU.

For reducing on the CPU, we modify the code from mapping on the CPU ever so slightly. As described above, we add the counts together when populating the dictionary, ensuring that the CPU is doing the additions rather than outsourcing that to the GPU. In some ways, this is the most basic attempt at MapReduce, but we can still exploit parallelism on the CPU through the @floop macro just like in the map phase. This code can be found in the CPU_MapReduce.jl file.

We used the v100 GPU available on JuliaHub, which includes 8 vCPUs.

Another way is simply to make the big dictionary have the words as keys and an array of the counts from each of the documents that the word appeared in as values. For example, if we are parsing three documents, and the word “the” appeared 10, 4, and 3 times respectively, then the big dictionary could either store (the, 17) or it could store (the, [10, 4, 3]).

In order to reduce on the GPU, we first need to convert the vectors of the counts for each key into a CuArray. In Julia, a vector of vectors cannot be converted into a CuArray, but a matrix can. Since not all of the vectors have the same lengths, we implement this by first initializing an array of zeros on the CPU that has as many rows as the number of keys and as many columns as the length of the longest vector of counts. Then, we populate the CPU array by filling in each row with the corresponding accounts. After that, we convert the array into a CUDA Array.

We also tried simply initializing a CUDA array of zeros and then populating each row with a CuArray of each of the vectors, but that took 3 times longer, so we do not pursue that approach. In the next section, we apply these algorithms and measure their timings and also see what the word frequencies are in Presidential speeches.

Here, we explore the timings for the CPU map phase and compare CPU vs GPU reduce timings. In the table, $n=1$ corresponds to reading in 4 documents: the (first) inauguration speeches given by Presidents Bush, Obama, Trump, and Biden. Then, for larger values of $n$, we simply take $n$ copies of each of these 4 inauguration speeches (so $n=100$ corresponds to counting the frequencies for $4n=400$ documents, where there are $n$ duplicates of each speech).

The CPU we use here is the 32 thread machine on JuliaHub, while the GPU is the v100 on JuliaHub. All timings in the following subsections were calculated by calling @btime from the BenchmarkTools package in Julia. Below is a table summarizing the results:

The graph below shows the time to do Map and Reduce on the CPU (third column above) vs the time to Map on CPU plus reduce on GPU (sum of second and fourth columns above) as $n$ gets larger:

We see that for small values of $n$ the speed up that the GPU offers for computing the additions is almost outweighed by the time it takes to initialize an array of zeros, populate it with the counts, and transfer that array to the GPU as a CuArray. However, this is not a concern because the MapReduce algorithm was created to work at scale on large numbers of documents at a time, so it is not relevant that it may not be as fast on the GPU when we are only dealing with small numbers of documents. In particular, we see that as $n$ reaches $250$ or $500$ (which corresponds to $1000$ or $2000$ documents), the GPU speedup is quite significant; in the case of $n=500$, the GPU allows the MapReduce operation to occur 138.005 - 99.963 - 18.911 = 19.131 milliseconds faster.

However, despite the significant speedup that the GPU offers, we can see that the bulk of the time taken still occurs in the Map phase on the CPU. Running the Map phase on a machine with more than 32 threads available would allow the @floop parallelization to speed up the code more. Alternatively, this motivates the construction of an optimized dictionary on the GPU, an idea that we discuss further in Section 5.2: Future Research.

If we purely time the call to the CUDA kernel for the reduce phase, and we disregard the time it takes to initialize the array and transfer the memory to the GPU, we get a time of about 4.29 microseconds independent of the above values of $n$. Thus the main time sink for the GPU implementation is transferring the memory from the CPU to the GPU.

Up until now, we have not studied Julia’s own MapReduce implementations. There is one for the CPU, and the FoldsCUDA library has one for the GPU. In order to best compare our MapReduce implementations with Julia’s own implementations, we compare them for finding the sum of the square roots of a set of positive numbers rather than for parsing documents. We focus on the optimized GPU implementation here rather than the CPU implementations.

For this reason, we make slight adjustments to our previous implementations of MapReduce to create a new implementation for this purpose. For the CPU implementation of map, we remove the loops for document parsing, and instead we simply add the numbers together using the @floop macro previously discussed. For the GPU implementation, we modify the row_sum kernel function that we discussed previously so that it sums square roots of numbers. The reason for mapping $x\to\sqrt{x}$ was because mapping $x\to x$ or $x\to x^{2}$ results in the compiler optimizing away the loop, thereby completing the computation independent of $n$ in approximately one nanosecond. On the other hand, there is no simple closed form for adding up square roots so this is a better way to compare the timings of the different algorithms.

In order to time Julia’s GPU implementation, we simply called @btime FoldsCUDA.mapreduce($\sqrt{x}$, +, 1:n). This code can be found in the MapReduceAdd.jl file. The computations below represent the time required to compute $\sum_{i=1}^{n}\sqrt{i}$ for each value of $n$ listed in the ways described above.

We can also visualize this graphically below:

We can see from this table and graph that our GPU implementation is much slower for $n<1000,$ but once $n\geq 10000,$ our kernel is faster than Julia’s FoldsCUDA.MapReduce package. This is only one specific example, and The FoldsCUDA MapReduce implementation is much more general. It is also worthwhile noting that our GPU implementation is very flat up until $n=10^{8}$ where it spikes dramatically, and the spike is clearly visible even with a log scale on the y axis. Meanwhile, the FoldsCUDA timing increases linearly with on this log log plot, which indicates that the timing increases linearly on a non scaled plot as well. However, this still shows that a relatively simple GPU kernel can outperform the GPU MapReduce for this specific type of problem. This kernel is using atomic add to ensure that the answer is correct; this could likely be optimized further, but it still beats the FoldsCUDA implementation.

In addition to doing $\sum_{i=1}^{n}\sqrt{i}$ we can also try computing $\sum_{i=1}^{n}\frac{1}{i}\cdot(-1)^{i+1}$ through our GPU implementation and the FoldsCUDA implementation. This works because the compiler will not be able to optimize the computation away. From calculus, we know that this alternating harmonic sum converges to $\ln 2\approx 0.693147.$ If we try this, we get a similar pattern in the timings:

Curiously, we once again observe the relative consistency of our GPU sum followed by a sudden spike for $n=10^{8}.$ However, even with $n=10^{8}$, our implementation is still about 100 times faster than Julia’s FoldsCUDA implementation for this problem. In the next section, we will return to reading documents, and we will apply our word frequency MapReduce algorithms to detect patterns in Presidential speeches and what words they use.

In the previous subsections, we described the times for running the GPU algorithm for MapReduce on the CPU and GPU. We saw that the GPU gives a slight speedup, particularly when there are more documents and they are longer. Here, we focus less on the speed of implementation (since as we saw, it is only a matter of microseconds) and more on the results from running the algorithm to see if we can discern anything from the word frequencies on specific documents.

There are a myriad of possible documents that could be studied. Here, we analyze Presidential speeches from Presidents George W Bush, Barack Obama, Donald Trump, and Joseph Biden, specifically their State of the Union speeches (retrieved from [9], [6], [8], and [10]) and Inauguration Speeches. All public Presidential speeches are legally required to be made available to the public by the Presidential Records Act, allowing for easy access to them. These documents are included in the folders containing speeches for each President. Many Presidents use similar words, from discussing unity and patriotism to more common words like “the” or “our.” But are there more unique words that each US President tends to use that distinguishes them (or perhaps, their speechwriters) from their fellow Presidents?

It turns out the answer is yes. President Biden is easily identifiable by his use of the word “troops.” While every commander in chief speaks about the US military. President Biden is the only one who ends every one of his speeches with “And may God protect our troops.” The other words in this sentence are frequently used by other Presidents and native English speakers very frequently (especially words like “our” and “and”), so they are ineffective in helping identify which President gave the speech.

President Trump stands out from other Presidents from his use of the word “God.” While every President invokes God from time to time, President Trump mentioned God an average of 5 times per speech in the speeches sampled, compared to 1 or 2 times per speech for the other Presidents. This is not surprising though, as President Trump’s base included many religious evangelical Christians who would be very supportive of their President invoking God.

President Obama stands out for his use of the word “Iran,” likely in reference to Foreign Policy and the 2015 Iran Nuclear Deal. Across these speeches, President Obama used the word “Iran” 12 times, while President Bush used it 3 times, President Trump used it 5 times, and President Biden has not used it at all.

President Bush stands out by his use of the word “terrorist” in his State of the Union addresses and inauguration speeches, likely in reference to the September 11th, 2001 attacks. He used this word an average of 4 times per speech, while his successors either never used the word “terrorist” or at most once or twice.

Another interesting trend is to look at the number of times each President received an applause, which is marked by the word “(Applause)” appearing in the speech transcript. In the speeches sampled, President Bush received an applause 220 times, President Obama received an applause 210 times, President Trump received an applause 131 times, and President Biden received an applause only 17 times. The fact that President Biden received so few is likely because his speeches did not consist of as many State of the Union addresses as his predecessors since he has not given many SOTU addresses yet. Nevertheless, the fact that the number of applauses has decreased since President Bush regardless of party may suggest further political polarization, where even applauding for the President of the United States is perhaps not as unifying as it used to be during President Bush’s term in the wake of 9/11.

Because there are so many words that all Presidents and speakers use, it is not practical to code an ML method like a neural network to analyze the most common words and predict which President said them. These observations were made by considering the political climate of the time and confirming that the President’s language in his speeches matched what we would expect given the time period.

We began with an MPI implementation, which aimed to parallelize both the map and reduce stages, but we found that there were some issues with fully reducing the key value pairs. Then, we switched to a GPU implementation, where we found that the map phase is best done on the CPU but the counting/reduce phase was optimally done on the GPU. We also saw that the GPU reduce implementations of $\sum_{i=1}^{n}\sqrt{i}$ and $\sum_{i=1}^{n}(-1)^{i+1}\frac{1}{i}$ are significantly faster than the FoldsCUDA implementation on this problem, highlighting the potential for future improvement of a faster GPU MapReduce algorithm in general.

After that, we switched to applying our MapReduce algorithm to count the frequencies of words in US Presidential speeches, and we found that while the Presidents used many common words, there were also some words that were different among them, a result of the political climate of the time and the Presidents themselves.

This paper revealed several possible approaches to implement MapReduce in parallel. One way to further the research in this field would be to combine MPI to perform map reduce across multiple GPUs, adding another layer of parallelism for particularly large computations. Another would be to first implement an optimized dictionary on GPUs, and then perform most if not all of the GPU algorithm exclusively on the GPU, rather than parsing the documents with dictionaries on the CPU.

A dictionary available on the GPUs would likely optimize both the map and reduce phases: the map phase could be performed in parallel across the GPU’s thousands of cores, and there would not need to be a memory transfer from the CPU to the GPU for the reduce phase to take place, cutting the reduce phase for the large numbers of documents that we saw in Section 3 from milliseconds to microseconds.

The algorithm can also be applied to more data. Here, we focused on counting words in Presidential speeches, which is only one of countless possible applications of counting the frequencies in documents. Another set of documents to study would be transcripts of ordinary dialogue of native speakers in different languages to compare what words are often used, and whether they are the same up to translation across languages or if there are specific wors used more in certain languages than others.

In addition, there are many applications where it is crucial to add up a large number of numbers, and this ties into the MapReduce algorithm we discussed in Section 3.5 about computing $\sum_{i=1}^{n}\sqrt{i}$ and $\sum_{i=1}^{n}(-i)^{i+1}\frac{1}{i}$ for large $n$. We only used a fairly simple GPU algorithm to optimize that computation, but that GPU kernel could almost certainly be optimized significantly more and that would allow a large number of MapReduce applications to be executed faster.