\documentclass[12pt]{beamer}

\usepackage{color}
\usepackage{graphics}
\usepackage{listings}
\usepackage{xspace}

\newcommand{\nsd}{\textit{NSD}\xspace}
\newcommand{\matvec}{\textit{mat-vec}\xspace}
\newcommand{\matvecs}{{\matvec}s\xspace}
\newcommand{\MatVec}{\textit{Mat-Vec}\xspace}
\newcommand{\mapreduce}{\textit{MapReduce}\xspace}
\newcommand{\map}{\textit{map}\xspace}
\newcommand{\shuffle}{\textit{shuffle}\xspace}
\newcommand{\reduce}{\textit{reduce}\xspace}
\newcommand{\mapper}{\textit{mapper}\xspace}
\newcommand{\reducer}{\textit{reducer}\xspace}
\newcommand{\maps}{\textit{maps}\xspace}
\newcommand{\reduces}{\textit{reduces}\xspace}
\newcommand{\mapcompute}{\textit{map-compute}\xspace}
\newcommand{\mapspill}{\textit{map-disk-spill}\xspace}
\newcommand{\mapsort}{\textit{map-merge}\xspace}
\newcommand{\reducesort}{\textit{reduce-merge}\xspace}
\newcommand{\hadoop}{\textit{Hadoop}\xspace}
\newcommand{\function}{\textit{function}\xspace}
\newcommand{\functions}{\textit{functions}\xspace}
\newcommand{\job}{\textit{job}\xspace}
\newcommand{\jobs}{\textit{jobs}\xspace}
\newcommand{\phase}{\textit{phase}\xspace}
\newcommand{\phases}{\textit{phases}\xspace}
\newcommand{\task}{\textit{task}\xspace}
\newcommand{\tasks}{\textit{tasks}\xspace}
\newcommand{\pegasus}{\texttt{Pegasus}\xspace}
\newcommand{\pagerank}{\emph{PageRank}\xspace}
\newcommand{\hits}{\emph{HITS}\xspace}
\newcommand{\simrank}{\emph{SimRank}\xspace}
\newcommand{\isorank}{\emph{IsoRank}\xspace}

\lstset{
language=Java,
frame=tblr,
tabsize=2,
showstringspaces=false,
captionpos=b,
floatplacement=ht,
basicstyle=\scriptsize\ttfamily,
keywordstyle=\bfseries,
morekeywords={map,reduce,gmap,lmap,lreduce,greduce}
}

\mode<presentation>
{
  %% \usetheme[compress]{Ilmenau}
  
  %% yellow colortheme
  %% \usecolortheme{crane}
  
  %% ok
  %% \usetheme[compress]{Ilmenau}
  %% \usetheme[compress]{Berlin}

  %% ok, manual
  %% \useoutertheme[footline=authorinstitutetitle]{miniframes}
  
  %% ok, no bottom bar
  % \usetheme{JuanLesPins}
  % \usetheme{Antibes}
  
  %% too broad navigation bar
  %% \usetheme{Malmoe}
  %% \usetheme{Luebeck}
  \usetheme[]{Madrid}
  \setbeamercovered{transparent}

\defbeamertemplate*{footline}{my infolines theme}
{
  \leavevmode%
  \hbox{%
  \begin{beamercolorbox}[wd=.28\paperwidth,ht=2.25ex,dp=1ex,center]{author in head/foot}%
    \usebeamerfont{author in head/foot}\insertshortauthor~(\insertshortinstitute)
  \end{beamercolorbox}%
  \begin{beamercolorbox}[wd=.42\paperwidth,ht=2.25ex,dp=1ex,center]{title in head/foot}%
    \usebeamerfont{title in head/foot}\insertshorttitle
  \end{beamercolorbox}%
  \begin{beamercolorbox}[wd=.30\paperwidth,ht=2.25ex,dp=1ex,right]{date in head/foot}%
    \usebeamerfont{date in head/foot}\insertshortdate{}\hspace*{2em}
    \insertframenumber{} / \inserttotalframenumber\hspace*{2ex}
  \end{beamercolorbox}}%
  \vskip0pt%
}
}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%           Title page and footnote settings          %%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\title[Big Data Analytics Infrastructure] {\textbf{ Part 1:
Systems Infrastructure for Big-Data Analytics}}

