\documentclass[11pt]{article}
\usepackage{geometry}
\geometry{letterpaper}
\usepackage{graphicx}
\usepackage{amsmath}
\usepackage{amssymb}
\usepackage{epstopdf}
\usepackage{enumerate}
\usepackage{url}
\usepackage{verbatim}
\usepackage{moreverb}
\usepackage{epsfig}
\usepackage{listings}
\usepackage{color}
\usepackage{hyperref}
\usepackage{algorithmic}
\usepackage{amssymb}
\usepackage{hyperref}
%\usepackage{biblatex}

\newcommand{\minus}{\scalebox{0.75}[0.75]{$-$}}
\newcommand{\newcontent}[1]{{\color{blue}{\bf\sf [newcontent: #1]}}}

\definecolor{dkgreen}{rgb}{0,0.5,0.3}
\definecolor{gray}{rgb}{0.5,0.5,0.5}
\definecolor{mauve}{rgb}{0.58,0,0.82}

\lstset{frame=tb,
  language=Java,
  aboveskip=3mm,
  belowskip=3mm,
  showstringspaces=false,
  columns=flexible,
  basicstyle={\small\ttfamily},
  numbers=none,
  numberstyle=\tiny\color{gray},
  keywordstyle=\color{blue},
  commentstyle=\color{dkgreen},
  stringstyle=\color{mauve},
  breaklines=true,
  breakatwhitespace=true,
  tabsize=3
}

\title{10-605 HW 7: Gibbs Sampling LDA using a Parameter Server}
\author{Due Tuesday, December 6, 23:59 via Autolab}
\date{Out Tuesday November 29, 2016}

%\bibliography{references.bib}

\begin{document}

\maketitle

\section*{Policy on Collaboration among Students}

These policies are the same as were used in Dr. Rosenfeld's previous version of 10601 from 2013. The purpose of student collaboration is to facilitate learning, not to circumvent it. Studying the material in groups is strongly encouraged. It is also allowed to seek help from other students in understanding the material needed to solve a particular homework problem, provided no written notes are shared, or are taken at that time, and provided learning is facilitated, not circumvented. The actual solution must be done by each student alone, and the student should be ready to reproduce their solution upon request. The presence or absence of any form of help or collaboration, whether given or received, must be explicitly stated and disclosed in full by all involved, on the first page of their assignment. Specifically, each assignment solution must start by answering the following questions in the report:

\begin{itemize}
\item Did you receive any help whatsoever from anyone in solving this assignment? Yes / No.
If you answered `yes', give full details: \underline{\hspace{3cm}} (e.g. ``Jane explained to me what is asked in Question 3.4'')
\item Did you give any help whatsoever to anyone in solving this assignment? Yes / No.
If you answered `yes', give full details: \underline{\hspace{3cm}} (e.g. ``I pointed Joe to section 2.3 to help him with Question 2''.
\end{itemize}

Collaboration without full disclosure will be handled severely, in compliance with CMU's Policy on Cheating and Plagiarism.
As a related point, some of the homework assignments used in this class may have been used in prior versions of this class, or in classes at other institutions. Avoiding the use of heavily tested assignments will detract from the main purpose of these assignments, which is to reinforce the material and stimulate thinking. Because some of these assignments may have been used before, solutions to them may be (or may have been) available online, or from other people. It is explicitly forbidden to use any such sources, or to consult people who have solved these problems before. You must solve the homework assignments completely on your own. I will mostly rely on your wisdom and honor to follow this rule, but if a violation is detected it will be dealt with harshly. Collaboration with other students who are currently taking the class is allowed, but only under the conditions stated below.

\section{Important Note}
Yulan Huang (yulanh@andrew.cmu.edu), and Jingyuan Liu (jingyual@andrew.cmu.edu) are the contact TAs for this assignment. Please post clarification questions to Piazza, and the instructors can be reached at the following email address: \texttt{10605-Instructors@cs.cmu.edu}.

\section{Background}

\subsection{Parameter Server and JBosen}

Parameter server (PS) is an abstraction for distributed, data-parallel ML. Each worker holds a partition of the dataset but has access to the full model parameters stored on the server machines through a fully connected bipartite network topology (Fig.~\ref{fig:ps}). ML application programmers are presented with a distributed shared memory (DSM) abstraction through a key-value store interface. In Petuum's Bosen PS we expose a table interface where each row is accessed via a row ID. Note that servers and workers are logical components. Since a worker uses more CPU while a server uses more network, in practice it is often convenient to co-locate a server and a client on each node to increase resource utilization, which is currently what we do in Bosen.

In this assignment we will implement Gibbs sampling for LDA using JBosen, a minimal Java version of Bosen. JBosen exposes the same interface as Bosen, which is a hash-table view into the parameter server. Each table maps row IDs (integers) to rows, where each row contains columns of elements. Rows can be either sparse (implemented using a map) or dense (implemented using an array), depending on the performance requirements of the application. Custom rows can also be defined, but this is a more advanced feature that we will not use in this assignment. A table has two main API methods:
\begin{itemize}
\item \texttt{table.get(row\_id, col\_id)} retrieves a value from the table. This value might be a stale value within the staleness bound set by the application.
\item \texttt{table.inc(row\_id, col\_id, i)} increments a value in the table by \texttt{i}. Because of Stale Synchronous Parallel (SSP) execution, this change may not immediately be reflected in the table. Even when using BSP, the change will only be reflected in the table on the next iteration.
\end{itemize}
anyJBosen also provides two important methods that are not associated with  table:
\begin{itemize}
\item \texttt{PsTableGroup.clock()} informs JBosen to proceed to the next clock/iteration. JBosen will take care of managing consistency so that values in the table are not more than \texttt{s} clocks out of date. More specifically, when a worker thread calls clock(), the system will:

1) Send all updates applied to tables by the thread to the servers. (Whenever you call table.inc(), the update is actually cached locally until you call clock());

2) Block until all other worker threads have reached clock $c - s + 1$, where c is the clock at which clock() was called for the current thread, and s is the SSP staleness bound. Doing this maintains proper SSP execution.

\item \texttt{PsTableGroup.globalBarrier()} forces tables to be completely synchronized. After this call exits, all workers will be at the same spot in the code and will have the freshest values in the tables.
\end{itemize}
You may find the source code for JBosen, as well as detailed documentation, at \url{https://github.com/petuum/jbosen}.

\begin{figure}[ht]
    \centering
    \includegraphics[width=0.6\textwidth]{ps.png}
    \caption{Parameter Server logically consists of a set of server nodes and worker nodes forming a fully connected bipartite graph.}
    \label{fig:ps}
\end{figure}

\subsection{Stale Synchronous Parallel (SSP)}

Since the network is much slower than main memory reference ($10^4\sim 10^5$x), reading parameters off server nodes for every access is a big bottleneck. We can cache them by storing a local copy in the main memory. There are various ways to synchronize different copies of parameters. Bulk-synchronous parallel (BSP) is a popular one used by Hadoop/Spark. However, BSP could suffer from stragglers and does not overlap communication and computation time effectively. Fully asynchrounous is fast but has no theoretical guarantee. This is where bounded delay consistency models come to the rescue.

JBosen implements Staleness Synchronous Parallel (SSP)~\cite{ho2013more}. The SSP consistency model ensures bounded staleness when reading the model state from the PS (Fig.~\ref{fig:ssp}). Specifically, given a worker at logical clock $c$, reading any parameter must return a value that has been updated by all updates computed before and during clock $c - s - 1$, where $s$ is the staleness threshold. Such a guarantee is provided by using a vector clock (one clock per worker) for each parameter in the client library. To reduce the memory overhead of one vector clock per parameter, we restrict clock to be called on all parameters and therefore only one vector clock is needed. When PS cannot provide parameter satisfying the staleness requirement (e.g,. a straggler is more than s clocks behind), the reader is blocked until the parameter is properly updated.


\begin{figure}[ht]
    \centering
    \includegraphics[width=0.8\textwidth]{SSPworker.pdf}
    \caption{An execution of PS under Stale Synchronous Parallel (SSP). Here staleness is 3, black and green denotes the updates that are visible to worker 2 (green updates are visible due to read-my-write consistency), while blue updates {\it could} be visible by worker 2 but is unspecified under SSP. Worker 2 is blocked at clock 5 because worker 1 hasn't finished clock 1, violating the SSP constraint. Specifically, worker 2 is blocked because updates from worker 1 clock 1 is not visible to worker 2 yet, and thus worker 2's read at clock 5 is blocked.}
    \label{fig:ssp}
\end{figure}






\subsection{LDA with Gibbs Sampling}

Just a quick review of the problem and algorithm. Latent Dirichlet Allocation is a generative model that represents set of documents as mixtures of topics or clusters. In case of textual data, the likelihood of occurrence of words in the entire corpus can be well represented by a multinomial distribution. The Dirichlet distribution being a conjugate of the multinomial distribution blends well with the overall generative story of the mixture models in Latent Dirichlet Allocation. The plate notation is shown in Fig.~\ref{fig:lda}. $\phi$ captures word-topic distribution for each topic k $\in$ K, $\theta$ captures document-topic distribution for each document d $\in$ M, z is the latent variable which estimates the probability of a word w in a document lying in cluster k. The probability distribution models and a Gibbs sampler over all the variables are described below

\begin{algorithmic}
  \STATE $n_{kv},n_{dk} \gets [][], n_k,n_d\gets[]$
  \FORALL{documents $m \in [1,M]$}
   \FORALL{words $n \in [1,N_m]$ in document m}
    \STATE sample topic index $z_{m,n} = k \thicksim Multi(1/K)$
    \STATE increment document-topic count : $n_{dk}[d,z]$ + 1
    \STATE increment topic-word count : $n_{kv}[z][n]$ + 1
    \STATE increment topic-word sum : $n_k[z]$ + 1
	\ENDFOR    
  \ENDFOR
  \WHILE{$t \leq |iterations|$}
   \FORALL{documents $m \in [1,M]$}
     \FORALL{words $n \in [1,N_m]$ in document m}
  	  \STATE $z \gets z_{m,n}$
      \STATE $n_{dk}[d,z] \minus\minus $
      \STATE $n_{kv}[z,n] \minus\minus $
      \STATE $n_k]k] \minus\minus $
      \STATE $ p(z_i=k|z_{\neg i, words}) = \frac{n_kv[][n]+\beta}{n_k + \beta * V } (n_{dk}[d][]+\alpha) $
      \STATE $p(z_i=k|z_{\neg i, words}) /= \sum p(z_i=k|z_{\neg i, words}) $
  	  \STATE $z \gets$ sample a topic from posterior p(z)
      \STATE $n_{dk}[d,z]++$
      \STATE $n_{kv}[z,n]++ $
      \STATE $n_k]k]++ $
     \ENDFOR    
	\ENDFOR
  \ENDWHILE

