% by Mirella M. Moro; version: January/18/2012 @ 04:16pm
% -- 01/18/2012: more discussion on SBBD + JIDM; overall revision
% -- 09/03/2010: bib file with names for proceedings and journals; cls with shrinked {received}
% -- 08/27/2010: appendix, table example, more explanation within comments, editors' data

\documentclass[jidm,a4paper]{jidm} % NOTE: JIDM is published on A4 paper
\usepackage{graphicx,url}  % for using figures and url format
\usepackage[T1]{fontenc}   % avoids warnings such as "LaTeX Font Warning: Font shape 'OMS/cmtt/m/n' undefined"
%\usepackage{caption}
%\usepackage{subcaption}
%\usepackage{float}

\usepackage{listing} %% TENTAR INCLUIR... "listing" ou "listings"?

%\usepackage{cite} % NOTE: do **not** include this package because it conflicts with jidm.bst

% Standard definitions
\newtheorem{theorem}{Theorem}[section]
\newtheorem{conjecture}[theorem]{Conjecture}
\newtheorem{corollary}[theorem]{Corollary}
\newtheorem{proposition}[theorem]{Proposition}
\newtheorem{lemma}[theorem]{Lemma}
\newdef{definition}[theorem]{Definition}
\newdef{remark}[theorem]{Remark}

% New environment definition
\newenvironment{latexcode}
{\ttfamily\vspace{0.1in}\setlength{\parindent}{18pt}}
{\vspace{0.1in}}

% ALL FIELDS UNTIL BEGIN{document} ARE MANDATORY

% The following data (volume, number and page) are given by the editors prior to publishing your article
\jidmVolume{volX}
\jidmNumber{numY}
\jidmYear{13}
\jidmMonth{October}
\setcounter{page}{1}


% Includes headers with simplified name of the authors and article title
\markboth{R.J.M. Mogrovejo et. al.}
{JIDM - Journal of Information and Data Management}
%  -> \markboth{}{}
%         takes 2 arguments
%         ex: \markboth{M. M. Moro}{Any article title}


% Title of the article
\title{Towards a Statistical Evaluation of PigLatin Joins}

