The four major stages in our pipeline were as follows: upon reducing our dataset, we first had to pre-process our text; this included removing stop words such as "and" or "the" and lowercasing. We then could extract the unique words from the text, otherwise known as a dictionary. However, for most purposes, we decided to use a fixed-length dictionary of the top 20 or 100 words across all news articles to save computations. The next step was to create an occurrence matrix, showing which words appear in which article, which was multiplied with itself to achieve a co-occurrence matrix for each article. We created numerous pipelines to process each of these four steps in parallel, which was a permutation of four technologies we learned in the CS205 module: Spark, MapReduce, OpenMP, and MPI, as well as a sequential implementation/use of a fixed value to generate the dictionary, as explained above. The pipelines we used were as follows:

MapReduce and NLP Text Preprocessing

We first explore using MapReduce to do NLP text preprocessing on the news articles. The NLP preprocessing tasks include removing duplicated news articles and news articles that only contain a URL, removing punctuations, converting sentences to lowercase, tokenization and removing stopwords. The mapReduce workflow is straightforward in the NLP text preprocessing stage. All the data preprocessing code is implemented in the mapper and the mapper will do most of the data preprocessing. The reducer mainly serves the role to remove duplicated news articles in the data set and emit a corpus of cleaned news articles.



We can see that as we increase the number of worker nodes, the running time gradually decreases. With a 4-worker node(m4.xlarge) architecture in AWS, MapReduce is able to preprocess all the 1 millions news articles within one minute.

At the same time, however, it is worth mentioning that there are two things that are not so desirable for using MapReduce to do NLP data preprocessing. First of all, many MapReduce service providers do not support running MapReduce jobs with pre-built NLP packages such as NLTK. As a result, developers will have to manually implement NLP processing code in the mapper or reducer instead of directly calling a package. Such manual implementation is simple for basic tasks such as removing stopwords, but not so easy to do with more advanced NLP techniques such as lemmatization.

MapReduce and Creating Top Word Dictionary

The second thing we explored with MapReduce is to create the word dictionary that contains the most frequent words in the entire news corpus. Through our performance testing, we realized that using MapReduce is not efficient for this task. There are two major overheads for using MapReduce in this stage.

The first overhead is that this task will require two passes of MapReduce. The first pass is to perform a word count for each word in the cleaned news corpus. The second pass is then to calculate the most frequent words based on the output from the first pass. This would be very time-consuming because the first pass will first output the data to a S3 budget, and then the second pass will need to read the data from the S3 bucket, and upload the data to its own HDFS before the second pass can start. The I/O cost here is very high.

Another major overhead is that only one reducer can be used during the second pass to generate the most frequent words among the entire news corpus. If we use more than one reducer in the second pass, each reducer will be generating its own most frequent words based on the inputs sent from their corresponding mappers.



As a result, we believe that MapReduce does not seem to be a good choice in creating the top word dictionary. We will explore other alternatives in the following parts.

Spark Data Workflow

After MapReduce, we also explored using pySpark to implement our data workflow. We found that Spark is much more straightforward than MapReduce. We are able to do NLP-data preprocess, create the top words dictionary, and create the word count matrix all in the same spark program step by step.

Spark also has the well-developed Spark Machine Learning Library(MLlib) where many NLP techniques are already implemented. We are able to directly call these methods in our program instead of implementing them manually as in MapReduce.



With a 6-node(m4.xlarge) Spark architecture on AWS, we are able to complete preprocessing all the 1 million news articles, searching for the most frequent words for top words dictionary, and creating the word count matrix based on the dictionary in as few as around 73 seconds. This is a significant improvement over the MapReduce workflow.

The last step in creating the co-occurrence matrix is the matrix multiplication of the word count matrix with itself. Through our research, however, Spark is not a stand tool for scalable matrix multiplication. Using Spark for matrix multiplication is still an ongoing research in the academic field. As a result, we decided to use OpenMP and MPI for matrix multiplication during the last stage of our data pipeline.

OpenMP Occurrence Matrix

