% This file was converted from HTML to LaTeX with
% gnuhtml2latex program
% (c) Tomasz Wegrzanowski <maniek@beer.com> 1999
% (c) Gunnar Wolf <gwolf@gwolf.org> 2005-2010
% Version : 0.4.
\documentclass{article}
\usepackage{url}

\begin{document}
\title{Data Flow Programming Tutorial (NAIAD)}

\maketitle
\section*{Before you start}
Before you start be sure you understand the following points:
\begin{itemize}

\item The only way to access virtual machines deployed in EC2 is using ssh with a keypair. This means that you will be responsible of your own private key. Your public key can be exchanged with others as described below.

\item Do not confuse your EC2 keypair username and the username of your EC2 instance. For this lab you will choose the first one, the second one will always be \texttt{ubuntu}.

\item Be sure you know how to edit text files using the command line and editors such as \texttt{vim} or \texttt{nano}.

\item In order to know the IP address of a machine run \texttt{hostname -I}


\end{itemize} 
 
\section{Setting up an EC2 instance}
 You should have received some instructions containing a link to a tarball that holds the necessary files for this tutorial. Inside the archive you will find:
  
 \begin{itemize}
 	\item EC2 API command line tools
 	\item Security certificates for connecting to AWS
 	\item Scripts for setting up the environment
 \end{itemize}
 
 You will first need to extract the contents of the archive to a folder on your local machine. This will create a folder named ec2-bin. The \texttt{ec2-config.sh} script defines a set of environment variables that are needed in order to utilise the EC2 command line tools.