\author[Ananth Grama]{\textbf{Ananth Grama}}

\institute[Purdue University] {Center for Science of Information \\
  Dept. of Computer Science, Purdue University\\ ~ \\ With help from
  Giorgos Kollias, Naresh Rapolu, Karthik Kambatla, and Adnan Hassan \\ ~ \\
  Thanks to NSF, DoE, the Center for Science of Information, Microsoft, and Intel.} 

\date[]{}

\begin{document}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \titlepage
\end{frame}

% %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}Outline}
  \begin{itemize}
  \item Motivating applications -- PageRank, Graph Alignment
  \item Case study: single \matvec in MapReduce
  \item Asynchrony and speculation: Optimizations across iterations
    \begin{itemize}
    \item Asynchronous algorithms through Relaxed Synchronization
    \item Speculative parallelism through \texttt{TransMR}
      (transactional \mapreduce)
    \item Locking techniques for efficient distribution transactions
    \end{itemize}
  \item Online Analytics
     \begin{itemize}
     \item Motivating applications
     \item Streams (Storm) and Machine Learning (Vowpal Wabbit)
     \item Runtime and optimizations
     \end{itemize}
  \end{itemize}
\end{frame}




%%%%%%%%%%%%%




%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Motivating Example: Functional PageRank (PR)}
\small{ 
{\bf Computing PageRank (PR)}
 \begin{itemize}
\item PageRank as a \emph{random surfer process}: Start surfing from a random node and keep following links with probability $\mu$ restarting  with probability $1 - \mu$; the node for restarting will be selected based on a personalization vector $v$. The ranking value $x_i$ of a node $i$ is the probability of visiting this node during surfing.
\item PR can also be cast in power series representation as
$x =  (1-\mu) \sum_{j=0}^k \mu^{j} S^j v$; $S$ encodes column-stochastic adjacencies.
  \end{itemize}
}

\small{
{\bf Functional rankings} 
 \begin{itemize}
 \item A general method to assign ranking values to graph nodes as $x =  \sum_{j=0}^k \zeta_j S^j v$. PR is a functional ranking, 
$\zeta_j = (1-\mu) \mu^j$.
\item Terms attenuated by outdegrees in $S$ \emph{and} damping coefficients $\zeta_j$.
  \end{itemize}
}
\end{frame}





%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Motivating Example: Graph Matching}
\small{
  \begin{columns}[h]
    \begin{column}{3.8cm}
      \includegraphics[width=\textwidth]{figures/match-graphs.pdf}
    \end{column}
    \begin{column}{5.8cm}
      \begin{itemize}
      \item
        \textbf{Node similarity:} Two nodes are similar if they are linked by other similar node pairs. By pairing similar nodes, the two graphs become \emph{aligned}.
      \end{itemize}
    \end{column}
  \end{columns}
\begin{itemize}
\item
Let $\tilde{A}$ and $\tilde{B}$ be the normalized adjacency matrices of the graphs (normalized by columns),
$H_{ij}$ be the independently known similarity scores (preferences matrix) of nodes $i \in V_B$ and
$j \in V_A$, and $\mu$ be the fractional contribution of topological similarity.
\item To compute $X$, IsoRank iterates:
$$
X \leftarrow \mu \tilde{B} X {\tilde{A}}^{T} + (1-\mu) H
$$
\end{itemize}
}
\end{frame}