\end{algorithmic}

\begin{figure}[ht]
    \centering
    \includegraphics[width=0.6\textwidth]{LDA.png}
    \caption{Plate notation for Latent Dirichlet Allocation.}
    \label{fig:lda}
\end{figure}

\section{Getting Started}

You are provided with a starter kit for this assignment, which takes care of boilerplate code, and provides scripts to compile and run your implementation. It also contains a dataset you can use to test your code. You can find it at \url{http://curtis.ml.cmu.edu/w/courses/images/7/70/Lda.zip}.
\begin{itemize}
\item {\bf Boilerplate code.} You are provided with two complete Java files, \texttt{Lda.java} and \texttt{DataLoader.java}. \texttt{Lda.java} provides some initialization code and \texttt{DataLoader.java} provides code for loading data. You are also provided with an incomplete Java file, \texttt{LdaApp.java}, which has some initialization and code stubs. This is where you should complete your implementation. By default, the provided code reads data from \texttt{20news.csv} and outputs the word-topic table to \texttt{word-topic.csv}, as well as the likelihood per iteration data to \texttt{likelihood.csv}. These can be changed using arguments to \texttt{lda\_run.py}.
\item {\bf Gradle script.} The starter kit contains a stand-alone version of Gradle, which is a Java dependency management and compilation framework. Simply run
\begin{verbatim}
./gradlew fatJar
\end{verbatim}
to compile your program. This will automatically download dependencies, including JBosen, compile your program, and puts everything into a single jar file.
\item {\bf Run/kill scripts.} \texttt{lda\_run.py} and \texttt{lda\_kill.py} make it easy to test your program using multiple nodes. After compiling your program using Gradle, simply run
\begin{verbatim}
python lda_run.py
\end{verbatim}
to launch a set of nodes specified in \texttt{hosts.txt}. \texttt{lda\_run.py} also handles the command line arguments passed to your LDA implementation. For more information,
\begin{verbatim}
python lda_run.py --help
\end{verbatim}

{\bf Hint:} If you are getting an ``address already in use'' error, try using \texttt{lda\_kill.py} to terminate any left-over processes first. If \texttt{lda\_kill.py} does not work, SSH to each machine and run \texttt{killall java}.
\item {\bf Host file.} \texttt{hosts.txt} is the default location that \texttt{lda\_run.py} looks for a list of hosts. Each line of the file specifies an endpoint, in the form \texttt{<ip>:<port>}, that you wish to launch a JBosen node on.

{\bf Hint:} If you are testing with multiple processes on the same machine, make sure to space out the ports used. JBosen may bind to several ports adjacent to the port specified.
\item {\bf Dataset.} \texttt{20news.csv} and \texttt{vocab.csv} contain a processed version of the 20 newsgroups dataset, with stop words and E-mail headers removed. Words in \texttt{20news.csv} are expressed as the index of the word in \texttt{vocab.csv}, so that \texttt{0} represents the first word in \texttt{vocab.csv}, \texttt{1} represents the second word, and so on. You may use this dataset to test your program and it will also be used in the questions you will answer.

 \item {\bf AWS Experiments.} Please follow the guidelines given here \url{https://goo.gl/fVTzI5} for running experiments on AWS. If you have almost used up your AWS credits, please send an email to Bhuwan (bdhingra@andrew.cmu.edu) with your detailed AWS credit usage, including left credits and which homeworks you spent the credits on. You may get more credits based on this.

\end{itemize}

\section{Tasks and Questions}

%TODO: Write this section.
%The first task is graded by Autolab. Submit your L and R csv file from task T1 as well as \texttt{MatrixFactCore.java}. Answer the rest (Q2-6) in the report file \texttt{[andrewid]-report.pdf}. All experiments except T1 use 4 instances of AWS EC2 m3.xlarge. To minimize time wasted on EC2, it's recommended to complete all problems and test run on two micro instances and run all experiments on the same set of 4 EC2 nodes in one go (EC2 is charged by hour, and each experiment should only run for about 10 minutes). Again, detailed guide for setting up experiments can be found at \url{http://bit.ly/605_hw8_guide}.

%\subsection{General guide}

%TODO: Write this section.
%The parameters are listed in Table~\ref{tab:params}. The learning rate uses $\eta_t = \eta_0 \alpha^t$ where $t$ is epoch, i.e., the number of sweeps through the dataset, and $0 < \alpha \le 1$ is the decay rate (\texttt{learning\_rate\_decay} in the script). While multiplicative decay doesn’t satisfy theoretical requirement (sum of learning rate does not diverge), but in practice I find it easier to tune than square root rule ($\eta_t \sim O(\frac{1}{\sqrt{t}})$). Feel free to try other rule and tune yourself. For T1, we use a small 1k$\times$1k dataset (41981 ratings, 1k users, 1k movies) that's in the git repo when you clone. For T2-3 we will use a bigger dataset that can be downloaded at \url{http://bit.ly/netflix_10m} (10m ratings, 13706 users, 17770 movies). Both dataset have contiguous user and movie ID starting from 0.

\begin{enumerate}[1.]

\item {\bf Implementation} Starting from the code in the starter kit, complete the implementation of LDA Gibbs sampling using the algorithm described above. To test your program, you may use the dataset provided with the starter kit. 

\item {\bf Topics learned by LDA [20 pts]} Run your implementation of LDA on the dataset provided in the starter kit, \texttt{20news.csv}, using \( 20 \) topics, \( \alpha = 0.1 \) and \( \beta = 0.1 \). You should run it using \( 1 \) worker with \( 1 \) thread for \( 100 \) iterations, \( 25 \) clocks per iteration, with a staleness bound of \( 0 \).
\begin{enumerate}[(a)]
\item Report the plot log-likelihood values vs. the iteration number. In your plot, logarithmically scale the likelihood.
\item For each topic learned by your implementation, list the 5 words with the highest probability. You may use \texttt{vocab.txt} to retrieve the textual words. The actual topics of each newsgroup can be found at \url{http://qwone.com/~jason/20Newsgroups/}. Can you identify any of these from the topics you learned (you should be able to identify a few of them)? If so, which ones? As a sanity check for your results, the top 3 terms learned by our solution for a selection of topics are:
\begin{itemize}
\item bike, car, bmw
\item hocky, season, teams
\item religion, atheism, atheists
\end{itemize}

\end{enumerate}

\item {\bf Effect of staleness [20 pts]} In this question, you will explore how the staleness bound affects the convergence rate of LDA. Run five experiments, using \texttt{staleness} settings 0, 2, 4, 8, and 16 on 4 m3.xlarge instances. Use the following settings for running your experiments: 
\begin{itemize}
\item \texttt{alpha = 0.1}
\item \texttt{beta = 0.1}
\item \texttt{num\_topics = 20}
\item \texttt{num\_iterations = 30}
\item \texttt{num\_clocks\_per\_iteration = 25}
\end{itemize}
% On 4 m3.xlarge instances, obtain the convergence curves for 30 epochs with varying staleness using the 10m dataset (Table~\ref{tab:params}, T2). Plot and clearly label (a) loss vs epoch, and (b) loss vs seconds. Make sure your plot is not thrown off by diverging runs (y-axis should roughly zoom at $6\times 10^6\sim 7\times 10^7$). Answer the followings:
\begin{enumerate}[(a)]
\item Plot the log-likelihood values vs. iteration number. Which staleness setting converges the fastest per iteration?
\item Plot the log-likelihood values vs. the total time in seconds. Which staleness setting converges the fastest per second?
\item Explain the reason behind your results in terms of freshness of parameters, overhead of synchronization, and convergence per iteration.
\item What results would you expect if the application is more sensitive to staleness? Less sensitive to staleness?
\end{enumerate}

\item {\bf Effect of clock frequency [20 pts]} In this question, you will explore how the frequency of synchronization affects the convergence rate of LDA. Run five experiments on 4 m3.xlarge machines, using \texttt{num\_clocks\_per\_iteration} settings 1, 5, 10, 50, and 100. 
\begin{itemize}
\item \texttt{alpha = 0.1}
\item \texttt{beta = 0.1}
\item \texttt{num\_topics = 20}
\item \texttt{num\_iterations = 30}
\item \texttt{staleness = 0}
\end{itemize}
\begin{enumerate}[(a)]
\item Plot the log-likelihood values vs. iteration number. Which clock frequency setting converges the fastest per iteration?
\item Plot the log-likelihood values vs. the total time in seconds. Which clock frequency setting converges the fastest per second?
\item Explain the reason behind your results in terms of freshness of parameters, overhead of synchronization, and convergence per iteration.
\item What results would you expect if the application is more sensitive to staleness? Less sensitive to staleness?
\end{enumerate}

%\item {\bf Data Movements. [20 pts]} In JBosen we only move the model, but fix the data on the workers. Let's say we have $P$ servers and $P$ workers on $2P$ nodes. Consider 1 unit of model parameters split across $P$ servers such that each server has $\frac{1}{P}$ of the parameters. Answer the followings:

%\begin{enumerate}[(a)]
%\item If each worker node needs the full model parameters, how many units of network traffic is each server machine in one model broadcast?
%\item If each worker only needs $\frac{1}{P}$ unit of parameters (which can happen if data is sparse and thus $\frac{1}{P}$ dataset need a partition of the parameters), how many units of network traffic does each server has in one model broadcast?
%\end{enumerate}

%Our experiments use Netflix dataset and rank $K=1000$ for the factor matrices on 8 nodes, and the parameters are partitioned equally on 8 server nodes which are collocated with the worker nodes (i.e., the 8 EC2 instances are both clients and servers). Assume that each worker access $\frac{1}{4}$ of the parameters and  How much traffic 

%\item {\bf Checkpointing [20 pts]} Bosen uses checkpointing for fault tolerance. Young\cite{young1974first} derived a first-order approximation to the optimal checkpoint interval:
%\begin{align*}
%T_{\text{interval}} = \sqrt{2 T_{\text{checkpoint}} T_{\text{MTBF}}}
%\end{align*}
%where $T_{\text{checkpoint}}$ is the time it takes to perform checkpoint, and $T_{\text{MTBF}}$ is the mean time between failure for the experiment machines. Assuming a per-machine MTBF to be 1 year, independent for each node, and it takes 0.1 hour to do checkpointing.
%\begin{enumerate}[(a)]
%\item How often, in hours, should we need to perform checkpoint with 4 machines?
%\item Do we need to checkpoint any of the experiments you've run so far?
%\item MapReduce essentially checkpoint at every phase. According to this approximation, do you think it's necessary in most use cases?
%\end{enumerate}

\item {\bf Collaboration Policy} Answer the questions in the collaboration policy on page 1.

\end{enumerate}

\section{Deliverables}
Set the argument \texttt{num\_local\_worker\_threads} to 4 and submit the four likelihood files generated. Submit a tar file containing 5 files: \texttt{LdaApp.java, likelihood*.csv} all at the top level directory via AutoLab to HW7: LDA. Submit your {report.pdf} to HW7: Report. You should implement the algorithm by yourself instead of using any existing machine learning toolkit.


%\section{Marking breakdown}
%You will be graded based on the runtime (40 points) for your code, and the correctness of your implementation (40 points). The correctness of your implementation will be reflected by the precision, and recall of the pages in your generated subgraph, as well as the corresponding values’ Spearman’s correlation. The report will be graded manually (20 points).

% \printbibliography

\bibliographystyle{plain}
\bibliography{./references}

\end{document}