% List of authors
%IF THERE ARE TWO or more institutions, please use:
%\author{Name of Author1\inst{1}, Name of Author2\inst{2}, Name of Author3\inst{2}}
\author{Renato Javier M. Mogrovejo\inst{1}, Jos\'e Maria Monteiro\inst{2}, Javam C. Machado\inst{2}, Carlos Juliano M. Viana\inst{3}, S\'ergio Lifschitz\inst{3} }


%Affiliation and email
\institute{Universidade Catolica San Pablo (Arequipa), Peru \\ \email{rmarroquin@ucsp.edu.pe}
% IF THERE IS ANOTHER INSTITUTION:
%\and Name_of_the_second_institution \\
%\email{address@whatever.com}
\and
Universidade Federal do Cear\'a (Fortaleza), Brasil \\ \email{\{monteiro,javam\}@ufc.br}
\and
Pontif\'{\i}cia Universidade Cat\'olica do Rio de Janeiro, Brasil \\ \email{\{cviana,sergio\}@inf.puc-rio.br}
}


% Article abstract - it should be from 100 to 300 words
\begin{abstract}
There are different scalable data management solutions which can take advantage of cloud features making them more attractive for a deployment in such environments. One of the most critical operations in data processing is joining large data sets.
This is one of the most expensive and hardest operations to optimize.
We are mainly  concerned here with join operations expressed in PigLatin, an abstract query language for a high-level platform, called Pig, which creates {\tt MapReduce} programs with Hadoop. In this work we explore statistical methods, namely multiple regression analysis, in order to predict three distinct join types execution times, comparing them with actual running times.
\end{abstract}


% ACM Computing Classification System categories
\category{H.2}{Database Management}{Experimentation}
%\category{H.3}{Information Storage and Retrieval}{Miscellaneous}
%\category{I.7}{Document and Text Processing}{Miscellaneous}

% Categories and Descriptors are available at the 1998 ACM Computing Classification System
% http://www.acm.org/about/class/1998/
%  -> \category{}{}{}
%         takes 3 arguments for the Computing Reviews Classification Scheme.
%         ex: \category{D.3.3}{Programming Languages}{Language Constructs and Features}
%                   [data types and structures]
%                   the last argument, in square brackets, is optional.

% Article keywords
\keywords{Cloud databases, MapReduce, PigLatin, Hadoop, Join Evaluation, Multiple Regression}
%  -> \keywords{} (in alphabetical order \keywords{document processing, sequences,
%                      string searching, subsequences, substrings})


% THE ARTICLE BEGINS
\begin{document}

% This is optional  --> agradecimentos de bolsas, estagiarios etc:
\begin{bottomstuff}
% similar to \thanks
% for authors' addresses; research/grant statements
\end{bottomstuff}

\maketitle


\section{Introduction}

An important contribution to large scale data processing has been the  {\tt MapReduce} programming model ~\cite{Dean:2004:MSD:1251254.1251264}.
It was designed to deal with fault-tolerant systems, to enable high-availability  and to run on heterogeneous hardware.
It has become almost a {\it de facto} standard for massive data set computations in the cloud. However, the amount of code generated to perform relatively simple data management tasks such as projections and filters may become a problem. For instance, join execution in Hadoop (an open-source platform for the {\tt MapReduce} paradigm) is a very hard task for users even though joins are one of the most critical operations in data processing. Along with known problems such as estimating the output result size  and join selectivity, new challenges arise when considering {\tt MapReduce} abstractions e.g. data localization and skewness.

Research projects like Pig ~\cite{Olston:2008:PLN:1376616.1376726}
and  Hive ~\cite{Thusoo:2009:HWS:1687553.1687609}
try to make things simpler by creating an abstraction layer for {\tt MapReduce} programming tasks.
Hive is a data warehouse that uses Hadoop Distributed File System (HDFS) for storing its data, with a SQL-like language called HiveQL to access the data. Pig works on top of Hadoop framework~\cite{hadoop}
using a high-level programming language called {\tt PigLatin}. PigLatin
has some similarities with {\tt SQL} because
both perform some similar operations from relational algebra, but they aim at different goals. PigLatin expresses transformations as a sequence of steps while {\tt SQL} describes the desired outcome. This feature is considered an advantage because writing a sequence of steps is more natural to developers. This procedural view allows them to write more complex data-flows
when compared to plain {\tt MapReduce}, or even trying to express their data-flows with a regular data oriented language.

These high-level programming languages help improving the usability of {\tt MapReduce} by simplifying the use of {\tt MapReduce's} API and tuning parameters as they provide a query language similar to traditional structured query language (SQL).
In spite of the fact that these programming languages abstract and simplify some {\tt MapReduce} complexity, they still need further improvements. For example, cost-optimizers are still work in progress and most of the data is stored with no corresponding metadata, among other extra features of traditional database systems. We explore a statistical approach to find relationships among the known information about our data set and usual data center infrastructures. We evaluate query execution time by using a set of join queries from the TPC-DS database benchmark. We hypothesize that using statistical models we could run "{\it what-if}" workload scenarios by analyzing an input workload and measuring its performance in order to predict it.

This article aims to investigate the expected efficiency of systems like Pig and Hive, by using a statistical method for modeling join operations' execution time and, then, being able to predict their performances. In our approach, the construction of such a model is done by building up a training data set and  performing cross-validation to verify it.

One of the challenges encountered in this research work was the lack of a data set to evaluate specific operations in large scale systems. We used a known database benchmark, {\tt TPC-DS}, to simulate real data with real data distributions. Nevertheless, queries proposed in {\tt TPC-DS} are not suitable for systems based on the {\tt MapReduce} programming model. To overcome this issue, we have converted relations described in the {\tt TPC-DS} into {\tt PigLatin} queries to specifically evaluate join operations.


This article is structured as follows: Section~\ref{sec::backg}
presents
the motivation, some background information and related research works.
In the following section, we present the Hadoop Joins as they are implemented by the Pig system.  In Section~\ref{sec::stati}, the approach and design decisions about our main proposal are explained. We give also some experimental results obtained as well as the insights gathered from them. Finally, Section~\ref{sec::conclusions}
presents the conclusions of this work and discusses some future work.

\section{Background and Motivation}
\label{sec::backg}

In order to take advantage of the {\it infinite} resource supply offered by the cloud computing paradigm, specific parallel query languages were needed. A very important contribution to the cloud computing research area is the programming model {\tt MapReduce}~\cite{Dean:2004:MSD:1251254.1251264}. It has become  a {\it de facto} standard for massive data set computations in the cloud, a nice
abstraction to facilitate distributed coding to programmers. The {\tt MapReduce} framework
provides characteristics such as fault tolerance, load balancing and task parallelization in a transparent way.

\subsection{Hadoop, Pig and PigLatin}

There are many implementations of this distributed computing paradigm.
One of the most used is the open source framework Hadoop~\cite{hadoop},
created to support the Nutch~\cite{apache-nutch} search engine initiative. Later, it was taken by {\tt Yahoo!}, now
 {\tt Apache}, in order to address critical business needs like managing
 increasing volumes of data.
 Hadoop is used in many other open source projects due to its ability to process data in a distributed way while using commodity hardware.

To take full advantage of the distributed computation, {\tt MapReduce} uses an underlying distributed file system, {\tt Google File System} (GFS)~\cite{Ghemawat:2003:GFS:945445.945450}, to locally access data. {\tt GFS} has a master/slave architecture
with a large number of working nodes. All the files stored in the GFS are broken up into fixed size chunks
in order to have a better performance in I/O operations due to size similarity with sectors in regular file systems. These file chunks are then stored and distributed along with all the available working nodes, or chunk servers, while the metadata associated with all of them is kept in the Master node.
There is also an open-source implementation of GFS that works along with the Hadoop framework called {\tt Hadoop Distributed File System} (HDFS)~\cite{apache-hdfs}.

In the open source world, Pig and Hive are important projects.
Particularly, Pig is an abstraction layer of the Hadoop framework that compiles data analysis tasks into {\tt MapReduce} jobs and executes them on Hadoop.
Pig uses a language for expressing its dataflow called {\it PigLatin}. One of {\it PigLatin}'s biggest advantages is that it enables programmers to express data transformation tasks in few lines. {\it PigLatin} also differs from SQL in which the former allows expressing transformations as a sequence of steps, and the latter describes desired outcome. This is seen as
an advantage because writing a sequence of steps comes natural to developers and allows them to write more complex data flows.

For example, consider the following SQL query that filters the data by applying a user-defined function called '{\tt Clean}'. It then counts the number of remaining entries in the log per region.

{\fontsize{3mm}{5mm} \selectfont
\begin{verbatim}
SELECT U.regions, count(*) as total
FROM UsersRegistry U
WHERE Clean(U.query) = 10
GROUP BY U.regions
\end{verbatim}
}

%\lstset{numbers=left}
%\lstset, numberstyle=\tiny, stepnumber=1, numbersep=5pt, language = SQL}

%\begin{lstlisting}[frame=single,caption={SQL example}]
%\begin{listing}
%SELECT U.regions, count(*) as total
%FROM UsersRegistry U
%WHERE Clean(U.query) = 10
%GROUP BY U.regions
%\end{listing}

In PigLatin, this example could have been written as:

{\fontsize{3mm}{5mm} \selectfont
\begin{verbatim}
user_registry = LOAD 'UsersRegistry.dat'
                AS (userId, userRegions, userQuery);
users_filtered = FILTER user_registry BY Clean(userQuery);
users_grous = GROUP users_filtered BY userRegions;
output = FOREACH users_grous
         GENERATE $0 AS regions, count($1) as total;
STORE output INTO 'results.txt' USING PigStorage();
\end{verbatim}
}

These works on Pig, Hive and Hadoop show that high level programming languages for programming large-scale parallel computations are going to be more predominant than plain {\tt MapReduce}
jobs.

\subsection{Related Work}
\label{sec::Related}

Predicting job execution times is a known problem in the distributed systems field. In a very interesting research work (\cite{4228004}) the construction of a system prediction services is explained. The system was built using an instance-based learning technique to predict job execution times, batch scheduling queue wait times and file transfer times of the {\it Texas Advanced Computing Center Lonestar System}~\cite{tacc}. Even though the approach shows relatively high prediction error for the training data (ranging from 37\% to 115\%), the author discusses that it would be possible to reduce the prediction error by using a larger set of training data. However, it might be possible that the system on which they are working might be simply less predictable than other systems better performances. Nevertheless, the results obtained show a lower prediction error in more structured and well behaved workloads when compared to the work discussed in~\cite{Vazhkudai:2002:PPW:645610.660893}.

On the other hand, estimating query execution time is also a very interesting research topic for the database community due to the fact that this information can help improving the whole system's performance, but can also lead to more proactive decisions of the DBMS (maintaining statistics, choosing better execution plans, etc.). In spite of the similarities of the database world with {\tt MapReduce} environments, they are still different in many aspects, and a very important one is the amount of metadata kept by databases. {\tt MapReduce} environments tend to keep only operational information (i.e. information they need to keep functioning) in order to remain simple, but reliable.

A known scenario with similar features to {\tt MapReduce} environments is the one studied by Multi Database Management Systems (MDBS)~\cite{Bright:1994:ARS:176567.176569}.
For instance, Zhu~\cite{Zhu:1993:IME:962367.962388} discusses the need of good  selectivity estimates before query execution so that the global query optimizer in a MDBS can choose a low cost execution plan from many alternatives. The author also presents an integrated method to estimate selectivity in a MDBS based on the discussion. He accomplishes this by applying the techniques proposed by Lipton et al.~\cite{Lipton:1995:QSE:211040.211053} to an MDBS scenario overcoming difficulties such as not knowing the local structure of a relation, or not being able to modify it. In~\cite{Zhu:1994:QSM:645479.655114}, Zhu et al. proposed a query sampling method to derive cost estimation formulas for autonomous local database systems in an MDBS. The main idea of their work is to classify local queries with similar costs, sample each query class, and to use the observed costs of the sample queries and statistically derive local cost estimation formulas for the queries performed on the local databases i.e. using query sampling for estimating local query costs in an MDBS. If the available information were not enough, they would classify queries with similar performance behavior into the same class so that their estimation errors would not be large.

Likewise Olston et al.~\cite{Olston:2008:AOP:1404014.1404035} describe the needs in high-level dataflow languages built on top of large-scale parallel dataflow systems. Due to the fact that they provide a faster program development environment and an easier way to maintain the code, they also bring new difficulties and opportunities
e.g. automatic optimization and query rewriting.  The authors argue about the different possible types of optimizations in Data-Intensive Scalable Computing {\tt (DISC)}~\cite{Bryant07data-intensivesupercomputing:} and in database systems.
They also point out the differences that make these two contexts similar.

We claim that high-level dataflow systems using the {\tt MapReduce} paradigm could benefit from the use of cost estimates. Those estimates should not have a negative impact on performance e.g. longer execution times, extra complexity. Then, they should be obtained using only known information of the job execution such as cluster capacity, input size, among other known execution parameters.

Therefore, we would like to explore different approaches to extract the relationships between input parameters and measured performance i.e. execution time. Using statistical methods to identify these relationships, we could extrapolate "{\it what-if}" workload scenarios using the statistical model created as a functional instance of the whole system. Our interest lies specifically on join operations because they are  generally  more time-consuming, and they are still user-driven in the Pig system.


\section{Join types in the PIG framework}
\label{sec::join_types_pig}

The {\it join} operation tries to match records from two distinct input sets. Join execution is one of the workhorses of data processing and likely to be present on the majority of {\tt PigLatin} scripts. There are four types of join implemented in the Pig Framework: {\tt hash join}, {\tt fragment-replicate join}, {\tt merge join}, and {\tt skew join}. We will evaluate only the first three because the skew joins depend on data with a skewed distribution. We would have to know {\it a priori} the underlying data distribution, which is not possible most of the time.

\subsection{Hash Join}
\label{subsec::hash_join}
This is the default join operator in the Pig framework. The main idea of this join is to use keys for each input. Then, when those keys are the same, the two records will be joined and the records that did not fulfill the condition are dropped. Fig~\ref{fig::hash_join} illustrates this main idea.

Pig implements this strategy by tagging each record with which input it came from in the map phase, and using the join key as the shuffle key. This type of join needs a reduce phase where all the records with the same value for the join key are collected together, and then it performs a cross product between the records from both inputs. Pig tries to minimize memory usage by making the records of the first input arrive first, cache them, so when the records of the second input arrive, they can be probed against each record from the left side to produce an output record~\cite{pig-prog}.

\begin{figure}[!htb]
\hspace{0.6cm}
  \includegraphics[scale=0.45]{./images/hash_joinBB.eps}
  \caption{MapReduce Hash Join}
  \label{fig::hash_join}
\end{figure}

An important feature of the hash join is that when using a multi-way join all the inputs except the last one will be held in memory. So, the last input can be streamed through. In this way, it is a good practice to place at last the input with more records per given value of the key. Thus, a better memory usage will be made and the Pig script's performance can be improved.

The code below shows how this type of join is implemented by a PigLatin script. The example  shows the sequence of steps to perform a join between two relations, {\it inventory} and {\it item}. The first line shows how to load the "{\it inventory.dat}" file from the "{\it pigData}" folder, using the '\(|\)' character as column delimiter. The second line shows the same loading operation, but for the "{\it item.dat}" relation. And the third line shows the join operation between both relations.

{\fontsize{3mm}{5mm} \selectfont
 \begin{verbatim}
  --hash.join.pig
  inventory = LOAD 'pigData/inventory.dat' using PigStorage('|') AS
              (inv_date_sk:int, inv_item_sk:int,
              inv_warehouse_sk:int, inv_quantity_on_hand:int);

  item = LOAD 'pigData/item.dat' using PigStorage('|') AS
         (i_item_sk:int, i_item_id:chararray, i_rec_start_date:chararray,
         i_rec_end_date:chararray, i_item_desc:chararray);

  join_inventory_item = JOIN inventory BY inv_item_sk, item BY i_item_sk;
 \end{verbatim}
}


\subsection{Fragment-Replicated Join}
\label{subsec::fr_join}

We could run a routine task and perform lookups using a smaller input data. For instance, if we would have to process all the sales for a long period of time, where we would use the item's description rather than the item's code number to identify it. It would be easier to load all of our items into memory and then translate items' codes into their descriptions. This is because we have less items than the total number of sales.
We could avoid a reduce phase by sending the smaller relation to every working node
and performing the join locally without copying data through the network.
This join strategy is called {\it fragment-replicate join} because we fragment one file (the bigger one, whose fragments will be processed by the mappers) and replicate the smaller one (that will be replicated to all the mappers).

In order to implement this join strategy, Pig takes advantage of a tool the Hadoop framework provides, the {\it distributed cache}. This tool enables us to send a file to the working nodes, and to pre-load it onto the local disks, so the mappers or reducers can access it if needed. The main advantage of using this tool is not straining HDFS when a small file is needed by many mappers. For instance, if a fragment-replicate join would need 100 mappers, then opening a file stored in HDFS from 100 different nodes at a specific moment will certainly put pressure on the NameNode and on the nodes which have the replicated blocks of that file. Such situations are the motivations for using the distributed cache so no extra pressure is put on the HDFS. Another situation where it is very useful is when multiple mappers are executed on the same working node because all these tasks can share the files in the distributed cache. Therefore, a file has to be copied a fewer times around the network.

\begin{figure}[!htb]
\centering
  \includegraphics[scale=0.4]{./images/fr_joinBB.eps}
  \caption{MapReduce Fragment-Replicated Join}
  \label{fig::fr_join}
\end{figure}

In addition, Pig executes a {\tt MapReduce} job to pre-process the file in order to get it ready for sharing through the distributed cache. If there are other operations such as filtering, then these operations are also part of this initial job, so the file passed to the working nodes is as small as it can be. Therefore, the join operation itself will be done in the second map-reduce job.

The code below shows how this type of join is implemented by a PigLatin script. It is similar to the example presented for the hash join operator, but to tell Pig to use this specific type of join, we have to specify the type of join to use. We accomplish this by adding the {\tt "USING '{\it replicated}'"} hint.


{\fontsize{3mm}{5mm} \selectfont
 \begin{verbatim}
    --frag-rep.join.pig
    web_sales = LOAD 'pigData/web_sales.dat' using PigStorage('|') AS
                (ws_sold_date_sk:int, ws_sold_time_sk:int,
                 ws_ship_date_sk:int, ws_item_sk:int);

    item = LOAD 'pigData/item.dat' using PigStorage('|') AS
           (i_item_sk:int, i_item_id:chararray, i_rec_start_date:chararray,
            i_rec_end_date:chararray, i_item_desc:chararray);

    join_ws_item = JOIN web_sales BY ws_item_sk, item BY i_item_sk USING 'replicated';
 \end{verbatim}
}

When writing this type of join, we must be careful enough to always put the smaller input as the second one. This is because the second input listed will be always loaded into memory. If Pig cannot load it into memory, the join query will fail.

\subsection{Merge Join}
\label{subsec::merge_join}

The sort-merge join is a %known
join strategy in traditional relational databases and consists in sorting both inputs on the join key and
scan them simultaneously performing the join. Sorting would require a full {\tt MapReduce} job, as Pig's default join does; therefore it does not become the most efficient strategy. In spite of that, if both inputs are already sorted on the join key, then this join strategy could be the most efficient. In this case, both inputs could be opened in the map phase and iterate over them. That is why this strategy is known as merge join, but the sort has to be done before performing the join. The following code shows how this type of join is implemented by a PigLatin script:

\vspace*{-0.3cm}

{\fontsize{3mm}{5mm} \selectfont
 \begin{verbatim}
  --merge.join.pig
  catalog_sales = LOAD 'pigData/catalog_sales.dat' using PigStorage('|') AS
                  (cs_sold_date_sk:int, cs_sold_time_sk:int, cs_ship_date_sk:int,
                   cs_bill_customer_sk:int, cs_bill_cdemo_sk:int, cs_bill_hdemo_sk:int);

    date_dim = LOAD 'pigData/date_dim.dat' using PigStorage('|') AS
               (d_date_sk:int, d_date_id:chararray, d_date:chararray);

    join_date_cs = JOIN date_dim BY d_date_sk, catalog_sales
                   BY cs_sold_date_sk USING 'merge';
\end{verbatim}
}

Fig~\ref{fig::merge_join} shows the Merge Join job schema execution,
which runs a {\tt MapReduce} job to get samples of the second input in order to build an index which will be used as the value of the join key. These samples are the first records of every input split i.e. from each HDFS block, indeed
a very fast sampling process.
Pig will use another {\tt MapReduce} job which will use the first input (in this case date\_dim) as its input. When each mapper reads the records in its split, it will take it and look it up in the index built by the previous job. It will look for the entry that is less than the value it read from the first input. Then, it will open the specific split of the second input pointed by the index entry.

\begin{figure}
  \centering
  \includegraphics[scale=0.4]{./images/merge_joinBB.eps}
  \caption{MapReduce Merge Join}
  \label{fig::merge_join}
\end{figure}

Once the correct block of the second input is opened, Pig will start checking for a match of the index. When it finds a match, it gets all the records that matched into memory, and then perform the join. It then gets another record of the first input, and if the key is the same as the last one, then it performs the join. If it is not, it then checks the index to see which the proper file split is to get it and to check again. If it cannot find a match within the second input, it simply advances to another record of the first input and checks again. This type of join only supports currently two way joins and inner joins. It is also more efficient than a hash join because it can be done without a reduce phase.

Our work aims to explore statistical methods to help systems take decisions on which join operator to use, but without user's hints. We will present our approach and practical results in the next section.


\section{Statistical Approach and Practical Evaluation}
\label{sec::stati}

The Pig system allows the use of four different types of join operators (merge, hash, fragment-replicated, and skew joins). However, to use them the user needs to pass a hint in PigLatin to the optimizer specifying the type of join to be used. This type of communication with the optimizer might be suitable for some experienced users who probably can choose wisely between these operators. However, for new users the constraints and advantages of each type of join operator may not be so clear. In this way, new users can be mistaken when choosing the correct type of join operator, which can lead to bad performances and misused of computational resources.

One of the main goals of this research work is to investigate possible statistical methods that could fit well for use within our system architecture depicted in Figure~\ref{fig::system_architecture}. We may investigate the parameters involved in creating this statistical model, specifically the design decisions to model and evaluate execution times for the different types of PigLatin join operators. This model would help a rule-based optimizer to choose the most appropriate join operator for each situation.

\begin{figure}[!htb]
\centering
  \includegraphics[scale=0.4]{./images/SystemPerformanceOfflineModelBB.eps}
  \caption{System Architecture for Our Statistical Model}
  \label{fig::system_architecture}
\end{figure}

In that way, we see a possibility of applying studied concepts from the database field into large scale data processing. More specifically, we see a chance in integrating these prediction capabilities into the Pig system~\cite{apache-pig} to improve its query execution capabilities. Figure~\ref{fig::system_architecture} shows how integrating these concepts into the system would be like. The left part of the figure exhibits the model constructor of the system. This component should use only {\it a priori} information about the input relations and the operations to create a model that characterizes them.
Thus, we can use the model to obtain query execution times estimates beforehand. The right part of the figure shows how this {\it a priori} information can
be used by the
optimizer in order to make good decisions
and create better execution plans.


\subsection{Choice of a Data Set}
\label{pract}

Research has also been done to evaluate different aspects of these non-traditional data stores.
Shi et al.~\cite{Shi2010} benchmark five different tasks:
data load, grep query, range query, aggregation and fault tolerance.
Cooper et al.~\cite{Cooper2010} supply several workloads with different combinations of insert, read, update and scan operations. Nevertheless, all
these workloads
focus on  systems like PNUTS, which provide online read and write access to data.
Analytical systems are not
mentioned in these research works as they try to evaluate different aspects of cloud data management that make the data sets used
specialized for their specific benchmark scenarios.

As there is no data set available for testing {\tt MapReduce} jobs\footnote{although there is the SWIM (https://github.com/SWIMProjectUCB/SWIM/wiki) ongoing project.}, we have decided to start with a popular benchmark from the database community. There are several benchmarks used, some specific for transaction processing and others for analytical processing.
We have not chosen any of the OLTP benchmarks because their main motivation
is to evaluate transactions. Moreover, the MapReduce model has been designed for analytical workloads due to its batch-oriented nature and not for single
operations. We have considered then TPC-DS~\cite{tpcds}, which is a decision support workload
from the {\it Transaction Processing Performance Council}~\cite{tpc}.
Raghunath et al.~\cite{Nambiar:2006:MT:1182635.1164217} explain that using both synthetic and real world data for designing the TPC-DS has many advantages over benchmarks that use only one type of data. This is because synthetic data sets built using studied distributions such as the Normal or Poisson distributions have many positive points, but they are not well suited for dynamically substituting bind variables. The {\tt TPC-DS} utilizes traditional synthetic distributions, yielding uniformly distributed values with a Gaussian distribution.

Scaling a data set can be done in two different ways: The number of tuples in the data set is expanded, but the underlying value sets (domains) remain static, or number of tuples remains fixed, but the domains used to generate them are expanded. In the case of {\tt TPC-DS}, an hybrid approach was chosen, so most table columns employ data set scaling instead of domain scaling, especially fact table columns; Some small tables' columns use domain scaling. As a result of this approach, fact tables scale linearly with the scale factor while dimensions scale sub-linearly.

We decided to use the {\tt TPC-DS} data because we could have some information about the data set as well knowing how scaling it would affect the underlying data distribution. The data obtained from the {\tt TPC-DS} were plain text files which we load into our local HDFS to perform our tests.

Due to the fact that the {\tt TPC-DS} has been designed as a
 relational database benchmark, queries outlined in~\cite{tpcds} were meant to evaluate a data warehouse and are not suitable for our experiments. Queries were specifically designed for testing different aspects of DBMS and characterize general purpose queries. Our work aims to characterize a specific (join) operator.

We decided to analyze the database schema proposed by~\cite{Nambiar:2006:MT:1182635.1164217} for the {\tt TPC-DS}, extract the relationships between tables and create queries based on such relationships. For example, Figure~\ref{fig::SystemPerformanceDataSet} shows all the relationships between the table {\it Store\_Sales} and the ones related to it such as some fact tables ({\it Item, Promotion,} etc.) and some dimension tables ({\it Time\_Dim, Date\_Dim}). We generated join queries using such entity's relationships, but keeping in mind the restrictions posed by each type of join.
We have created 96 PigLatin queries using a Fragment Replicate Join,
23 PigLatin queries using a Merge Join and 102 queries using a Hash Join. More details in \cite{Mogrovejo2011}.



\begin{figure}[!htb]
\centering
  \includegraphics[scale=0.3]{./images/SystemPerformanceDataSetBB.eps}
  \vspace*{-0,5cm}
  \caption{Relationships from some {\tt TPC-DS}' tables.}
  \label{fig::SystemPerformanceDataSet}
\end{figure}


\subsection{Model Construction}
\label{model}

The parameters used to build the multiple regression analysis are chosen having in mind that only {\it a priori} knowledge about the job execution must be used.
Statistical techniques, such as multiple regression analysis, aim to find relationships between workload features (e.g. number of computers used, network latency, input size) and performance features  (e.g. execution time, I/O metrics). In our case, we have considered those features listed in Table~\ref{tab::workload_features}.

%\vspace*{-0,5cm}

\begin{table}
 \caption{Workload features.}
  \begin{center}
    \begin{tabular}{|c|c|c|}
      \hline {\bf Feature} & {\bf Type} & {\bf Multiple Regression}\\

      \hline Job name & Categorical & Not used \\

      \hline Number of Map processes &  Numeric & Used \\

      \hline Number of Reduce processes & Numeric & Not Used \\

      \hline Map input bytes & Numeric & Used \\

      \hline Map input records & Numeric & Used \\

      \hline Join Selectivity & Numeric & Used \\

      \hline
    \end{tabular}
  \end{center}
   \label{tab::workload_features}
\end{table}


We could have chosen other statistical approaches. Indeed, {\it clustering techniques}, are common for  data analysis in different fields e.g. data mining, pattern recognition, image analysis, and bioinformatics.  These techniques consist in assigning a set of observations into subsets, called clusters, based on the similarity of multiple features. This is considered a method of unsupervised learning because it tries to find hidden structures in unlabeled data defining a distance measure between points in the data set. Partition clustering algorithms such as {\it Kmeans}~\cite{kmeans} are used to identify a set of points that are the nearest ones to a test data point, but they are not suitable for performance modeling because the clustering process would have to be applied on the workload features and on the performance features separately. The similar points with respect of the workload features do not actually reflect the points that cluster together with respect to some other performance features.

Another approach is known as Principal Component Analysis (PCA) and is mostly used for exploratory data analysis, but also for making predictive models. It is the oldest technique for finding relationships in a multivariate data set~\cite{hotelling-33}. The main idea of PCA is to identify dimensions of maximal variance in a data set and to project raw data onto these dimensions by performing eigenvalue decomposition of the data covariance matrix. The main drawback of using PCA for performance modeling is that the dimensions of maximal variance in workload found do not resemble dimensions that most affect performance. In addition, PCA is not able to correlate workload features with performance ones which makes it not suitable for our needs.
Canonical Correlation Analysis (CCA) is a generalization of PCA~\cite{hotelling36cca}. The main idea is that it evaluates pair-wise data set in order to find dimensions of maximal correlation between the workload features and the performance features. This method and its {\it kernelized} variant (Kernel Canonical Correlation Analysis) are other statistical machine learning techniques that we find interesting for future work. Kernel Canonical Correlation Analysis (KCCA)~\cite{DBLP:journals/jmlr/BachJ02} captures the similarity between features using a kernel function, and the correlation analysis is done on pair-wise distances, and not on the raw data itself. Studying these methods is part of our future work.

Finally, in linear multiple regression analysis, the goal is to predict, knowing the measurements collected on {\it N} subjects, a dependent variable {\it Y} from a set of {\it J} independent variables denoted {\it \{X\(_1\),..,X\(_j\),..,X\(_J\)\}}. In this manner, we can map these independent variables to each workload feature and treat our performance metric as a dependent variable '{\it y}'.  The goal of this statistical method is to solve the equation {\it a\(_1\)X\(_1\) + a\(_2\)X\(_2\) + ... + a\(_n\)X\(_n\) = y} for all the coefficients {\it a\(_i\)}.

In this research work we have decided to adopt multiple regression analysis because (i) it can make good predictions from multiple predictors (ii) it avoids dependance on a single predictor and (iii) it is based on optimal combination of predictors. Furthermore, our work aims to predict a given performance metric, namely, query execution time. Multiple regression analysis becomes an interesting tool for characterizing join operators because our model also has many different parameters to take into consideration but only one feature to be predicted. In what follows, we will explore multiple regression to study PigLatin join's performances.


\subsection{Multiple regression analysis}

Our workload features will be used as our independent variables ({\it \{X\(_1\),..,X\(_j\),..,X\(_J\)\}}), and the query execution time is
our dependent variable ({\it y}). We then use the resulting formulas to predict other queries execution times.
An additional parameter that we have added to our model is the join selectivity for each job execution. It enables the estimation of jobs' output
with a close relationship to the amount of work to be done. Due to space limitations, the reader may check \cite{Mogrovejo2011} for further details about it.

One important thing here is that we must be aware of certain problems that can be caused due to our parameters. For example, the parameter representing the number of {\it Reduce processes} is linearly dependent on the jobs executed which can lead the multiple regression model to unstable estimates. The number of reducers to be used is dependent on what the output will be used for, the reduce capacity of the cluster, the amount of data needing to be reduced, and the time needed to perform the reduce operation. In our case, this parameter refers to the number of reducers that will be used in the join operation. All join algorithms used in the Pig Framework try to take advantage of all resources available in the cluster and to avoid the network overhead of copying data from mappers to reducers.

This parameter does not vary in our jobs' execution because the number of reducers corresponds to the capacity of our cluster. Thus, we decided not to use the number of reducers in our model construction process because it would cause unstable estimates. The number of map processes is determined by the number of HDFS blocks for the input files and can be computed by dividing the number of bytes to be processed by the size of HDFS blocks. To predict the execution time for our validating queries, we estimate the number of map processes and use it for constructing our model.

One important challenge being
is the lack of metadata in the {\tt MapReduce} computing model, also in its open-source implementation, Hadoop. Systems like Hive~\cite{apache-hive} and Cassandra~\cite{projcassandra} are trying to incorporate basic statistics into their systems. Once accomplished, these systems will be able to develop efficient cost models for handling big data.

Likewise, Pig framework~\cite{apache-pig} lacks of metadata like column definitions or relation's basic statistics. The Pig framework
also inherits some drawbacks of HDFS design
as the simple metadata kept by the {\tt Namenode}. It maintains information about the location of each block  and to which file it belongs, guaranteeing that two block files never have the same block ID. In spite of the fact that keeping simple statistics make the system faster, their absence make the system more rigid to perform different operations than the established ones. Therefore, we need to obtain these basic statistics manually in order to estimate the parameters we need for our model (see \cite{Mogrovejo2011}.


\subsection{Experiments and Practical Results}
\label{sec::experimental_setup}


We used the queries designed for each specific type of join to create two different set of experiments. The first experiment consists in performing cross validation of the model generated for each type of join. The second experiment tries to simulate a real workload where many similar queries are executed in short periods of time. We have decided to create such workload within Pig Framework based on previous observations made by similar research on parallel computer systems. However, due to space limitations, we will discuss only the cross validation experiments in this article. The reader may refer to \cite{Mogrovejo2011} for the complete set of experiments of this research work.

Cross validation is used for evaluating how the results of a statistical analysis will generalize to an independent data set. In {\it K-fold cross-validation}, the original sample is randomly partitioned into {\it K} subsamples. Out of the {\it K} subsamples, a single one is retained as the validation data for testing the model and the remaining $K - 1$ subsamples are used as training data. The process is then repeated {\it K} times ({\it folds}), with each of the {\it K} subsamples used exactly once as the validation data. The {\it K} results from the folds  can be combined to produce a single estimation. The advantage of this method over repeated random sub-sampling is that all observations are used for both training and validation and each observation is used for validation exactly once. Our model was validated using a 3-fold validation due to the relatively small size of our data set, and to the long running times of our tests.

We describe and discuss next the results obtained from executing cross-validation on our data set of queries. The figures presented show the difference between the real query execution times against the estimated query execution time for the 3-fold validation process. We performed such validation on three types of join operators implemented by the PigLatin framework.

{\bf Fragment Replicated Joins:} We executed 96 queries of the fragment-replicated joins but using 32 queries for validation on each fold. We plot the results
in two figures (Figure~\ref{fig::FRJoin_AllData_WvsWOvsReal} and Figure~\ref{fig::FRJoin_80th_WvsWOvsReal}).
Figure~\ref{fig::FRJoin_AllData_WvsWOvsReal} shows the query execution times ({\it QE Time}) and the estimates obtained using the join selectivity parameter ({\it QE Time With JSel}), and not using the join selectivity parameter ({\it QE Time WO JSel}). A first observation is that the variation between the estimates gotten using or not the join selectivity parameter could almost pass unnoticed. The error within our join selectivity parameter prevents the model to take full advantage of it. Two jobs  took longer than the rest of them because more data were moved through the network.

\begin{figure}[!htb]
    \begin{minipage}{.45\linewidth}
    \includegraphics[width=\textwidth]{./images/FRJoin_AllData_WvsWOvsReal.eps}
    \caption{Query execution times vs time estimates for fragment replicated joins.}
   \label{fig::FRJoin_AllData_WvsWOvsReal}
    \end{minipage}
    \hspace{0.5cm}
    \begin{minipage}{.45\linewidth}
    \includegraphics[width=\textwidth]{./images/FRJoin_80th_WvsWOvsReal.eps}
    \caption{Fragment replicated joins: execution times vs time estimates for the 80th percentil of the executions.}
    \label{fig::FRJoin_80th_WvsWOvsReal}
    \end{minipage}
\end{figure}

Figure~\ref{fig::FRJoin_80th_WvsWOvsReal} shows the query execution times from the 80th percentile of the jobs and their execution time estimates.
By restricting our results we can omit  jobs that suffered from performance problems, execution environment changes and poor linear models. For example, the jobs that in our case suffered from network latency, a key piece for the {\tt MapReduce} framework.

Finally,  Figure~\ref{fig::FRJoin_80th_ErrorPercen} shows a comparison between the relative error from the two different estimates, either or not using the join selectivity parameter. The error goes from less than 2\% up to 205\% for all jobs, but for the 80th percentile the error ranges from less than 2\% until almost 85\%.

\begin{figure}[!htb]
    \begin{minipage}{.45\linewidth}


  \includegraphics[width=\textwidth]{./images/FRJoin_80th_ErrorPercen.eps}
  %\vspace{20}
  \caption{Fragment replicated joins: error percentage for the 80th percentil of the executions.}
    \label{fig::FRJoin_80th_ErrorPercen}
    \end{minipage}
    \hspace{0.5cm}
\ \begin{minipage}{.45\linewidth}


  \includegraphics[width=\textwidth]{./images/MrgJoin_80th_ErrorPercen.eps}
  %\vspace{20}
  \caption{Merge join: error percentage for the 80th percentil of the executions.}
    \label{fig::MrgJoin_80th_ErrorPercen}
    \end{minipage}
\end{figure}


{\bf Merge joins:} We executed the 23 queries created for this specific join operator and used 8 queries for validation. One of the biggest problems for building such queries is the difficulty of creating join operations between ordered data when most data is unordered. This type of join needs both relations to be in ascending order in the join key. Many relations of our data set have the primary key - foreign key relationship, they do share the same key but they are not sorted in the same way on both relations.

Figure~\ref{fig::MrgJoin_AllData_WvsWOvsReal} shows the differences between the actual query execution times ({\it QE Time}) against the estimates obtained using the model generated with and without the join selectivity parameter ({\it QE Time With JSel}, and {\it QE Time WO JSel} respectively).
The estimates obtained by using the join selectivity parameter did not differ considerably from one to the other. There were two estimates in which the model predicted long execution times. The first one is due to a self-join of the third biggest relation (226MB).
Nevertheless, the query was executed in a very fast manner, only 21 seconds to be completed. We hypothesize that this was due to the fact that the column used for joining is sufficiently skewed for the implementation of the merge join operator. The other high point in the figure is also a join involving the third biggest relation.
This join execution time was expected due to the size of the input and to the environment variables (e.g. network latency). That is why the predictions do not differ too much from what the time actually took. It differs 7\% with the prediction that did not use the join selectivity parameter and 13\% with the one that used it.

\begin{figure}[!htb]
    \begin{minipage}{.45\linewidth}


  \includegraphics[width=\textwidth]{./images/MrgJoin_AllData_WvsWOvsReal.eps}
  %\vspace{20}
  \caption{Query execution times vs time estimates for merge joins.}
        \label{fig::MrgJoin_AllData_WvsWOvsReal}
    \end{minipage}
    \hspace{0.5cm}
    \begin{minipage}{.45\linewidth}

  \includegraphics[width=\textwidth]{./images/MrgJoin_80th_WvsWOvsReal.eps}
  %\vspace{20}
  \caption{Merge joins: execution times vs time estimates for the 80th percentil of the executions.}
    \label{fig::MrgJoin_80th_WvsWOvsReal}
    \end{minipage}
\end{figure}


Figure~\ref{fig::MrgJoin_80th_WvsWOvsReal} shows the actual query execution times against the estimated values for the 80th percentile of the queries. In this case our error ranges from 7\% to 378\% for the model not using the join selectivity operator, and for the model using the join selectivity operator its error ranges from less than 14\% to 423\%. This difference in the error from both models is due to the fact that the join selectivity parameter has already the error of the cardinality estimate in addition to the estimate itself.



Figure~\ref{fig::MrgJoin_80th_ErrorPercen} shows the percentage error obtained for the 80th percentile of the queries using both models (applying the join selectivity parameter ({\it Error With JSel}), and not applying it ({\it Error WO JSel})). The highest points in the graphic are due to the fact that the merge join operator takes advantage of not having to read all of the big relations. Rather,  it samples one of the relations and creates a sparse index on it. Then, it uses this index to access the block directly at probing time. This makes this operator the least memory intensive one because it does not have to load a whole block of data into memory. An interesting thing here is that the model using the join selectivity parameter seems to have a more stable behavior compared to the one that uses the join selectivity parameter. This is also due to the error accumulated into the join selectivity parameter, which ends up not being so useful for our discussed model.


{\bf Hash Joins:} We executed 102 queries of the hash join operator using 34 queries for validation. Similarly to the other join operators, we will describe the figures comparing the actual execution time against the estimated execution time using the models built. Figure~\ref{fig::HshJoin_AllData_WvsWOvsReal} shows the difference between the real execution time ({\it QE Time}) and the prediction values either using the join selectivity parameter ({\it QE With JSelectivity}) or not using it ({\it QE WO JSelectivity}) for building the model. One important observation about it is that the actual execution times where really fast compared to the estimated times. We think this is due to the fact that this operator relies heavily on the hardware i.e. on main memory for maintaining and probing tuples from one relation against the other. Even though we decide to keep only the 80th percentile of all the queries,  Figure~\ref{fig::HshJoin_80th_WvsWOvsReal} does not change substantially. The real execution times are lower when compared to the estimated execution times.


\begin{figure}[!htb]
    \begin{minipage}{.45\linewidth}


  \includegraphics[width=\textwidth]{./images/HshJoin_AllData_WvsWOvsReal.eps}
  %\vspace{20}
  \caption{Query execution times vs  time estimates for hash joins.}
    \label{fig::HshJoin_AllData_WvsWOvsReal}
    \end{minipage}
    \hspace{0.5cm}
    \begin{minipage}{.45\linewidth}
  \includegraphics[width=\textwidth]{./images/HshJoin_80th_WvsWOvsReal.eps}
  %\vspace{20}
  \caption{Hash join: query execution times vs time estimates for the 80th percentil of the executions.}
    \label{fig::HshJoin_80th_WvsWOvsReal}
    \end{minipage}
\end{figure}




Therefore, we were expecting high errors in our estimates. Figure~\ref{fig::HshJoin_80th_ErrorPercen} compares the estimate's error when
using or not join selectivity as a parameter. The difference between both estimates is that the join selectivity parameter causes the query execution time to vary along the plot. The error considering this parameter is probably too high and our model cannot take advantage.

\begin{figure}[!htb]
    \begin{minipage}{.45\linewidth}
  \includegraphics[width=\textwidth]{./images/HshJoin_80th_ErrorPercen.eps}
  %\vspace{20}
  \caption{Hash join: error percentage for the 80th percentil of the executions.}
    \label{fig::HshJoin_80th_ErrorPercen}
   \end{minipage}
    \hspace{0.5cm}
    \begin{minipage}{.45\linewidth}
  \includegraphics[width=\textwidth]{./images/HshJoin_QE_Variability.eps}
  %\vspace{20}
  \caption{Hash join query execution variability.}
    \label{fig::HshJoin_QE_Variability}
   \end{minipage}
\end{figure}

This huge difference between our estimates and the actual values motivated us to search for an explanation. That is the reason why we decided to look into the actual jobs executions. In Figure~\ref{fig::HshJoin_QE_Variability} we show the variance between the different query executions of the cross-fold validation. We recorded each execution time from each run and then compared them against the other query executions. As we can see, the job execution time varies in hundreds of orders of magnitude from each different execution. This means that this join operator heavily depends on how the cluster is being used in the specific moment when the hash join is executed. It varies because this join operator always loads into memory the relation from the left and the other one is streamed through. Besides that, the Pig framework expands four times the size of data when loading it from disk into memory.
This join operator is the less stable compared to other operators because it heavily depends on main memory which can vary along query execution.




\section{Conclusions}
\label{sec::conclusions}

PigLatin needs to receive user hints in order to enable its compiler to choose correctly between the different types of join it has implemented.
Thereby, creating a statistical model to characterize the different types of joins in order to estimate its execution time beforehand would mean that the most efficient join operator could be chosen automatically.

The main contributions of this article are (i) a statistical model for query execution time prediction based on multiple regression applied to join queries in the Pig Framework
and (ii) the construction of a data set for PigLatin join queries with a higher number of queries for each type of join, consequently, a more realistic evaluation.

We perceived through our experiments that we could expand the number of queries to be executed as well as their variety. In this work we decided to use the standard queries and data model originally defined by TPC-DS as it is an accepted database community benchmark.  This has led us to obtain a relatively small volume of join operations' data that resulted in an overfitted model.  Hence, we are working on building a bigger data set and, at the same time, studying some different approaches to avoid over fitting of our regression model (e.g., shrinkage methods).
We also think about studying the {\tt TPC-WS} - not OLAP - as a data set. In addition, we believe that exploring other statistical methods are very appealing to the approach presented here. Kernel methods such as those used in~\cite{Ganapathi:EECS-2009-181} show great advantages compared to regular statistical methods. Locally weighted learning~\cite{Kavulya:2010:ATP:1844765.1845224} and instance based learning~\cite{4228004} seem also suitable.

\bibliographystyle{jidm}
\bibliography{sbbd2013renato}

\begin{received}
\end{received}

\end{document}