%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{MapReduce: Basics} 

  Execute in parallel user-defined functions on individual data-items
  distributed across machines.

  \begin{itemize}
  \item Simple programming model --- \map and \reduce functions
  \item Scalable, distributed execution of these functions on massive
    amounts of data on commodity hardware
  \end{itemize}

  \begin{figure}
    \vspace{-0.1in}
    \center
    \includegraphics[width=0.55\textwidth]{figures/MapReduce-Model.pdf}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}MapReduce: DataFlow}
  \begin{figure}
    \center
    \includegraphics[width=0.9\textwidth]{figures/MapReduce-DataFlow.pdf}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}[fragile]
  \frametitle{Example: Shotest Path (\matvec) in MapReduce}
  \begin{lstlisting}
    map(node, adjList) {
      for each arc in adjList {
        output(arc.dst, node.distance + arc.weight)
      }
    }
  \end{lstlisting}
  \begin{lstlisting}
    reduce(node, newDistList) {
      node.distance = min(newDistList)
    }
  \end{lstlisting}
  \begin{lstlisting}
    while(not converged) {
      runMapReduceJob(map, reduce, Ax)
    }
  \end{lstlisting}
  \begin{itemize}
  \item We call this the \textbf{Naive \matvec} --- \map takes an
    adjacency list as input
  \item Pegasus (CMU) implementation reads one edge per \map
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}Naive \matvec: Resource Utilizaton}
 % Resource Utilizaton: CPU, Network, Disk
  \begin{columns}[t]
    \begin{column}{0.4\textwidth}
      \begin{figure}
        \center
        \includegraphics[width=\columnwidth]{figures/nm-cpu.pdf}
      \end{figure}
    \end{column}
    \begin{column}{0.4\textwidth}
      \begin{figure}
        \center
        \includegraphics[width=\columnwidth]{figures/nm-netio.pdf}
      \end{figure}
    \end{column}
  \end{columns}
  \begin{figure}
    \includegraphics[width=0.4\textwidth]{figures/nm-disk.pdf}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}Optimizing the \matvec further}
  \begin{figure}
    \center
    \includegraphics[width=0.4\textwidth]{figures/split-times.pdf}
  \end{figure}
  \begin{itemize}
  \item Stages overlap; hence, sum of percentages $>100$
  \item I/O time $>$ Computation time
  \item For performance:
    \begin{itemize}
    \item Batch read data
    \item Each \map processes more data (as much as can fit in memory)
    \end{itemize}
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}Partitioned \matvec}
  \begin{itemize}
    \item Each \map operates on a \emph{graph partition} --- a set
      of adjacency lists
    \item Partition size is constrained by the heap size 
    \item On our setup, maximum partition size was $2^{19}$ nodes
  \end{itemize}
  \begin{figure}
    \center
    \includegraphics[width=0.6\textwidth]{figures/part-time-psize.pdf}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Performance: Pegasus vs Naive vs Partitioned}
  \begin{columns}[t]
    \begin{column}{0.45\textwidth}
      \center
      16 Amazon EC2 nodes
      \begin{figure}
        \includegraphics[width=\textwidth]{figures/pega-naive-part-16.pdf}
      \end{figure}
    \end{column}
    \begin{column}{0.45\textwidth}
      \center
      32 Amazon EC2 nodes
      \begin{figure}
        \includegraphics[width=\textwidth]{figures/pega-naive-part-32.pdf}
      \end{figure}
    \end{column}
  \end{columns}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Optimizations Across Iterations}
  Algorithmic optimizations:
  \begin{itemize}
  \item {\bf Asynchronous algorithms}: Algorithms that allow
    asynchrony --- ordering of updates doesn't affect the correctness
    of the algorithm\\
    e.g., PageRank, Alignments
  \item {\bf Speculative Parallelism}: Algorithms where concurrent
    computations can have potential conflicts, the conflicts are rare
    and can only be detected at runtime\\
    e.g., Boruvka's MST, Single Source Shortest Path
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Asynchronous Iterations}
  \begin{itemize}
  \item Improve performance in parallel environments.
    \begin{itemize}
    \item Infrequent synchronization reduces communication
    \item Examples
      \begin{itemize}
        \item Graph algorithms, Numerical methods, ML kernels, etc.
      \end{itemize}
    \end{itemize}
  \item More pronounced gains in distributed environments
    \begin{itemize}
    \item Higher communication and data-movement costs
    \item Read once, process multiple times allows more computation
      for the same I/O cost
    \end{itemize}
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Relaxed Synchronization}
  \begin{itemize}
  \item Synchronize once every few iterations
  \item Approach: Two levels of MapReduce 
    \begin{itemize}
    \item Global MapReduce: The regular MapReduce
      \begin{itemize}
      \item Requires global synchronization
      \end{itemize}
    \item Local MapReduce: MapReduce within a global map
      \begin{itemize}
      \item Each global map runs a few iterations of local MapReduce
      \item Partial synchronization of data of a single global map task
      \end{itemize}
    \item Input data partitioning for fewer dependencies across partitions
    \end{itemize}
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{PageRanks Using Traditional MapReduce}
  \begin{figure}
    \center
    \includegraphics[width=0.9\textwidth]{figures/General-PageRank.pdf}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{PageRank with Relaxed Synchronization}
  \begin{figure}
    \center
    \includegraphics[width=0.9\textwidth]{figures/Relaxed-PageRank.pdf}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%\begin{frame}