Using the dictionary generated by Spark, we used OpenMP to parallelize the generation of the occurrence matrix. We read the articles from a file and parallelized the loops used to lookup each word in the dictionary. We tested this code for two AWS instances: t2.2xlarge and m4.xlarge, with different numbers of threads. The t2.2xlarge instance has eight cores, and the m4.xlarge has four, so we should see this reflected in the performance. We see the results below.



For one thread, the results were as expected. However, we see some very strange performance as we move to more than one thread. The program actually takes longer to run with multiple threads than it does with one, and the runtime increases as we increase the number of threads. We see this for both AWS instances. The jump in runtime from 4 threads to 8 threads is the most surprising. A calculation that we could do in a few seconds on one thread takes over 6 minutes with 8. We tried for a significant amount of time to debug this but could not solve it before the project was due. We have several hypotheses. OpenMP does create overhead, but the overhead is not at the scale that we saw in the runtime. There could be some race condition that we could not find that is causing the increase in runtime. We could also be repeating calculations somehow, making the multiple threads redundant. There could also be some issue with memory. We would like to explore this more and find out what the problem is, but this was very confusing to us.

OpenMP Co-Occurrence Matrix

Next, we use OpenMP to calculate the co-occurrence matrix, which is the matrix transpose multiplied by the original matrix. Again, we use the same two AWS instances and the same variation of threads, and see the results below.



Here, we see the performance that we would expect. As we increase the number of threads, the runtime decreases by about what we would expect. We do not see perfectly linear speedup, as the speedup from one thread to eight is only about two. However, the dictionary size we are using here is very small, and the overhead created by OpenMP likely causes us to see much less than linear speedup.

The m4.xlarge seems to perform better than the t2.2xlarge by a pretty significant amount. We could not find exact numbers for the t2.2xlarge to compare, but it seems like the m4.xlarge must have a faster processor than the t2.2xlarge. We also see that the runtime between 4 threads and 8 threads is about the same for the m4.xlarge, which makes sense as this instance only has 4 cores.

C++ Top Word Dictionary

In addition to MapReduce, we used C++ to generate the dictionary. Doing this was fairly simple, and done in two steps similar to the MapReduce version. We use a hashmap to record the frequency of each word in the dataset, then sort the words by their frequency with the highest frequency words first. We then added the top n desired words to the dictionary, which was either 20 or 100 in our later tests.

We used two data structures to store the dictionary: an array, or a vector data type in C++, and a hashmap. We see the performance of the two compared here.



We see, somewhat unsurprisingly, that the hashmap is much faster than the array-based dictionary. We also see that the amount that the runtime increases as the dictionary size increases is almost negligible for the hashmap.

The C++ dictionary creation was done sequentially. We created the dictionary independently on each MPI process. The corpus file was static across all processes and the dictionary creation was deterministic, so the dictionary was guaranteed to be the same on each MPI process. It was unclear how we could use MPI to calculate the dictionary in parallel; computing the frequency of each word would be easy to do in parallel, but we were not sure how to aggregate the results on the master node as it would involve sending variable strings and the counts from each node to the master. This was not too much of a problem because even the array-based dictionary took less than ten seconds to create for a dictionary size of 100.

MPI Occurrence Matrix

Next, we create the occurrence matrix. To create this matrix, for every word we must look up its location in the dictionary so that we know which element to increment in the matrix. The data structure used matters a lot here, since a lookup in an array is O(n), but in a hashmap the lookup is O(1).

We parallelized the occurrence matrix creation using MPI. We assign each process a certain set of lines to process and then send the results back to the master node. This worked nicely since we have already computed the dictionary, and each line of the file, i.e. each article, was independent of all of the other articles, so each process could compute the rows of the matrix independently of the others then just aggregate the matrix in the end.

Below, we see the performance of the occurrence matrix for a dictionary size of 20 and 100.





We see that the trends in these two plots are identical. The hashmap is significantly faster than the array-based dictionary since the lookups are almost instant. We do see that the speedup is lower for the hashmap than the array, but this is likely because the hashmap version runs so much faster, so the speedup is overshadowed by the MPI overhead. We are already running in less than two seconds, so the speedup is not as relevant.

MPI Co-Occurrence Matrix