\begin{verbatim}
tar zxf <tarball>.tgz
cd ec2-bin
. ec2-config.sh
\end{verbatim}

 (NB that's a dot followed by the script)
 
You may need to alter the \texttt{ec2-config.sh} script to point to the location of the Java Runtime Environment on your machine (which is used by the EC2 API tools).
 
 The environment variables have now been loaded into the current shell session. The next step is to create a personal private key for your own cluster. The following command will request and download a private key with the filename \texttt{pk-[USER]} into the current folder.
 
\begin{verbatim}
./create-key.sh pk-USER 
\end{verbatim}

This script is executing the \texttt{ec2-create-keypair} command and dumping the result into a file called \texttt{USER}. This file contains your private key. Your public key is stored at the EC2 and it will be automatically added to new instances so you can access through SSH. Your machine will be distinguished from other students' machines using this private key. The next step is to actually provision the cluster.
  
\begin{verbatim}
./cluster-setup.sh pk-USER
\end{verbatim}
  
  
  The above command will request an "on-demand" instance and provision it with a basic Ubuntu system image. For this tutorial, only "micro" instances with minimal resources will be used. The script creates three files in the local directory:
  
  \begin{itemize}
  	\item hosts-pub - contains the public DNS hostnames of each instance
  	\item hosts-prv - contains the private (EC2 internal) hostnames of each instance
  	\item hosts-ins - contains the instance IDs utilised by the EC2 API for each instance
  \end{itemize}
  
  After the command has completed, the request will take some further time to process (usually a few seconds). 
  
  Following the successful completion of the software deployment process, we can connect to the machine. 
  
\begin{verbatim}
master=`cat hosts-pub`
ssh -i pk-USER ubuntu@\$master
\end{verbatim}
  
  You will be connected to your virtual machine, and have full sudo access. You may want to install some additional packages using \texttt{sudo apt-get install}. The nano and vim text editors are installed by default, and these can be used to perform the tasks in the tutorial. Alternatively, you can edit files locally on your machine, and copy to/from your EC2 instances using \texttt{scp -i pk-USER source destination}.
  
  Once your instance is running, you will need to install Mono to compile and run Naiad programs. Run the following while logged into your instance:
  
\begin{verbatim}
sudo apt-get install mono-complete
\end{verbatim}
  
  We can then download and build Naiad from the GitHub repository:
  
\begin{verbatim}
sudo apt-get install git
git clone https://github.com/MicrosoftResearch/Naiad.git
cd Naiad
git checkout tags/v0.4.2
sh ./build_mono.sh
\end{verbatim}

   
   To begin with, you will utilise a single EC2 instance to familiarise yourself with Naiad. Later in the tutorial, we will join the collective instances together to form a cluster.


   \section{Authoring a Naiad program}
   Documentation for the Naiad SDK library is available at the following URL (alternatively, it can be built directly from the source code, if you have access to Visual Studio).
   
   \url{http://microsoftresearch.github.io/Naiad/}
   
   To begin with, read through ``Writing your first Naiad program''. At a low level, Naiad programs consist of dataflow vertices (also known as ``operators''), which are composed into partitioned segments called ``stages''. Notice however that none of this terminology is used in the source code on that page. This is because Naiad provides a number of wrappers around these basic building blocks for the sake of convenience. A number of higher-level languages (or DSLs) are also provided within the Naiad library. 
   
The code below corresponds to the example shown in the aforementioned tutorial.

\begin{verbatim}
using Microsoft.Research.Naiad;
using Microsoft.Research.Naiad.Input;
public class TestProgram{
	static void Main(string[] args)
    {
	// 1. allocate a new dataflow computation.
	using (var computation = NewComputation.FromArgs(ref args))
	{
		// 2. define an object which accepts input strings.
		var source = new BatchedDataSource<string>();

		// 3. convert the data source into a Naiad stream of strings.
		var input = computation.NewInput(source);

		// 4.request a notification for each batch of strings received.
		var output = input.Subscribe(x =>
		{
			foreach (var line in x)
				System.Console.WriteLine(line);
		});
		// 5. start the computation, fixing the structure of the dataflow graph.
		computation.Activate();
		// 6. read inputs from the System.Console as long as the user supplies them.
		for (var l = System.Console.ReadLine(); l.Length > 0; l = System.Console.ReadLine())
			source.OnNext(l.Split());
		// 7. signal that the source is now complete.
            source.OnCompleted();
        // 8. block until all work is finished.
            computation.Join();
    	}
       }
     }
\end{verbatim}
   
   Copy and paste the code above into a file named \texttt{program.cs}. To compile the program, copy the Naiad core library built earlier into the same directory as your source code (it should be in ~/Naiad/Naiad/bin/Debug/Microsoft.Research.Naiad.dll).
   
   Then use the following command:
   
\begin{verbatim}
dmcs program.cs /reference:Microsoft.Research.Naiad.dll
\end{verbatim}

Then, you can execute the program using:
   
\begin{verbatim}
mono program.exe
\end{verbatim}

   \section{Using the Naiad library}
   Depending on the level of customisation required, there are a number of different ways to utilise the library. We will examine a few of the Naiad example programs to highlight the various approaches. Source code for the examples can be found by navigating to the following directory:
   
\begin{verbatim}
cd ~/Naiad/Examples
\end{verbatim}
   
   To begin with, look at the ``Throughput'' example (Examples/Naiad/Throughput.cs). This example is designed to test the limits of the Naiad pipeline, by pumping messages as quickly as possible between two trivial dataflow vertices. In this example, the ``producer'' and ``consumer'' vertices are created by subclassing the standard ``Vertex'' abstract class. 
   
   Examine the source code and try to answer these questions:
   \begin{itemize}
   	\item What is the purpose of the OnNotify / OnRecv methods in the producer and consumer?
   	\item What is the purpose of the MakeStage methods?
   	\item What does the VertexLocation do? What is the pattern of communication among these processes?
   	\item How could the program be modified to achieve a ``shuffle''-style communication pattern?		\end{itemize}
   
   In most cases, it won't be necessary to to construct custom vertices in this manner. Instead, a number of general-purpose dataflow vertices are available, which allow the developer to focus on the actual computation rather than the distribution mechanism. 
   
   Now look at the ``Wordcount'' example in the same directory (Examples/Naiad/Wordcount.cs). This example makes use of the C\# ``extension method'' syntax (described further on MSDN). Instead of overriding the entire class, this allows the developer to make use of a generic ``Unary vertex'' operator, and define a custom method to perform the actual computation.
   \begin{itemize}
   	\item Find the definition of ``StreamingCount''. How is it invoked? What does it do?\end{itemize}
   
   The same example is also implemented separately using ``NaiadLINQ'', in the differential dataflow examples directory (Examples/DifferentialDataflow/Wordcount.cs). LINQ (Language-INtegrated Query) is a DSL that defines a number of relational-style operations on input streams. These operators have been implemented in the Naiad framework, using the approaches defined above.
   \begin{itemize}
   \item Compare the two implementations of Wordcount - how does the LINQ query map to the extension method defined earlier?
   \end{itemize}
   
   \section{Running jobs on a cluster}
%   Notice that each example program is instantiated using a ``Computation'' object. The ``FromArgs'' method allows configuration parameters to be passed to the Naiad execution engine. It is also possible to run multiple computations using a ``Controller'' object. Run the Examples.exe program without parameters to see a listing of Naiad engine options.
%   
%   To run a multi-machine distributed computation, it is necessary to copy the executable to each machine, and launch each instance with a separate process ID (the -p option). 
%   \begin{itemize}
%   \item Find a partner and exchange your SSH public keys (in ~/.ssh/authorized\_keys)
%   \item You will need to set up a separate directory to store your program on the remote machine (to avoid overwriting each other's files). Remember to copy the Naiad DLL. 
%   \item Simply prepend \texttt{ssh <remote\_machine>} to the command line, in order to launch it on the remote machine. It may be necessary to disable firewalls on either side. 
%   \item By repeating this process with other pairs, the size of the cluster can be expanded.\end{itemize}
%   
%   This approach is somewhat cumbersome, however, Naiad also contains library hooks to allow programs to make use of Windows Azure compute and storage directly from source.
Now we are going to run Naiads using several machines. First, we need to locate our public key. In your local machine you can find your private key (remember that you created that file under name \texttt{pk-USER}). Do not share that key, that's only for you. Your public key is stored in the set of allowed keys that can access your EC2 instance via SSH. Run the following commands to get your public key:
\begin{verbatim}
USER="your username"
cat ~/.ssh/authorized_keys > ~/$USER"".pub
\end{verbatim}

Now a plain text file containing your public key should be in your home folder. Find a partner to exchange your public key with. Simply send the file via email to your partner. Your partner must add your public key to the \texttt{authorized\_keys} in his EC2 instance. To do that:
\begin{verbatim}
USER="id of my requesting partner"
cat \$USER"".pub >> ~/.ssh/authorized_keys
\end{verbatim}

Now from your local machine you should be able to SSH your partner's machine.

\begin{verbatim}
PARTNER_MACHINE="IP of your partner's machine"
ssh -i USER ubuntu@$PARTNER_MACHINE
\end{verbatim}

Now that if you can SSH other machine you can simultaneously run two NAIADS instances. Simply open another command line console in your local machine. Using your private key now you can run a distributed NAIAD workflow.

Your EC2 instance:
\begin{verbatim}
master=`cat hosts-pub`
ssh -i USER ubuntu@$master
cd ~/Naiad/Examples/bin/Debug
mono Examples.exe wordcount -n 2 -p 0 -h IP1:2101 IP2:2101
...
\end{verbatim}

Your partner's instance:
\begin{verbatim}
partner="your partner machine ip"
ssh -i USER ubuntu@$partner
cd ~/Naiad/Examples/bin/Debug
mono Examples.exe wordcount -n 2 -p 1 -h IP1:2101 IP2:2101
...
\end{verbatim}

Similarly you can run the other examples. Be sure to indicate the IPs in the same order and to indicate different process numbers (\texttt{p} parameter) for each instance.


   \section{Extensions}
   In this section we will make use of the concepts introduced thus far to design a more realistic dataflow computation. Twitter exposes a streaming API that allows developers to download public tweets as they are published. These streams are marked up using JSON syntax. The following commands will give you access to a set of files containing data recently downloaded from Twitter (you will need to substitute [SERVER] with a hostname provided by the supervisor).
   
\begin{verbatim}
apt-get install nfs-common
modprobe nfs
mkdir pubstore
mount [SERVER]:/pubstore pubstore   
\end{verbatim}

   The pubstore directory will now contain several text files, each filled with approximately 100MB of tweets in JSON format. 
   
   Using the LINQ-based Wordcount example as a template, try to write a program that calculates the most popular hash tags in the downloaded sample. You will have to do the following:\begin{itemize}
   \item Wrap the file in an input stream.
   \item Select and filter all hash tags (if present)
   \item Group and count the distinct hashes
   \item Output the top 10 (or however many you want)
\end{itemize}
   
   Once you are satisfied that it is working in LINQ, try to implement the hashtag counter using extension methods (as in the second example above). 
   
   Once you have both versions of the program working, use the method described in the previous section to run the computation on multiple machines. Have each machine process a separate chunk of the file. Compare the runtime performance of the LINQ-based and extension-method implementations.
   
   \end{document}