%  \frametitle{Relaxed Synchronization: Semantics}
%  \begin{figure}
%    \center
%    \includegraphics[width=0.9\textwidth]{figures/Semantics-IterativeMapReduce.pdf}
%  \end{figure}
%\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}[fragile]
  \frametitle{Realizing Relaxed Synchronization Semantics}
  \begin{itemize}
  \item Code \textit{gmap, greduce, lmap, lreduce}
    \begin{itemize}
    \item lmap, lreduce use EmitLocalIntermediate() and Emit Local()
    \item Synchronized hashtables for local storage
    \end{itemize}
  \end{itemize}
  \begin{lstlisting}[frame=tblr]
    gmap(xs : X list) {
      
      while(no-local-convergence-intimated) {
        for each element x in xs {
          lmap(x); // emits lkey, lval
        }
        
        lreduce(); // operates on the output of lmap functions 
      }
      
      for each value in lreduce-output{
        EmitIntermediate(key, value);
      }
    }
  \end{lstlisting}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}Evaluation --- Relaxed Synchronization}
  \begin{itemize}
  \item Sample applications
    \begin{itemize}
    \item Single Source Shortest Path (MST, transitive closure, etc.)
    \item PageRank (mat-vec: eigen value and linear system solvers)
    \end{itemize}
  \item Experimental Testbed
    \begin{itemize}
    \item 8 Amazon EC2 Large Instances
      \begin{itemize}
      \item 64-bit compute units with 15 GB RAM, 4x 420 GB storage
      \item Hadoop 0.20.1; 4 GB heap space per slave
      \end{itemize}
    \end{itemize}
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}Single Source Shortest Path}
  \begin{figure}
    \center
    \includegraphics[width=0.9\textwidth]{figures/SSSP.pdf}
    \label{fig:performance-sssp}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}PageRank}
  \begin{itemize}
  \item Input: Partitioned using METIS ( less than 10 seconds)
  \item Damping factor = 0.85
  \end{itemize}
\begin{table}[th]
\centering { \small
\begin{tabular}{|l|l|l|}
\hline
\small 
\ & \textbf{GraphA}& \textbf{GraphB}\\ 
\hline
\textbf{Nodes} & 280,000 & 100,000 \\
\textbf{Edges} & 3 million & 3 million \\
\hline
\end{tabular}
}
\label{table:page-rank}
\end{table}
\end{frame}



%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}PageRank Performance: GraphA}
  \begin{figure}
    \center
    \includegraphics[width=0.9\textwidth]{figures/GraphA.pdf}
    \label{fig:performance-grapha}
  \end{figure}
\end{frame}



%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{\hspace{0.2in}PageRank Performance: GraphB}
  \begin{figure}
    \center
    \includegraphics[width=0.9\textwidth]{figures/GraphB.pdf}
    \label{fig:performance-graphb}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