Next, we create the co-occurrence matrix. If we let the word occurrence matrix be X, then the co-occurrence matrix is computed as XTX. We were actually able to do this with some clever indexing that allowed us to avoid explicitly calculating XT so that we didn’t have to waste extra memory.

After computing the word occurrence matrix, we aggregate it on the master node then broadcast it back to each node, since the entire matrix is required for computing even just a small part of the co-occurrence matrix. We then let each process compute a certain number of designated rows of the co-occurrence matrix.

Below, we see the performance of calculating the co-occurrence matrix with MPI for a dictionary size of 20 on the left and 100 on the right.





We see some strange trends for the smaller dictionary size; however, all of the runtimes are less than one second, and this is likely due to the MPI overhead causing some strange behavior. For example, as we go from one to two processes, we actually see an increase in the runtime. For the larger dictionary size, we see the trends we would expect. As we increase the number of cores, the runtime decreases by a proportional amount. The speedup is not linear; with 8 processes, the speedup is only around 3.75. There are likely many optimizations that could be made to increase this performance, and the overhead limits the amount of speedup we can achieve, but these trends seem reasonable.

MPI Pipeline

We now look at the full C++ MPI pipeline. This involves the dictionary generation which is done sequentially on each process, and the calculation of the word occurrence matrix and co-occurrence matrix done in parallel. Below we compare the results for the entire pipeline, again for a smaller and larger dictionary size.





The results for the entire pipeline are pretty similar to the results for the co-occurrence matrix calculation. There is some strange behavior with the smaller dictionary, likely caused by the MPI overhead with such a small calculation. The entire pipeline runs in less than 5 seconds. We see the trends we would expect with the larger dictionary. With these results, we can also do some code profiling. We see that the runtime is very close to the runtime of calculating the co-occurrence matrix. This means that the bulk of computation is spent calculating the co-occurrence matrix. The dictionary and word occurrence matrices do take a non-negligible amount of time, but this is only a fraction of the time when we have to calculate XTX for a large dictionary size.

MapReduce

1. workflow to preprocess text data
   Input: raw news articles stored in HDFS
   Output: NLP preprocessed and cleaned news articles
   Code file name:
   mapper.py
   reducer.py
   Execution command:
   Mapreduce Streaming on AWS EMR
2. workflow to count words:
   Input: NLP preprocessed and cleaned news articles
   Output: (word, count) pairs
   Code file name:
   mapper1.py
   reducer1.py
   Execution command:
   Mapreduce Streaming on AWS EMR
3. workflow to search for the most frequent N words:
   Input: (word, count) pairs
   Output: most frequent N words
   Code file name:
   mapper2.py
   reducer2.py
   Execution command:
   Mapreduce Streaming on AWS EMR

---

Spark

Workflow to preprocess data, create top words dictionary, and generate word count matrix
Input: raw news articles stored in HDFS
Output: 1. A dictionary containing the most frequent word in the NLP preprocessed and cleaned news articles. 2. Word count matrix based on the clean news articles and the dictionary
Code file name:
sparkaws.py
Execution command:
$ spark-submit --num-executors X --executor-cores Y sparkaws.py
(assuming a csv file named data.csv containing raw news article are already put onto HDFS)

---

MPI

MPI X^TX (Gram) matrix:
$ mpicc gram_mpi.c -o gram_mpi -O3
$ mpirun -np [num processes] ./gram_mpi [input csv file] [rows of input] [cols of input]

Example:
$ mpirun -np 8 ./gram_mpi m.csv 41771 20

MPI Total Pipeline:
$ mpic++ mpi_pipeline.cpp -o mpi_pipeline -O3
OR
$ mpic++ mpi_pipeline_hashmap.cpp -o mpi_pipeline -O3

$ mpirun -np [num processes] ./mpi_pipeline [input corpus] [dictionary size]

Example:
$ mpirun -np 8 ./mpi_pipeline corpus_big.txt 20

---

OpenMP

OpenMP Word Occurrence Matrix:
$ gcc -fopenmp -O3 create_X_omp.c -o create_X
$ ./create_X corpus_big.txt > x_mat.txt

OpenMP X^TX (Gram) matrix
$ gcc -fopenmp -O3 XT_X_omp.c -o XT_X
$ ./XT_X x_mat.txt