\frametitle{Beyond Data-Parallelism: Speculation}
  {\bf Speculative parallelism}
  \begin{itemize}
  \item
    Most of the data can be operated on in parallel. 
  \item
    Some executions conflict. These can only be detected at runtime.
      [Pingali et.al., PLDI'11]
  \item
    Online algorithms/ Pipelined workflows
    \begin{itemize}
      \item
        MapReduce Online [Condie'10] is an approach needing havy checkpointing.
    \end{itemize}
  \item
    Software Transactional Memory (STM)
\end{itemize}
\end{frame}


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
\frametitle{TransMR: Transactional MapReduce}
\begin{itemize}
\item Goal: Exploit speculative data-parallelism
\item Support data-sharing across concurrent computations to detect
  and resolve conflicts at runtime
\item Solution:
  \begin{itemize}
  \item Use distributed key-value stores as shared address space across computations
  \item Address inconsistencies arising due to the disparate fault-tolerance mechanisms
  \item Transactional execution of \map and \reduce functions
  \end{itemize}
\end{itemize}

\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
\frametitle{TransMR: System Architecture}
\begin{itemize}
\item Distributed key-value store provides a shared-memory abstraction
  to the distributed execution-layer.
\end{itemize}
  \begin{figure}[th]
  \center
  \includegraphics[width=0.90\textwidth]{figures/system-eps-converted-to.pdf}
  \label{fig:system}
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Semantics of the API}
\begin{itemize}
\item
Data-centric function scope -- Map/Reduce/Merge etc, -- termed as a
Computation Unit (CU), is executed as a transaction.
\item
Optimistic reads and write-buffering. Local Store (LS) forms the
write-buffer of a CU.
\begin{itemize}
  \item
    Put (K, V):  Write to LS, which is later atomically committed to GS.
  \item
    Get (K, V):  Return from LS, if already present; otherwise, fetch from GS and store in LS.  
  \item
    Other Op:  Any thread local operation.
\end{itemize}
\item
  The output of a CU is always committed to the GS before being
  visible to other CU’s of the same or different type.
  \begin{itemize}
  \item
    Eliminates the costly shuffle phase of MapReduce.
  \end{itemize}
\end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Design Principles}
  \begin{itemize}
  \item
    Optimistic concurrency control over pessimistic locking
    \begin{itemize}
      \item
        Locks are acquired at the end of the transaction. Write-buffer
        and read-set is validated against those of concurrent Trx
        assuring serializability.
      \item
        Client is potentially executing on the slowest node in the
        system; in this case, pessimistic locking hinders parallel
        transaction execution.
    \end{itemize}
  \item
    Consistency (C) and Tolerance to Network Partitions (P) over
    Availability (A) in CAP Theorem for Distributed transactions.
    \begin{itemize}
      \item
        Application correctness mandates strict consistency of
        execution. Relaxed consistency models are application-specific
        optimizations.
      \item
        Intermittent non-availability is not too costly for
        batch-processing applications, where client is fault-prone in
        itself.
    \end{itemize}
  \end{itemize}
\end{frame}
    
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Evaluation}
  \begin{itemize}
    \item
      We show performance gains on two applications, which are
      hitherto implemented sequentially without transactional support;
      both exhibit Optimistic data-parallelism.
    \item
      Boruka's MST
      \begin{itemize}
        \item
          Each iteration is coded as a Map function with input as a
          node. Reduce is an identity function. Conflicting maps are
          serialized while others are executed in parallel.
        \item
          After n iterations of coalescing, we get the MST of an n
          node graph.
        \item
          A graph of 100 thousand nodes, with average degree of 50,
          generated based on the forest-fire model.
      \end{itemize}
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Boruvka's MST}
  \begin{itemize}
   \item Speedup of 3.73 on 16 nodes, with less than 0.5 \% re-executions due to aborts.
  \end{itemize}
  \begin{figure}[th]
    \center
    \includegraphics[width=0.90\textwidth]{figures/boruvkas-eps-converted-to.pdf}
  \label{fig:boruvkas}
  \end{figure}
\end{frame}    

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Maximum Flow Using Push-Relabel Algorithm}
  \begin{itemize}
    \item
      Each Map function executes a Push or a Relabel operation on the
      input node, depending on the constraints on its neighbors.
    \item
      Push operation increases the flow to a neighboring node and
      changes their ``Excess''.
    \item
      Relabel operation increases the height of the input node if it
      is the lowest among its neighbors.
    \item
      Conflicting Maps -- operating on neighboring nodes -- get
      serialized due to their transactional nature.
    \item
      Only sequential implementation possible without support for
      runtime conflict detection.
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Maximum flow using Push-Relabel algorithm}
  \begin{itemize}
   \item Speedup of 4.5 is observed on 16 nodes with 4\% re-executions
     on a windown of 40 iterations.
  \end{itemize}
  \begin{figure}[th]
    \center
    \includegraphics[width=0.90\textwidth]{figures/preflow-eps-converted-to.pdf}
  \label{fig:preflow}
  \end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{TransMR: Intermediate Lessons}
  \begin{itemize}
    \item
      TransMR programming model enables data-sharing in data-centric
      programming models for enhanced applicability.
    \item
      Similar to other data-centric programming models, the programmer
      only specifies operation on the individual data-element without
      concerning about its interaction with other operations.
    \item
      Prototype implementation shows that many important applications
      can be expressed in this model while extracting significant
      performance gains through increased parallelism.
  \end{itemize}
{\bf BUT: What about the locking operations!}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Distributed Transactions on Key-Value Stores}
  \begin{itemize}
  \item Transactions are costly in a large scale distributed settings
    \begin{itemize}
    \item two-phase locking (concurrency control) 
    \item two-phase commit (atomicity)
    \end{itemize}
  \item Careful examination of the protocols and optimizations crucial
    to performance of TransMR-like systems
  \item These optimizations also useful for general purpose
    transactions on databases using key-value store as the underlying
    storage
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Lock Management in Distributed Transactions}
  \begin{itemize}
  \item Lock management the major bottleneck affecting the latency of
    distributed transactions.
%All the other operations can be batched to exploit parallelism. Locks also need to be persisted to account for %node failures.
  \item Consider Strong Strict two phase locking (SS2PL) -- waiting
    case: The lock-acquiring stage is the only sequential stage. The
    other stages can be parallelized to finish in a single round-trip.
  \item Holds true even in optimistic-concurrency techniques where the
    locks are acquired at the end.
\end{itemize}
    \begin{figure}[th]
      \center
      \includegraphics[width=0.50\textwidth]{figures/pessimistic-concurrency-eps-converted-to.pdf}
      \label{fig:pessimistic-concurrency}
    \end{figure}    
    \begin{figure}[th]
      \center
      \includegraphics[width=0.50\textwidth]{figures/waiting-case-eps-converted-to.pdf}
      \label{fig:waiting-case}
    \end{figure}    
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Workload Aware Lock Management}
  \begin{itemize}
    \item
      \textbf{Contention based Lock-ordering}: Order the locks so as
      to decrease the total amount of waiting time.
    \item
      For the waiting case, the lock with the least contention should
      be acquired first. This increases pipelining while decreasing
      lock-holding times.
    \item
      Contention order is a runtime characteristic, and is updated
      consistently. All clients should adhere to the same order to
      avoid deadlocks.
  \end{itemize}
  \begin{figure}[th]
    \center
    \includegraphics[width=0.7\textwidth]{figures/lock-ordering-grouping-eps-converted-to.pdf}
    \label{fig:lock-ordering-grouping}
  \end{figure}    
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Constrained k-way Graph Partitioning}
  \begin{itemize}
    \item
      Graph partitioning algorithm to split the locking into k
      non-overlapping partitions, minimizing the sum of weights on
      cut-edges, while approximately balancing the total weight (sum
      of node-weights) of individual partitions.
    \item
      The result of the partitioning algorithm is the
      load-balanced-partitioning of locks among k storage nodes. 
  \end{itemize}
  \begin{figure}[th]
    \center
    \includegraphics[width=0.6\textwidth]{figures/partitioning-eps-converted-to.pdf}
    \label{fig:partitioning}
  \end{figure}    
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\begin{frame}
  \frametitle{Evaluation}
  \begin{itemize}
    \item
      A cluster of 20 machines was used for all evaluations. Each
      machine had a Quad-core xeon processor with 8 GB of
      RAM. HBase is the underlying key-value store.
    \item
      The YCSB benchmark was extended with the atomic multi-put
      operation. A client transaction involves an atomic
      Read-Modify-Write operation on a set of keys.
    \item
      The keys for the atomic operation are generated using a Zipfian
      generator with variable zipfian parameter. Each transaction
      updates 15 keys out of 50K keys.
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Two Phase Locking: Waiting Version}
  \begin{figure}[th]
     \center
      \includegraphics[width=0.50\textwidth]{figures/throughput-15-50000-ordering-based-waiting-case-5000-eps-converted-to.pdf}
    \caption{Performance of Lock Partitioning}
  \end{figure}
  \begin{itemize}
  \item Ordering of keys in increasing-order of their contention
    significantly better than the decreasing order.
   %The reason is mainly
   %due to the pipelining effect induced by the increasing-order
   % strategy.
  \item
    The increasing-order reduced lock holding time for highest
    contended locks reducing waiting time for other transactions.
  \end{itemize}
\end{frame}


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Two Phase Locking: Waiting Version}
  \begin{figure}[th]
      \center
      \includegraphics[width=0.50\textwidth]{figures/throughput-15-50000-partitioning-based-waiting-case-5000-eps-converted-to.pdf}
    \caption{Performance of Lock Partitioning}
  \end{figure}
  \begin{itemize}
  \item
    Partitioning is done using Metis and partitions are placed at
    separate nodes.  Lock-partitioning improves the throughput by
    reducing the number of network-roundtrips needed for sequential
    locking by the client.
  \end{itemize}
\end{frame}


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Optimistic Concurrency Control -- No-waiting Version}
  \begin{figure}[th]
      \center
      \includegraphics[width=0.50\textwidth]{figures/throughput-15-50000-ordering-based-3-SI_NB-eps-converted-to.pdf}
      \caption{Performance of Lock Ordering}
  \end{figure}
  \begin{itemize}
%    \item
%      In OCC, locks are acquired after the read-phase of the
%      transaction and thus held for a small duration.
%    \item
%      In the no-waiting version, all the locks are acquired in
%      parallel. However, even if a single lock could not be acquired,
 %     all locks are immediately given up. 
  \item
      Smaller improvement for OCC mainly due to the shorter duration of
      locking.
    \item
      At similar contention levels, the throughput of optimistic
      concurrency control is significantly lower.
%This is in-line with the reasoning that,
%      optimistic techniques are not suitable at high contention levels
%      as the time spent in reading and local updating gets wasted when
 %     conflicted during commit.
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
  \frametitle{Optimistic Concurrency Control -- No-waiting Version}
  \begin{figure}[th]
      \center
      \includegraphics[width=0.50\textwidth]{figures/locks-acquired-per-trx-15-50000-ordering-based-3-SI_NB-eps-converted-to.pdf}
      \caption{Lock wastage due to restarts}   
  \end{figure}
  \begin{itemize}
    \item
Optimistic techniques not suitable at high contention levels as the
time spent in reading and local updating gets wasted due to conflicts
during commit.
  \end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Lock Optimization: Conclusions}
  \begin{itemize}
  \item
    The waiting version of SS2PL with increasing-contention-order and
    partitioning outperforms the other protocols significantly.
  \item
   Restarts due to conflicts constitute a major overhead in
 distributed transactions.  Reducing restarts by busy-waiting for
    locks is an important step towards increasing performance.
  \item
    Understanding the workload - even simple statistics on contention
    - is enough to achieve significant gains (up to 200\%).
  \item
    Lock-partitioning through graph clustering and partitioning
    techniques can be performed dynamically to achieve performance
    gains.
  \end{itemize}
\end{frame}


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Towards Online Learning}
\begin{figure}[th]
    \center
    \vspace*{-1.0in}
    \includegraphics[width=0.750\textwidth]{figures/online.pdf}
    \caption{}   
\end{figure}
\end{frame}



%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Motivating Applications of Online Learning}
\begin{itemize}
\item
Ad-Servers (learning from click-streams, spatial parameters)
\item
Real-time sentiment classification (learning from twitter feeds, blogs, etc.)
\item
Read-time content recommendation (correlating tweets, hashtags, etc.)
\end{itemize}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Case Study: Stochastic Gradient Descent}

{\tt
\ \ \ \ \ \ \ \ Randomly shuffle training samples \\
\ \ \ \ \ \ \ \ repeat \{  \\
\ \ \ \ \ \ \ \ \ \ \ \ for i := 1, 2, ..., m \{  \\
\ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \  T\_j := T\_j - a(ht(x(i) - y(i)) x(j)  \\
\ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ (for every j = 0, 1, ..., n)  \\
\ \ \ \ \ \ \ \ \ \ \ \ \}  \\
\ \ \ \ \ \ \ \ \}  \\
}


Can be used in applications like Supervised Semantic Indexing (e.g., Minimize loss
function $F = 1 - q^*W^+d^+ + q^*W^+d^-$, where $q$ is the query document,
$W$ is a weight matrix, $d^+$ corresponds to a positive document and $d^-$ corresponds
to a negative document.)
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Stream Processing Solutions}

Storm is a stream processing envine built at Twitter. It executes a DAG of
operators consisting of Spouts and Bolts.

\begin{figure}[th]
    \center
    \vspace*{-1.5in}
    \includegraphics[width=1.0\textwidth]{figures/storm.pdf}
    \caption{}   
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%\begin{frame}
%
%\begin{figure}[th]
%    \center
%    \vspace*{-1.0in}
%    \includegraphics[width=1.0\textwidth]{figures/lmax.pdf}
%    \caption{}   
%\end{figure}
%\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Building Online Applications: Storm and Vowpal Wabbit}
\begin{figure}[th]
    \center
    \vspace*{-1.0in}
    \includegraphics[width=0.75\textwidth]{figures/vowpal1.pdf}
    \caption{}   
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Limitations of Simple Integration}
Long reduction times, network bandwidth bottlenecks, static topologies and
dynamic system state.

\begin{figure}[th]
    \center
    \vspace*{-1.0in}
    \includegraphics[width=0.75\textwidth]{figures/vowpal2.pdf}
    \caption{}   
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Limitations of Simple Integration}

Impedance mismatch between synchronous reductions and asynchronous execution
engine (LMAX Disruptor).

\begin{figure}[th]
    \center
    \vspace*{-1.5in}
    \includegraphics[width=0.75\textwidth]{figures/vowpal3.pdf}
    \caption{}   
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{A dynamically orchestrated online learning framework}

Controller dynamically schedules reductions and forces rate control and routing
on input data streams

\begin{figure}[th]
    \center
    \vspace*{-1.5in}
    \includegraphics[width=0.75\textwidth]{figures/vowpal4.pdf}
    \caption{}   
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Controller Dexign: Asynchronous Reductions}

Schedules asynchronous reductions (staggered butterfly) while forcing rate control
(blacklisting reducers)

\begin{figure}[th]
    \center
    \vspace*{-1.5in}
    \includegraphics[width=0.75\textwidth]{figures/vowpal5.pdf}
    \caption{}   
\end{figure}
\end{frame}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Controller Design: Dynamic Mapping of Virtual Overlay to 
Physical Operators}

Controller uses operator metrics to decide on dynamic mapping and schedule reductions.
Metrics: Model sparsity triggers new reduction; Latency and throughput of
individual operators (scale-in/scale-out)

\begin{figure}[th]
    \center
    \vspace*{-1.5in}
    \includegraphics[width=0.75\textwidth]{figures/vowpal6.pdf}
    \caption{}   
\end{figure}
\end{frame}


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\begin{frame}
\frametitle{Continuing Efforts}

TransMR and Concurrency Control Modules available on request. Online analytics
framework currently being validated. Available in limited release.

\end{frame}

\end{document}
