RIoTBench: A Real-time IoT Benchmark for Distributed Stream Processing Platforms
Extracted source text · arXiv 1701.08530
\documentclass{article}
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Inline comments. Pick initials and color of your choice. \ysnote{} refers to Yogesh's note.
%
\usepackage[svgnames, x11names]{xcolor}
\newcommand{\ysnote}[1]{ {\textcolor{magenta} { ***Yogesh: #1 }}}
% \newcommand{\ysnoted}[1]{ {\textcolor{gray} {***TODO Later: \sout{#1}}}}
\newcommand{\ysnoted}[1]{}
\newcommand{\jknote}[1]{ {\textcolor{blue} { ***Jayanth: #1 }}}
\newcommand{\drnote}[1]{ {\textcolor{orange} { ***Dreamer: #1 }}}
\newcommand{\note}[1]{ {\textcolor{red} {#1}}}
%
%---------------------------------------------------------
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Change tracking for article revisions. Added, Deleted, Replaced, or Modified content.
%
\usepackage[normalem]{ulem} % required for strikeout font
\newcommand{\modc}[1]{{\textcolor{blue}{#1}}}
\newcommand{\addc}[1]{{\textcolor{teal}{#1}}}
\newcommand{\delc}[1]{ {\textcolor{gray} {\sout{#1}} }}
%\newcommand{\delc}[1]{} % uncomment this (and comment above line) to ignore showing deletion
\newcommand{\repc}[2]{ {\textcolor{gray} {\sout{#1}} }{\textcolor{teal} {#2}}}
%\newcommand{\repc}[2]{{\textcolor{teal}{#2}}} % uncomment this (and comment above line) to ignore showing deletion
%
%---------------------------------------------------------
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% When using XML fragments, using pretty-print is helpful.
%
\usepackage{listings}
% \usepackage{color}
% \definecolor{gray}{rgb}{0.4,0.4,0.4}
% \definecolor{darkblue}{rgb}{0.0,0.0,0.6}
%\definecolor{maroon}{rgb}{0.5,0,0}
% \definecolor{cyan}{rgb}{0.0,0.6,0.6}
\lstset{
basicstyle=\ttfamily,
columns=fullflexible,
showstringspaces=false,
commentstyle=\color{gray}\upshape
}
\lstdefinelanguage{XML}
{
basicstyle=\ttfamily\footnotesize,
morestring=[b]",
moredelim=[s][\bfseries\color{Maroon}]{<}{\ },
moredelim=[s][\bfseries\color{Maroon}]{</}{>},
moredelim=[l][\bfseries\color{Maroon}]{/>},
moredelim=[l][\bfseries\color{Maroon}]{>},
morecomment=[s]{<?}{?>},
morecomment=[s]{<!--}{-->},
commentstyle=\color{gray},
stringstyle=\color{blue},
identifierstyle=\color{red}
% morekeywords={type,id,value,impl}% list your attributes here
}
%
%---------------------------------------------------------
\usepackage[pdftex]{graphicx}
% declare the path(s) where your graphic files are
%\graphicspath{{./figures/}}
% and their extensions so you won't have to specify these with
% every instance of \includegraphics
\DeclareGraphicsExtensions{.pdf}
% *** MATH PACKAGES ***
%
% A popular package from the American Mathematical Society that provides
% many useful and powerful commands for dealing with mathematics.
\usepackage[cmex10]{amsmath}
\usepackage{amssymb}
\usepackage{mathtools}
% *** SUBFIGURE PACKAGES ***
\usepackage[]{subfig}
% *** SPECIALIZED LIST PACKAGES ***
%
\usepackage{algorithmicx}
\usepackage{algpseudocode}
\usepackage[ruled]{algorithm}
\definecolor{light-gray}{gray}{0.75}
\algrenewcommand{\algorithmiccomment}[1]{\hskip3em{{\footnotesize \textcolor{light-gray}{$\blacktriangleright$}}} #1}
% This package provides an algorithmic environment fo describing algorithms.
% You can use the algorithmic environment in-text or within a figure
% environment to provide for a floating algorithm.
\usepackage{multirow}
% *** ALIGNMENT PACKAGES ***
%
%\usepackage{array}
% Frank Mittelbach's and David Carlisle's array.sty patches and improves
% the standard LaTeX2e array and tabular environments to provide better
% appearance and additional user controls. As the default LaTeX2e table
% generation code is lacking to the point of almost being broken with
% respect to the quality of the end results, all users are strongly
% advised to use an enhanced (at the very least that provided by array.sty)
% set of table tools. array.sty is already installed on most systems. The
% latest version and documentation can be obtained at:
% http://www.ctan.org/tex-archive/macros/latex/required/tools/
% *** FLOAT PACKAGES ***
%
%\usepackage{fixltx2e}
% fixltx2e, the successor to the earlier fix2col.sty, was written by
% Frank Mittelbach and David Carlisle. This package corrects a few problems
% in the LaTeX2e kernel, the most notable of which is that in current
% LaTeX2e releases, the ordering of single and double column floats is not
% guaranteed to be preserved. Thus, an unpatched LaTeX2e can allow a
% single column figure to be placed prior to an earlier double column
% figure.
% *** PDF, URL AND HYPERLINK PACKAGES ***
%
\usepackage{hyperref}
\usepackage{xspace}
\usepackage[nocompress]{cite}
% IEEETrans class fix for enumitem. provide for legacy IED commands/lengths when possible
% http://comments.gmane.org/gmane.editors.lyx.general/68611
\let\labelindent\relax
\usepackage{enumitem}
% correct bad hyphenation here
\hyphenation{super-computer}
% define repetitive complex fragments here
\newcommand{\floe}{$\mathcal{F}{\ell}{o}{\varepsilon}$\xspace}
\newcommand{\prc}{\mathcal{P}}
\newcommand{\chn}{\mathcal{C}}
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% generate lorum ipsum placeholder text
\usepackage[english]{babel}
\usepackage{blindtext}
\newcommand{\blindtextc}{\color{gray}\blindtext\color{black}}
\newcommand{\Blindtextc}{\color{gray}\Blindtext\color{black}}
% Allow footnote in front page for \thanks
%\IEEEoverridecommandlockouts
\begin{document}
%
% paper title
% can use linebreaks \\ within to get better formatting as desired
\title{RIoTBench: A Real-time IoT Benchmark for Distributed Stream Processing Platforms}
%
%
% author names and IEEE memberships
% note positions of commas and nonbreaking spaces ( ~ ) LaTeX will not break
% a structure at a ~ so this keeps an author's name from being broken across
% two lines.
% use \thanks{} to gain access to the first footnote area
% a separate \thanks must be used for each paragraph as LaTeX2e's \thanks
% was not built to handle multiple paragraphs
%
%
%\IEEEcompsocitemizethanks is a special \thanks that produces the bulleted
% lists the Computer Society journals use for "first footnote" author
% affiliations. Use \IEEEcompsocthanksitem which works much like \item
% for each affiliation group. When not in compsoc mode,
% \IEEEcompsocitemizethanks becomes like \thanks and
% \IEEEcompsocthanksitem becomes a line break with idention. This
% facilitates dual compilation, although admittedly the differences in the
% desired content of \author between the different types of papers makes a
% one-size-fits-all approach a daunting prospect. For instance, compsoc
% journal papers have the author affiliations above the "Manuscript
% received ..." text while in non-compsoc journals this is reversed. Sigh.
\author{Anshu Shukla, Shilpa Chaturvedi and Yogesh Simmhan \\
\normalsize{\emph{Department of Computational and Data Sciences}}\\
\normalsize{\emph{Indian Institute of Science (IISc), Bangalore 560012, India}}\\
\normalsize{\emph{Email: shukla@grads.cds.iisc.ac.in, shilpa@grads.cds.iisc.ac.in, simmhan@cds.iisc.ac.in}}}
\date{}
% make the title area
\maketitle
\begin{abstract}
The Internet of Things (IoT) is an emerging technology paradigm where millions of sensors and actuators help monitor and manage, physical, environmental and human systems in real-time. The inherent closed-loop responsiveness and decision making of IoT applications make them ideal candidates for using low latency and scalable stream processing platforms. Distributed Stream Processing Systems (DSPS) hosted on Cloud data-centers are becoming the vital engine for real-time data processing and analytics in any IoT software architecture. But the efficacy and performance of contemporary DSPS have not been rigorously studied for IoT applications and data streams. Here, we develop \emph{RIoTBench}, a Real-time IoT Benchmark suite, along with performance metrics, to evaluate DSPS for streaming IoT applications. The benchmark includes $27$ common IoT tasks classified across various functional categories and implemented as reusable micro-benchmarks. Further, we propose four IoT application benchmarks composed from these tasks, and that leverage various dataflow semantics of DSPS. The applications are based on common IoT patterns for data pre-processing, statistical summarization and predictive analytics. These are coupled with four stream workloads sourced from real IoT observations on smart cities and fitness, with peak streams rates that range from $500-10,000~messages/sec$ and diverse frequency distributions. We validate the RIoTBench suite for the popular Apache Storm DSPS on the Microsoft Azure public Cloud, and present empirical observations. This suite can be used by DSPS researchers for performance analysis and resource scheduling, and by IoT practitioners to evaluate DSPS platforms.
\end{abstract}
%% bare_jrnl_compsoc.tex
%% V1.3
%% 2007/01/11
%% by Michael Shell
%% See:
%% http://www.michaelshell.org/
%% for current contact information.
%%
%% This is a skeleton file demonstrating the use of IEEEtran.cls
%% (requires IEEEtran.cls version 1.7 or later) with an IEEE Computer
%% Society journal paper.
%%
%% Support sites:
%% http://www.michaelshell.org/tex/ieeetran/
%% http://www.ctan.org/tex-archive/macros/latex/contrib/IEEEtran/
%% and
%% http://www.ieee.org/
%%*************************************************************************
%% Legal Notice:
%% This code is offered as-is without any warranty either expressed or
%% implied; without even the implied warranty of MERCHANTABILITY or
%% FITNESS FOR A PARTICULAR PURPOSE!
%% User assumes all risk.
%% In no event shall IEEE or any contributor to this code be liable for
%% any damages or losses, including, but not limited to, incidental,
%% consequential, or any other damages, resulting from the use or misuse
%% of any information contained here.
%%
%% All comments are the opinions of their respective authors and are not
%% necessarily endorsed by the IEEE.
%%
%% This work is distributed under the LaTeX Project Public License (LPPL)
%% ( http://www.latex-project.org/ ) version 1.3, and may be freely used,
%% distributed and modified. A copy of the LPPL, version 1.3, is included
%% in the base LaTeX documentation of all distributions of LaTeX released
%% 2003/12/01 or later.
%% Retain all contribution notices and credits.
%% ** Modified files should be clearly indicated as such, including **
%% ** renaming them and changing author support contact information. **
%%
%% File list of work: IEEEtran.cls, IEEEtran_HOWTO.pdf, bare_adv.tex,
%% bare_conf.tex, bare_jrnl.tex, bare_jrnl_compsoc.tex
%%*************************************************************************
% the "draftcls" or "draftclsnofoot", not "draft", option
% should be used if it is desired that the figures are to be displayed in
% draft mode.
%
%\documentclass{llncs}
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Inline comments. Pick initials and color of your choice. \ysnote{} refers to Yogesh's note.
%%
%\usepackage[svgnames, x11names]{xcolor}
%\newcommand{\ysnote}[1]{ {\textcolor{magenta} { ***Yogesh: #1 }}}
%\newcommand{\tsnote}[1]{ {\textcolor{green} { ***Tarun: #1 }}}
%\newcommand{\asnote}[1]{ {\textcolor{blue} { ***Anshu: #1 }}}
%\newcommand{\Note}[1]{ {\textcolor{red} { #1 }}}
%%
%%---------------------------------------------------------
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Change tracking for article revisions. Added, Deleted, Replaced, or Modified content.
%%
%\usepackage[normalem]{ulem} % required for strikeout font
%
%\newcommand{\modc}[1]{{\textcolor{blue}{#1}}}
%\newcommand{\addc}[1]{{\textcolor{teal}{#1}}}
%\newcommand{\delc}[1]{ {\textcolor{gray} {\sout{#1}} }}
%%\newcommand{\delc}[1]{} % uncomment this (and comment above line) to ignore showing deletion
%\newcommand{\repc}[2]{ {\textcolor{gray} {\sout{#1}} }{\textcolor{teal} {#2}}}
%%\newcommand{\repc}[2]{{\textcolor{teal}{#2}}} % uncomment this (and comment above line) to ignore showing deletion
%%
%%---------------------------------------------------------
%
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% When using XML fragments, using pretty-print is helpful.
%%
%\usepackage{listings}
%% \usepackage{color}
%% \definecolor{gray}{rgb}{0.4,0.4,0.4}
%% \definecolor{darkblue}{rgb}{0.0,0.0,0.6}
%%\definecolor{maroon}{rgb}{0.5,0,0}
%% \definecolor{cyan}{rgb}{0.0,0.6,0.6}
%
%\lstset{
% basicstyle=\ttfamily,
% columns=fullflexible,
% showstringspaces=false,
% commentstyle=\color{gray}\upshape
%}
%
%\lstdefinelanguage{XML}
%{
% basicstyle=\ttfamily\footnotesize,
% morestring=[b]",
% moredelim=[s][\bfseries\color{Maroon}]{<}{\ },
% moredelim=[s][\bfseries\color{Maroon}]{</}{>},
% moredelim=[l][\bfseries\color{Maroon}]{/>},
% moredelim=[l][\bfseries\color{Maroon}]{>},
% morecomment=[s]{<?}{?>},
% morecomment=[s]{<!--}{-->},
% commentstyle=\color{gray},
% stringstyle=\color{blue},
% identifierstyle=\color{red}
% % morekeywords={type,id,value,impl}% list your attributes here
%}
%%
%%---------------------------------------------------------
%
%
%\usepackage[pdftex]{graphicx}
%% declare the path(s) where your graphic files are
%\graphicspath{{./figures/}}
%% and their extensions so you won't have to specify these with
%% every instance of \includegraphics
%\DeclareGraphicsExtensions{.pdf}
%
%
%% *** MATH PACKAGES ***
%%
%% A popular package from the American Mathematical Society that provides
%% many useful and powerful commands for dealing with mathematics.
%\usepackage[cmex10]{amsmath}
%\usepackage{amssymb}
%\usepackage{mathtools}
%
%% *** SUBFIGURE PACKAGES ***
%\usepackage[caption=false,font=scriptsize]{subfig}
%
%
%
%% *** SPECIALIZED LIST PACKAGES ***
%%
%\usepackage{algorithmicx}
%\usepackage{algpseudocode}
%\usepackage[ruled]{algorithm}
%\definecolor{light-gray}{gray}{0.75}
%\algrenewcommand{\algorithmiccomment}[1]{\hskip3em{{\footnotesize \textcolor{light-gray}{$\blacktriangleright$}}} #1}
%
%% This package provides an algorithmic environment fo describing algorithms.
%% You can use the algorithmic environment in-text or within a figure
%% environment to provide for a floating algorithm.
%
%
%\usepackage{multirow}
%
%% *** ALIGNMENT PACKAGES ***
%%
%%\usepackage{array}
%% Frank Mittelbach's and David Carlisle's array.sty patches and improves
%% the standard LaTeX2e array and tabular environments to provide better
%% appearance and additional user controls. As the default LaTeX2e table
%% generation code is lacking to the point of almost being broken with
%% respect to the quality of the end results, all users are strongly
%% advised to use an enhanced (at the very least that provided by array.sty)
%% set of table tools. array.sty is already installed on most systems. The
%% latest version and documentation can be obtained at:
%% http://www.ctan.org/tex-archive/macros/latex/required/tools/
%
%
%
%
%% *** FLOAT PACKAGES ***
%%
%%\usepackage{fixltx2e}
%% fixltx2e, the successor to the earlier fix2col.sty, was written by
%% Frank Mittelbach and David Carlisle. This package corrects a few problems
%% in the LaTeX2e kernel, the most notable of which is that in current
%% LaTeX2e releases, the ordering of single and double column floats is not
%% guaranteed to be preserved. Thus, an unpatched LaTeX2e can allow a
%% single column figure to be placed prior to an earlier double column
%% figure.
%
%
%% *** PDF, URL AND HYPERLINK PACKAGES ***
%%
%%\usepackage{hyperref} % does not work with LLNCS style
%
%
%
%
%\usepackage{xspace}
%\usepackage[nocompress]{cite}
%
%% IEEETrans class fix for enumitem. provide for legacy IED commands/lengths when possible
%% http://comments.gmane.org/gmane.editors.lyx.general/68611
%\let\labelindent\relax
%\usepackage{enumitem}
%
%
%% correct bad hyphenation here
%\hyphenation{super-computer}
%
%% define repetitive complex fragments here
%\newcommand{\floe}{$\mathcal{F}{\ell}{o}{\varepsilon}$\xspace}
%\newcommand{\prc}{\mathcal{P}}
%\newcommand{\chn}{\mathcal{C}}
%
%\begin{document}
% % %
% % \frontmatter % for the preliminaries
% % %
% % \pagestyle{headings} % switches on printing of running heads
% % \addtocmark{Hamiltonian Mechanics} % additional mark in the TOC
% % %
% \mainmatter % start of the contributions
% %
% \title{Benchmarking Distributed Stream Processing Platforms for IoT Applications}
% %
% \titlerunning{IoT Benchmark for Stream Processing} % abbreviated title (for running head)
% % also used for the TOC unless
% % \toctitle is used
% %
% \author{Anshu Shukla and Yogesh Simmhan} %Tarun Sharma$^\ddagger$ $^\dagger$
% %
% \authorrunning{Shukla, et al.} % abbreviated author list (for running head)
% %
% %%%% list of authors for the TOC (use if author list has to be modified)
% % \tocauthor{Ivar Ekeland, Roger Temam, Jeffrey Dean, David Grove,
% % Craig Chambers, Kim B. Bruce, and Elisa Bertino}
% %
%
% \institute{Indian Institute of Science, Bangalore India\\
% %$^\ddagger$ NVIDIA Inc., Bangalore, India\\
% \email{shukla@grads.cds.iisc.ac.in, simmhan@cds.iisc.ac.in}\\
% % \and
% % Universit\'{e} de Paris-Sud,
% % Laboratoire d'Analyse Num\'{e}rique, B\^{a}timent 425,\\
% % F-91405 Orsay Cedex, France
% }
%
% \maketitle % typeset the title of the contribution
%
% \begin{abstract}
% Internet of Things (IoT) is a technology paradigm where millions of sensors monitor, and help inform or manage, physical, environmental and human systems in real-time. The inherent closed-loop responsiveness and decision making of IoT applications makes them ideal candidates for using low latency and scalable stream processing platforms. Distributed Stream Processing Systems (DSPS) are becoming essential components of any IoT stack, but the efficacy and performance of contemporary DSPS have not been rigorously studied for IoT data streams and applications. Here, we develop a benchmark suite and performance metrics to evaluate DSPS for streaming IoT applications. The benchmark includes $13$ common IoT tasks classified across functional categories and forming micro-benchmarks, and two IoT applications for statistical summarization and predictive analytics that leverage various dataflow patterns of DSPS. These are coupled with stream workloads from real IoT observations on smart cities. We validate the benchmark for the popular Apache Storm DSPS, and present the results.
% \keywords{stream processing, benchmark, workload, internet of things, smart cities, fast data, big data, velocity, distributed systems}
% \end{abstract}
\section{Introduction}% \note{2pgs}}
%\footnote{\note{Prepared for Special Issue on Resource Management for Big Data Platforms \url{http://www.journals.elsevier.com/future-generation-computer-systems/call-for-papers/special-issue-on-resource-management-for-big-data-platforms}}}
\label{sec:intro}
Internet of Things (IoT) is a technology paradigm wherein ubiquitous sensors numbering in the billions will able to \emph{monitor} physical infrastructure and environment, human beings and virtual entities in real-time, \emph{process} both real-time and historic observations, and \emph{take actions} that improve the efficiency and reliability of systems, or the comfort and lifestyle of society. The technology building blocks for IoT have been ramping up over for a decade, with research into pervasive and ubiquitous computing~\cite{zaslavsky2013internet}, and sensor networks~\cite{chandrasekaran:cidr:2003} forming precursors. Recent growth in the capabilities of high-speed mobile (e.g., 3G/4G) and \emph{ad hoc} (e.g., Bluetooth) networks~\cite{ericsson-report}, smart phones and devices, affordable sensing and crowd-sourced data collection~\cite{data:city}, Cloud data-centers, and Big Data analytics platforms have all converged to the current inflection point for IoT.
%IoT applications are pervasive, distributed, and often require near-real-time data processing and response.
Existing IoT deployments in vertical domains such as \emph{Smart Power Grids}~\cite{simmhan:cise:2013} and \emph{health and fitness monitoring}~\cite{fitness-iot} already have millions of sensing and actuation points that constantly stream observations and trigger responses. % , and Currently, the IoT applications are often manifest in vertical domains, such as demand-response optimization and outage management in \emph{smart grids}, or fitness and sleep tracking and recommendations by \emph{smart watches and health bands}~\cite{fitness-iot}.
% , such as cyber-physical systems like smart power grids and transportation at city-scales; smart appliances, TVs and thermostats at households; and health and fitness devices carried on a person. The
The IoT stack for such domains is tightly integrated to serve specific needs, but typically operates on a closed-loop \emph{Observe Orient Decide Act (OODA)} cycle~\cite{perera:ieee:2014}, where sensors communicate time-series observations of the (physical or human) system to the Cloud for analysis, and the resulting analytics drives recommendations that are enacted on the system to improve it, which is again observed and so on. In fact, this \emph{closed-loop responsiveness} is one of the essential and distinguishing design characteristics of IoT applications, compared to other Big Data domains.
%Present IoT verticals deploy $10^3$ to $10^6$ sensors, e.g. Fitbit fitness watches have $\sim$10~M users and the Los Angeles Smart Grid project is deploying $\sim$50,000 smart meters. But in current models, the real-time data analysis is localized to data streams from individual sensors or clusters of sensors, and often sub-sampled at the sensors itself to report readings for analysis in batches periodically, or sending coarse aggregates of the observations. E.g. smart meters while collecting kWh data at 15~min intervals may report it back to the utility only a few times a day, while Fitbit syncs with the Cloud every few minutes even as data is recorded every few minutes.
This low-latency cycle makes it necessary to process data streaming from sensors at fine spatial and temporal scales, in \emph{real-time}, to derive actionable intelligence. In particular, this streaming analytics has be to done at massive scales (millions of sensors, thousands of events per second) from across distributed sensors, requiring large computational resources. \emph{Cloud computing} offers a natural platform for scalable processing of the observations at globally distributed data centers, and sending a feedback response to the IoT system at the edge.
Recent \emph{Big Data platforms} like Apache Storm~\cite{toshniwal:sigmod:2014}, Spark Streaming~\cite{zaharia:usenix:2012} and Flink~\cite{carbone2015flink} provide an intuitive dataflow programming model for composing such streaming applications, with a scalable, low-latency execution engine designed for commodity clusters and Clouds.
These \emph{Distributed Stream Processing Systems (DSPS)} are becoming essential components of any IoT stack to support online analytics and decision-making for IoT applications. DSPS provide the ability to compose a dataflow graph of user-defined tasks that can process a continuous stream of opaque messages on distributed resources. This flexibility allows DSPS to incorporate a wide variety of business logic for real-time processing and online analytics necessary for a diverse and emerging domain like IoT. In fact, reference IoT solutions from Cloud providers like Amazon AWS\footnote{https://aws.amazon.com/iot/how-it-works/} and Microsoft Azure\footnote{https://www.microsoft.com/en-in/server-cloud/internet-of-things/overview.aspx} include their proprietary stream and event processing engines as part of the IoT software architecture.% -- Kinesis and Azure Stream Analytics, respectively.
Shared-memory stream processing systems~\cite{chandrasekaran:cidr:2003, aurora} have been investigated over a decade back for wireless sensor networks, with benchmarks such as \emph{Linear Road}~\cite{arasu:vldb:2004} being proposed. But there has not been a detailed review of, or benchmarks for, \emph{distributed} stream processing for IoT. IoT encompasses multiple domains, and applications go well beyond traditional social network and web traffic workloads for which DSPS were designed for~\cite{toshniwal:sigmod:2014}. They include a swathe of generalizable tasks for data pre-processing, statistical summarization and predictive analytics, as well as analytics for specific IoT application areas like Smart Transportation or health. As such, the efficacy and performance of contemporary DSPS have not been rigorously studied for \emph{IoT applications and data streams}. One reason is the absence of a well-defined IoT benchmark that realistically captures the domain features, exercises the unique compositional capabilities of DSPS, and validates them on real data streams. We address this gap in this paper.
This paper extends our prior published work, significantly increasing both the breadth and depth of the benchmark suite~\cite{shukla:tpctc:2016}. We add 14 new tasks to the earlier 13 tasks, including in new categories; % , including three for parse, one for filter, one for statistical, three for predictive, five for IO, one for visualization
two new streaming dataflow applications, besides updating the earlier two as well; and two new data workloads from the smart grid and personal fitness domains. We also include support for spatial scaling to increase the number of sensor streams, in addition to the temporal scaling used earlier to increase the stream rates. These make our benchmark comprehensive.
%We develop a benchmark suite for DSPS to evaluate their effectiveness for streaming IoT applications. The proposed workload is based on common building-block tasks observed in various IoT domains for real-time decision making, and the input streams are sourced from real IoT observations from smart cities.
%\ysnote{What will the benchmark evaluate? What will it inform the user? How can the results be interpreted?}
%\ysnote{2-3 lines on why existing DSPS benchmarks are inadequate for IoT apps}
Specifically, we make the following contributions in this article:%\ysnote{Revisit contributions...}
\begin{enumerate}
\item We classify different \emph{characteristics} of streaming applications, their composition semantics, and their data sources, in \S~\ref{sec:features}.
\item Then, in \S~\ref{sec:features:iot}, we propose \emph{categories of tasks} that are essential for IoT applications and the \emph{key features of input data streams} they operate upon.
\item We identify \emph{performance metrics} of DSPS that are necessary to meet the latency and scalability needs of streaming IoT applications, in \S~\ref{sec:metrics}.
\item We propose the \textbf{RIoTBench} real-time IoT benchmark for DSPS based on representative \emph{micro-benchmark tasks}, drawn from the above categories, in \S~\ref{sec:benchmark}. We design four reference \emph{IoT applications} that span Data pre-processing, Statistical analytics and Predictive Analytics, and are composed from these tasks. We also identify \emph{four real-world streams} with different distributions as workloads on which to evaluate them.
\item Lastly, we validate the proposed benchmark suite for the popular \emph{Apache Storm} DSPS, and present empirical results for the same in \S~\ref{sec:results}.
\end{enumerate}
% In this article, we make two key contributions. First, we develop a taxonomy of application, platform and infrastructure characteristics supported by, and desired from, contemporary DSPS in the context of to composing and executing IoT applications. Several research and production DSPS are discussed based on attributes that fall along these dimensions. Second, we identify key performance metrics to evaluate these DSPs for IoT applications. We further propose benchmarks to evaluate DSPS for IoT applications using real, public datasets from environmental sensing, smart power grids and smart transportation domains, along with micro and full applications that process these datasets. We validate these benchmarks by using them to evaluate two popular DSPS, \emph{Apache Storm} that was originally contributed to the public domain by Twitter, and \emph{Apache Spark} that comes from UC Berkeley's AMP Labs.
Our contributions benefit two classes of audience. One, for \emph{developers and users in IoT domains}, RIoTBench offers a set of realistic IoT tasks and applications that they can customize and configure to help evaluate candidate DSPS platforms for their performance and scalability needs. Two, for \emph{researchers on Big Data}, it provides a reference micro and application benchmark, along with datasets, that can be used as a baseline to uniformly compare the impact of their research advances in resource management, scalability and resiliency for DSPS on the emerging IoT domain.
\section{Background and Related Work}% \note{1.5pgs}}
\label{sec:related}
%\subsection{Background}
%\subsection{Distributed Stream Processing Systems (DSPS)}
Stream processing systems allow users to compose applications as a dataflow graph, with task vertices having some user-defined logic and streaming edges passing messages between the tasks. The systems then run the applications continuously over incoming data streams. %, with source and sink tasks nominally serving input streams to the application and draining the output streams from it, respectively.
%There are additional semantics, besides dataflow dependencies, associated with the application composition and determine messages routing between tasks. These applications are deployed and executed within a runtime environment, which follows the semantics of the composition and also manages the resources to meet quality of service and performance goals.
%
%\modc{Distributed stream processing platforms ( Recent )}\\
% These stream processing frameworks were focussed less on distributed aspects and more on the \emph{data semantics} and \emph{query processing} part of the problem.
Early Data Stream Management Systems (DSMS) extended Database Management Systems (DBMS) to support by sensor network applications, that have similarities to IoT~\cite{carney:vldb:2002,chen:sigmod:2000,babu:record:2001}. They supported continuous query languages with operators such as join and aggregation similar to SQL, but with a temporal dimension using time and tuple window operations. These have been extended to distributed implementations~\cite{balazinska:tods:2008,biem:sigmod:2010} and, more recently, complex event processing (CEP) engines for detecting sequences and patterns~\cite{cugola:csur:2012}. %that examines reliable distributed execution in Borealis DSMS that balances availability with consistency. There is still active work on distributed DSMS, such as IBM InfoSphere Streams~\cite{biem:sigmod:2010}, and distributed complex event processing.
Contemporary Distributed Stream Processing Systems (DSPS) like Apache Storm, Spark Streaming, Flink and Yahoo S4~\cite{toshniwal:sigmod:2014,zaharia:usenix:2012,carbone2015flink,neumeyer:icdmw:2010} were designed using Big Data fundamentals -- running on commodity clusters and Clouds, offering weak scaling, ensuring robustness, and supporting fast data processing over thousands of events per second. Unlike DSMS, DSPS do not support native query operators and instead allow users to plug in their own logic composed as dataflows that are executed on a cluster. Event processing and querying can be higher-level abstractions on top of these~\footnote{Apache Trident, http://storm.apache.org/releases/1.0.1/Trident-tutorial.html}. While developed for web and social network applications, such fast data platforms have found use in financial markets, astronomy, and particle physics. IoT is one of the more recent domains to consider them. %domain like transportation and smart power grids of more recent interest.
% In recent times, dealing with massive amounts of data commonly generated by the sensors, and smartphones, posed a serious challenge to process the queries in real-time and responsive fashion. The above discussed frameworks are not mature enough to work at such massive scales.
%Hence, application of distributed and cloud computing has become prime focus in order to catch up with the ever-increasing data.
%
%Contemporary stream processing systems are also cast in the context of Cloud infrastructure and other Big Data platforms. In DSPS, the execution of the application takes place on multiple machines distributed within a commodity cluster or a Cloud data center to scale the processing. The DSPS runtime is designed to meet various performance criteria for real-time applications, such as high throughput and low latency, apart from fault-tolerance, scalability and load balancing in a distributed setup. Contemporary systems include Apache Storm~\cite{toshniwal:sigmod:2014}, Apache Spark Streaming~\cite{zaharia:usenix:2012}, and Yahoo S4~\cite{neumeyer:icdmw:2010}.
There are design and architectural differences even within DSPS, which we highlight as part of our characterization. The types of programming semantics supported can vary, and determines the flexibility in composition. Spark Streaming uses micro-batch processing in contrast to per-tuple processing in Storm, with consequences trade-offs between latency and throughput. As a result, it is important to qualitatively and quantitatively evaluate these frameworks for specific application domains, and the distributed platform they target. Understanding the common set of feature dimensions and performance metrics, in addition to the actual IoT benchmark definitions, is necessary for fair comparison across the DSPS. We discuss these later for DSPS operating on Clouds to support IoT applications.
%
%\subsubsection{Apache Storm}
%\ysnote{todo}
%Apache Storm is an open-source distributed real-time computation engine, by Twitter Inc. It works as a black-box platform, and is opaque to the user-code logic for the topologies. Source tasks are called as \emph{spouts} and rest of the tasks are called as \emph{bolts}. It provides \emph{atleast-once} semantics by tracking each tuple (i.e. messages) with the help of special \emph{Acker tasks}, and if the message is not fully processed as per a given message timeout, then it is replayed at the \emph{spout}. It supports \emph{stateless} tasks only. \emph{Scalability} can thus be achieved by configuring the \emph{parallelism hint} of each \emph{spout} and \emph{bolt}. It does not guarantee ordering of the messages. Since, the framework by default does not support \emph{stateful} tasks, the end user has to incorporate the state logic in their code.
%
%It supports \emph{dynamic rebalancing} of number of threads with storm rebalance feature without the need to restart the topolpogy. But this scaling is manual.
%
%It is also fault-tolerant, as it uses Zookeeper cluster to restart, if either the master process (i.e. Nimbus) dies or any worker node or process (i.e. Supervisor) dies.
%
%The per-tuple processing approach used helps to optimize \emph{latency} in contrast to \emph{throughput}.
%
%\emph{Apache Trident} provides higher-level abstraction to Storm. It supports various white-box operators like join, aggregation, filter, etc. It is \emph{stateful} and has \emph{exactly-once} semantics. It keeps track of small batches of tuples, which ensures \emph{ordering} among the different batches.
%
%\subsubsection{Apache Spark Streaming}
%\ysnote{todo}
%
%This framework is designed keeping goal as to maximize the throughput while keeping the end-to-end latency below a given target. It works at granularity level of small batches of messages called as RDD (i.e. Resilient Distributed Datasets). These are immutable datasets. The operators are deterministic in nature. For recovering from failovers they rely on this \emph{deterministic} nature of the operations, apart from lineage information, parallel recovery and replication.
%
%\emph{Latency} It is defined as the sum of (Time to wait for a new input batch to start) and (Time from when records are sent to the system to when results incorporating them appear).
%
%To ensure same message (tuples) ordering for replicas, synch protocol like DPC (Delay, Process and Correct)~\cite{balazinska:tods:2008} is used.
%%\asnote{cite Fault-tolerance in the Borealis distributed stream processing system}
%
%\modc{Spark can not cope with applications with regularly fine-grained updates to shared data.
%Spark is not meant for applications like high-frequency trading which have latency needs below few hundred milliseconds.}
%
%%\asnote{Fault and straggler recovery - impact on recovery time by Varying the Checkpoint Interval - impact on recovery time by Varying the no. of nodes}
%
%
%%\modc{Applications for DSP} \\
%%Distributed Stream Processing encompasses various domains, In recent times as the IoT based applications are gaining attention, it becomes even more challenging and important to look into how to best utilize the distributed systems concepts to tame this \emph{big data}.
%%First of the benchmarks, in this domain, is the Linear Road Benchmark. It generates the toll and vehicle position data for a hypothetical highway, and aims at benchmarking the latency and throughput requirements.
%
%% Research: elasticity, state
%% Fernandez:sigmod:2013, Integrating Scale Out and Fault Tolerance in Stream Processing using Operator State Management, Raul Castro Fernandez, SIGMOD, 2013
%% ding:arXiv:2015, Efficient Operator State Migration for Cloud-Based Data Stream Management Systems, arXiv:1501.03619
%
%% IoT apps
%% DEBS challenge for smart grids, soccer
%% Grand Challenge: Scalable Stateful Stream Processing for Smart Grids. Raul Castro Fernandez, Matthias Weidlich, Peter Pietzuch and Avigdor Gal. 8th ACM International Conference on Distributed Event Based Systems (DEBS), 05/2014, Mumbai, India, (2014)
%
%%\modc{Surveys on stream processing system} \\
%\subsection{Information Flow Processing (IPF) systems \Note{Trim}} \cite{cugola:csur:2012} offer a detailed review of various Information Flow Processing (IPF) systems, that include active databases, data stream management systems (DSMS), and complex event processing (CEP) systems. They offer a taxonomy of the query and time models of these IFPs, and classify both historic and contemporary systems using this scheme. They do not emphasize on the performance or distributed characteristics for these IFPs, nor do they investigate them from the perspective of specific domains such as IoT.
%
%DSMS and CEP both differ from DSPS that we consider, both in the data model of events and in how their real-time processing logic is defined and executed. DSMS/CEP have a schema for events, possibly defined as a tuple of attributes, each with a name, type and value. Users define declarative queries over the events based on this schema which is executed by the DSMS/CEP engine, possibly in a distributed manner, on the real-time event streams. DSPS, on the other hand, treat messages as opaque objects, and only exposing message IDs and optionally keys that may be used for grouping and routing. Semantics are not associated with the message content. Users also explicitly provide the logic that has to be performed and their data dependencies, with some limited declarative semantics. Our focus here is evaluating DSPS.
%
%% balazinska:tods:2008, Fault-Tolerance in the Borealis Distributed Stream Processing System, MAGDALENA BALAZINSKA, TODS, March, 2008
%% babu:record:2001, Continuous Queries over Data Streams, Babu, Shivnath and Widom, Jennifer (2001) Continuous Queries over Data Streams. SIGMOD Record, 30 (3) [STREAM]
%% chen:sigmod:2000, NiagaraCQ: A Scalable Continuous Query System for Internet Databases. Jianjun Chen, David J. DeWitt, Feng Tian, Yuan Wang. SIGMOD 2000 , p379-390
%% chandrasekaran:cidr:2003, TelegraphCQ: Continuous Dataflow Processing for an Uncertain World, Sirish Chandrasekaran, Owen Cooper, Amol Deshpande, Michael J. Franklin, Joseph M. Hellerstein, Wei Hong, Sailesh Krishnamurthy, Samuel Madden, Vijayshankar Raman, Frederick Reiss, Mehul A. Shah, CIDR, 2013
%% carney:vldb:2002, Monitoring Streams: A New Class of Data Management Applications., D. Carney, U. Cetintemel, M. Cherniack, C. Convey, S. Lee, G. Seidman, M. Stonebraker, N. Tatbul, and S. Zdonik. In proceedings of the 28th International Conference on Very Large Data Bases (VLDB'02) [AURORA]
%
%It is worthwhile to note that work on DSMS in the early 2000's were motivated by sensor network applications, and hence offer similarities with the needs of IoT applications. Aurora~\cite{carney:vldb:2002}, NiagaraCQ~\cite{chen:sigmod:2000}, STREAM~\cite{babu:record:2001} and TelegraphCQ~\cite{chandrasekaran:cidr:2003} were early DSMSs that evolved from prior work in DBMS. They all had continuous query languages, similar to SQL but with a temporal flavor, and queries were decomposed into directed acyclic graphs (DAGs) with query operators such as join, aggregators, and windowed-join as tasks and streams of tuples forming edges. There has also been work on distributed execution of DSMS, such as ~\cite{balazinska:tods:2008} that examines reliable distributed execution in Borealis DSMS that balances availability with consistency. There is still active work on distributed DSMS, such as IBM InfoSphere Streams~\cite{biem:sigmod:2010}, and distributed complex event processing. One outcome of that body of work, relevant to this article was the Linear Road Benchmark (LRB)~\cite{arasu:vldb:2004} that was proposed as an application benchmark for DSMS. In the scenario, DSMS had to evaluate toll and traffic queries over event streams from a virtual toll collection and traffic monitoring system. This parallels with current smart transportation scenarios, and one of our application benchmarks is inspired by LRB.
%
%
%% goodhope:deb:2012, Building LinkedIn’s Real-time Activity Data Pipeline, Ken Goodhope, Data Engineering Bulleting, April, 2012
%Publish-subscribe systems are used for many-to-many event delivery, between many publishers and multiple subscribers, often with one or more brokers to mediate the event routing using topics and simple queries. For e.g., Apache Kafka is a distributed pub-sub system developed by LinkedIn to handle 100,000's of events per second~\cite{goodhope:deb:2012}. It performs real-time monitoring of user activity on the LinkedIn website, and subsequently analyze to deliver user content in a reliable manner by architecting it as a distributed commit log. While Kafka and other distributed pub-sub systems perform high-throughput stream processing in a manner of speaking, their role is restricted to event routing, query and delivery rather than processing and transformation of events. Also, they do not offer any compositional mechanisms for paths that the events take, instead offering a declarative syntax. Often, such pub-sub systems act as sources of streams that feed into DSPS, just as DSPS can feed into or incorporate DSMS. % batching of messages is done on application side by Kafka, to increathruput by 3x though latency increased.
%\modc{Surveys on stream distributed processing system}\\
%\subsection{Other Stream and Big Data Benchmarks}
\subsection{DSPS Benchmarks}
Work on DSMS spawned the \emph{Linear Road Benchmark (LRB)}~\cite{arasu:vldb:2004} that was proposed as an application benchmark. In the scenario, the DSMS had to evaluate toll and traffic queries over event streams from a virtual toll collection and traffic monitoring system. This has parallels with current smart transportation scenarios.
%
However, there have been few studies or community efforts on benchmarking DSPS, other than individual evaluation of research prototypes against popular DSPS like Storm or Spark.
%
These efforts define their own measures of success -- typically limited to throughput and latency -- and use generic workloads such as the Enron email dataset with empty operations (NoOps) as micro-benchmark to compare InfoSphere Streams ~\cite{nabi:streams:2014} and Storm. %SparkBench~\cite{agrawal:sparkbench:2015} uses two streaming applications, Twitter popular tag retrieving data from the twitter website to calculate most popular tag every minute and PageView over synthetic user clicks to get various statistics using spark.
%smart city dataset in CityBench~\cite{ali:citybench:2015} and
%on the dimensions and metrics for comparing different DSPS. Other than latency and throughput performance metrics, most current literature define their perform individual evaluation of their proposed DSPS against, occasionally compared against popular ones like Apache Storm or Spark Streaming.
%This is in contrast to DSMS and CEP systems which have active community participation in grand challenges~\cite{debs}, and have published surveys to evaluate DSPS.
% ~\cite{anshu:wbdb:2014}
\emph{SparkBench}~\cite{agrawal:sparkbench:2015} is a framework-specific benchmark for Apache Spark, and includes four categories of applications from domains spanning Graph computation and SQL queries, with one on streaming applications supported by Spark Streaming. The benchmark metrics include CPU, memory, disk and network IO, with the goal of identifying tuning parameters to improve Spark's performance. % leaving latency which is important for DSPS.It uses mixed synthetic workload for different dimensions of Big Data .
%
%\Note{Introducing TPCx-HS: The First Industry Standard for Benchmarking Big Data Systems, TPCTC 2014}
%\Note{CEPBen: A Benchmark for Complex Event Processing Systems, TPCTC 2013}\\
\emph{CEPBen}~\cite{li:tpctc:2014} evaluates the performance of CEP systems based of the functional behavior of queries. It shows the degree of complexity of CEP operations like filter, transform and pattern detection. The evaluation metrics consider event processing latency, but ignore network overheads and CPU utilization. Further, CEP applications rely on a declarative query syntax to match event patterns rather than a dataflow composition based on user-logic provided by DSPS.%Also impact of input rate on cpu and memory is not being considered.
\emph{StreamBench}~\cite{lu:ucc:2014} is the closest work that partially addresses our goals. The authors propose 7 micro-benchmarks on 4 different synthetic workload suites generated from real-time web logs and network traffic to evaluate DSPS. Metrics including performance, durability and fault tolerance are proposed. The benchmark covers different dataflow composition patterns and common tasks like grep and wordcount, and compare Storm and Spark Streaming. %While useful as a generic streaming benchmark, it does not consider aspects unique to IoT applications and streams.
The paper, while addressing the gap that existed in generalizable benchmarks DSPS, still falls short on several counts. % Their paper is focused on quantitative evaluation of metrics like throughput, latency and penalty factors for failure. It does not delve into qualitative dimensions for discriminating between DSPS.
It focuses on micro-benchmarks% , which are commonly found as sub-parts of applications,
and does not consider larger applications with more tasks and complex structures. % as we do for the IoT domain.
Design patterns like % that affect the message transfer between tasks (
duplicates and round-robin, and selectivity ratios % (relative number of output messages generated with respect to input)
are not explicitly considered. %, unlike in our study. % As a result, it is not possible to discern the behavior of the DSPS for representative applications of interest. However, in terms of input-data scale
The benchmark does not cover a broad range of realistic input data rates either.
% Further, while different scaling factors of data rates were considered, there is no discussion of datasets which have variable rates over time as is common in real-world data. For e.g., online retailers have a diurnal workload pattern on their geography of interest, and transport networks may have a bi-modal distribution of traffic rates, centered on morning and evening commutes. We explicitly discuss the input message rate distribution in our evaluation. Lastly, while durability and fault tolerance are evaluated, the non-deterministic behavior of Cloud VMs due to multi-tenance is not considered.
We address these gaps. At the same time, we do not emphasize durability or fault-tolerance metrics in our study, through these metrics can be added.% though the latter is a qualitative dimension.
% StreamBench~\cite{lu:ucc:2014} is the closest work that partially addresses our goals. The authors propose performance, durability, fault tolerance and multi-recipient workloads that can be applied across different DSPS, and evaluate these over two seed datasets, AOL web queries (text) and Internet trace (numeric). The workloads contain $7$ micro-benchmarks that range from 1-3 tasks each, including transformations, filter and grouping operations, and include both stateful and stateless operations. They have analyzed throughput and latency metrics for these workloads, comparing Apache Storm and Apache Spark Streaming, and further introduced latency and throughput penalty factors (LPF, TPF) to measure the impact of node failures.
In contrast to these DSPS benchmarks, RIoTBench offers relevant micro- and application-level benchmarks for evaluating DSPS, specifically for \emph{IoT workloads} for which such platforms are increasingly being used. Our benchmark is designed to be \emph{platform-agnostic}, \emph{simple} to implement and execute within diverse DSPS, and \emph{representative} of both the application logic and the data stream workloads observed in IoT domains. This allows for the performance of DSPS to be independently and reproducibly verified for IoT applications. % and \emph{verifiable}. We focus on IoT domains for the micro and application-level benchmarks.
% Qualitative measures are absent.
% Application workloads, design patterns. Selectivity.
% Impact of change in data rates?
% IoT specific datasets or metrics lacking.
% Durability and fault tolerance present, but metrics on impact of Cloud not verified.
%There have also been individual experiments, as opposed to comprehensive workloads that can be used for comparison, that have been used in literature, that can inform the assembly of a workload. IBM InfoSphere Streams~\cite{nabi:tr:2014}, which is a DSMS, has designed a real-world stream processing application benchmark that performs email classification for online spam detection~\cite{martin:ceas:2005}, and compares their performance against Apache Storm. Apache Spark Streaming has been evaluated over micro-benchmarks like Grep, Word-Count and Top-K-Count~\cite{zaharia:usenix:2012}. In addition, Spark Streaming has been evaluated on application benchmarks like video distribution monitoring and a machine learning algorithm for traffic conditions estimating over GPS data~\cite{anshu:link:2014}~\cite{hunter:socc:2011}.
% \note{Amongst other points regarding the paper are viz., not taken some diverse scenarios in to consideration like scalability,
% %Not including the two points as we are not dealing with these in this paper. \asnote{heterogeneous data input types} \asnote{windowing operations},
% combined real-time and historical data queries. Since our main focus here is on performance and scalability of the frameworks so we have not considered the fault tolerance as a benchmarking parameter.}
\subsection{Big Data and IoT Benchmarks}
There has been a slew of Big Data benchmarks that have been developed recently in the context of processing high volume (i.e., MapReduce-style) and enterprise/web data that complement our work. %, say, using Hadoop MapReduce, can also be used as a basis for suggesting DSPS benchmarks. %Hibench~\cite{huang:hibench:2010}, BigDataBench~\cite{gao:bigdatabench:2013} and BigBench~\cite{ghazal:acm:2013} are some examples.
\emph{Hibench}~\cite{huang:hibench:2010} is a workload suite for evaluating Hadoop with popular micro-benchmarks like Sort, WordCount and TeraSort, MapReduce applications like Nutch Indexing and PageRank, and machine learning algorithms like K-means Clustering. \emph{BigDataBench}~\cite{gao:bigdatabench:2013} analyzes workloads from social network and search engines, and analytics algorithms like Support Vector Machine (SVM) over structured, semi-structured and unstructured web data. Both these benchmarks are general purpose workloads that do not target any specific domain, but MapReduce platforms at large. %The wide prevalence of such benchmarks attest to the maturity of Hadoop and MapReduce programming model, and that level of study is lacking for DSPS, which we aim to address here.
\emph{BigBench}~\cite{ghazal:acm:2013} uses a synthetic data generator to simulate enterprise data found in online retail businesses. It combines structured data generation from the TPC-DS benchmark~\cite{nambiar:vldb:2006}, semi-structured data on user clicks, and unstructured data from online product reviews. Queries cover data \emph{velocity} by processing periodic refreshes that feed into the data store, \emph{variety} by including free-text user reviews, and \emph{volume} by querying over a large web log of clicks. We take a similar approach for benchmarking fast data platforms, targeting the IoT domain specifically and using real public data streams. %We target the IoT domain at large, using multiple applications from smart power grids, smart transportation and environmental monitoring within that space, but restrict our evaluation to DSPS rather than the entire data management ecosystem which is as yet emerging for IoT.In recent times as the IoT based applications are gaining attention, it becomes even more challenging and important to look into how to best utilize the distributed systems concepts to tame this \emph{big data}.%First of the benchmarks, in this domain, is the Linear Road Benchmark~\cite{arasu:vldb:2004}. It generates the toll and vehicle position data for a hypothetical highway and was meant to compare Data Stream Management Systems (DSMS) with Database Management Systems (DBMS).It uses response time,accuracy of the query and the peak sustained rate as the evaluation metrics ,ingnoring resource utilization.
%pavlo:sigmod:2009, A Comparison of Approaches to Large-Scale Data Analysis, Andrew Pavlo, SIGMOD 2009
%\cite{pavlo:sigmod:2009} offers an evaluation of processing and analysis platform for large tuple-based batch datasets, comparing Hadoop MapReduce and two parallel relational databases. They compare the architectural and programming models between these systems, as well as proposing benchmarks based on selection, aggregation and join tasks whose transaction time scaling are evaluated on 1 to 100 nodes. Our effort is on a similar vein but for distributed stream processing rather than static data, and with the emphasis on metrics, datasets and tasks relevant to IoT applications.
%\Note{SparkBench: A Spark Performance Testing Suite, TPCTS 2015}\\
%\Note{Adding a Temporal Dimension to the TPC-H Benchmark, TPCTC 2012}
%\Note{BDMS Performance Evaluation: Practices, Pitfalls, and Possibilities, TPCTC 2012}
%\subsection{IoT Workloads}
\emph{Chronos}~\cite{gu:chronos:2015} is a recent work to generate and simulate streams for benchmarking. Their aim is to generate realistic input data streams with a distribution similar to given sample events. They use elastic infrastructure to generate events at high rates, and validate their work for telecom, advertising and stock market data. Their work is complementary to ours, as we propose dataflow patterns and applications, as well as representative datasets as part of benchmarks which are run at their native and scaled rates. Chronos can be used to stress these benchmarks further with larger inputs and faster rates.
%\Note{IoTAbench: an Internet of Things Analytics Benchmark}
There has been some recent work on benchmarking IoT applications. In particular, the generating large volumes of synthetic sensor data with realistic values is challenging, yet required for benchmarking. \emph{IoTAbench}~\cite{arlitt:icpe:2015} provides a scalable synthetic generator of time-series datasets. It uses a Markov chain model for scaling the time series with a limited number of inputs such that important statistical properties of the stream is retained in the generated data. They have demonstrated this for smart meter data. The benchmark also includes six SQL queries to evaluate the performance of different query platforms on the generated dataset. Their emphasis is on the data characteristics and content, which supplements our focus on evaluating the runtime aspects of the DSPS platform. %The proposed SQL queries are data dependent and they can not be categorised for common IoT use cases.Metrics like supported peak rate,cpu and memory have not been taken into consideration.
%\Note{CityBench: A Configurable Benchmark to Evaluate RSP Engines Using Smart City Datasets}\\
\emph{CityBench}~\cite{ali:citybench:2015} is a benchmark to evaluate RDF stream processing systems. They include different generation patterns for smart city data, such as traffic vehicles, parking, weather, pollution, cultural and library events, with changing event rates and playback speeds. They propose fixed set of semantic queries over this dataset, with concurrent execution of queries and sensor streams. Here, the target platform is different (RDF database), but in a spirit as our work. %But data generation and application queries are restricted to smart city datasets only.
%\ysnote{http://www.eembc.org/iot-connect/about.php, IoT-Connect, Industry-Standard Benchmarks for Embedded Systems }
Benchmarks for IoT hardware is also becoming important. \emph{IoT-Connect}~\cite{iotconnect} is an Industry-Standard Benchmark for Embedded Systems to analyze the behavior of micro-controllers with various connectivity interfaces % e.g. testing the combination of microcontroller and RF components
like Bluetooth, Thread, LoRa, and WiFi. It also provides methods to determine the energy consumption for IoT devices. %In~\cite{chang:acm:2016}, the authors identify the limitations of Mobile Cloud Business process management system for IoT. They also mention the business use-cases for IoT in Logistics, Enterprise Resource Planning (ERP), Smart Building and Healthcare. Some of these domains resembles to the dataset domains used in our application benchmarks.
%\ysnote{http://blogs.cisco.com/datacenter/industry-standards-for-benchmarking-iot, Benchmarking Internet of Things (IoT)}
%\ysnote{Chii Chang, Satish Narayana Srirama, and Rajkumar Buyya. 2016. Mobile Cloud Business Process Management System for the Internet of Things: A Survey. ACM Comput. Surv. 49, 4, Article 70 (December 2016), 42 pages. DOI: https://doi.org/10.1145/3012000 }
%\Note{Reinventing the TPC: From Traditional to Big Data to Internet of Things}, \url{http://link.springer.com/chapter/10.1007/978-3-319-31409-9\_1}
%\Note{Benchmarking Internet of Things Deployments in Smart Cities}, \url{https://www.computer.org/csdl/proceedings/waina/2013/4952/00/4952b319-abs.html}
%\asnote{link for yahoo benchmark and topology -
%https://yahooeng.tumblr.com/post/135321837876/benchmarking-streaming-computation-engines-at}
%\tsnote{Check the below paragraph which states what all things we would be omitting for our paper. Also one of the citation has Springer so PDF yet to be put in related-work folder. Else omit this citation.}
% \modc{
% Various other types of \emph{specialized} stream processing benchmarks too exist, like PLR~\cite{karachi:iids:2012} that is used for probabilistic data stream management systems, geo-stream management systems~\cite{shen:gis:2011}, and SLUBM~\cite{nguyen:iswc:2013} for stream reasoning applications. However, for this paper we would not be considering these special cases.
% }
%\newpage
\section{Characteristics of DSPS Applications and Streams}% \note{1.5pgs}}%Streaming IoT Applications }
\label{sec:features}
% We classify different characteristics of streaming applications and their data sources, in § 3. We propose categories of tasks that are essential for IoT applications and the key features that are present in their input data streams.
In this section, we review the common application composition capabilities of DSPS, and the dimensions of the streaming applications that affect their performance on DSPS. These semantics help define and describe streaming IoT applications based on DSPS capabilities.
% and use those to A comprehensive view on the suitability of DSPS for IoT applications hosted on infrastructure Clouds requires us to consider various aspects of the application and its composition, features provided by the DSPS and its ability or shortcomings on the platform of choice, and quantitative metrics that help gage the suitability. These are covered in the next three sections.
%In this section on application characteristics, we first discuss compositional primitives that are commonly offered by DSPS, and the dimensions of the streaming applications themselves that are relevant to their performance on DSPS. Next we review existing popular and relevant streaming applications, and classify them along these dimensions. These further offer insights on what representative compositions to use for our benchmarking.
\subsection{Dataflow Composition Semantics}
\label{subsec:dataflow}
%http://storm.apache.org/documentation/Common-patterns.html
DSPS applications are composed as a \emph{dataflow graph}, where vertices are user provided \emph{tasks} and directed edges refer to \emph{streams of messages} that can pass between them. The graph need not be acyclic. Tasks in the dataflows can execute zero or more times, and a task execution usually depends on data-dependency semantics, i.e, when ``adequate'' inputs are available, the task executes. However, there are also more nuanced patterns that are supported by DSPS that we discuss.
\emph{Messages} (or events or tuples) from/to the stream are consumed/produced by the tasks. % DSMS, unlike DSPS, treat these messages as transparent tuples with well-defined attribute names, types and values, which we term a \emph{whitebox} data model. This allows them to offer built-in query operators as tasks.
DSPS typically treat the messages as opaque content, and only the user logic may interpret the message content. However, DSPS may assign identifiers to messages for fault-tolerance and delivery guarantees, and some message attributes may be explicitly exposed as part of the application composition for the DSPS to route messages to downstream tasks. %DSPS that support this are called \emph{graybox} data model.
\emph{Selectivity ratio}, also called \emph{gain}, is the average number of output messages emitted by a task on consuming a unit input message, expressed as $\sigma= \text{\emph{input rate}}:\text{\emph{output rate}}$. Based on this, one can assess whether a task amplifies or attenuates the incoming message rate. It is important to consider this while designing benchmarks as it can have a multiplicative impact on downstream tasks.
% we can categorize these DSPs as \emph{black box, gray box} and \emph{white box}. In \emph{black-box} DSPs, the user-code logic and the messages in the topology are opaque to the framework. This provides the users \emph{flexibilty} to compose their applications. While in \emph{white-box} DSPs, the application logic is composed of built-in operators like join, aggregators and so on, and the framework knows the schema of the messages. This is best for query processing applications, where one needs to perform common operations like project and filter. Gray-box DSPs, encompass both black and white box characteristics.
% Generally, \emph{white-box} application's variation in behavior and performance is more across different platforms than \emph{black-box} applications.
% Similar classification holds for the \emph{application type}, that is if the application is query processing type with SDMSs operators like project, filter, join and aggregators then it is called as \emph{white-box} type application, otherwise it is treated as \emph{black-box} type application. Some application types may have both the characteristics and are called \emph{gray-box} type applications.
%\tsnote{Distinguish between task/application as a white/black/gray box AND/OR the platform is a white/black/gray box}
%Continuous Dataflow applications are Directed-acyclic graphs with vertices as tasks and edges as channels along which messages are sent.
% \subsubsection{Dataflow Patterns}
% \label{sec:semantics}
\begin{figure*}[t]
\centering
\includegraphics[width=\textwidth]{figures/semantics.pdf}
\caption{Common task patterns and semantics in streaming applications.}
\label{fig:semantics}
\end{figure*}
There are message generation, consumption and routing semantics associated with tasks and their composition. Fig.~\ref{fig:semantics} captures the basic \emph{composition patterns} supported by modern DSPS. \texttt{Source} tasks have only outgoing edge(s), and these tasks encapsulate user logic to generate or receive the input messages that are passed to the dataflow. Likewise, \texttt{Sink} tasks have only incoming edge(s) and these react to the output messages from the application, say, by storing it or sending an external notification.
\texttt{Transform} tasks, sometimes called \emph{Map}~\footnote{Spark Programming Guide, http://spark.apache.org/docs/latest/programming-guide.html}, generate one output message for every input message received ($\sigma=1:1$). Their user logic performs a transformation on the message, such as changing the units or projecting only a subset of attribute values. \texttt{Filter} tasks allow only a subset of messages that they receive to pass through, optionally performing a transformation on them ($\sigma=N:M$, $N \ge M$). Conversely, a \texttt{FlatMap} consumes one message and emits multiple messages ($\sigma=1:N$). An \texttt{Aggregate} pattern consumes a \emph{window} of messages, with the window width provided as a \emph{count} or a \emph{time} duration, and generates one or more messages that is an aggregation over each message window ($\sigma=N:1$). Specific DSPS may expose additional dataflow patterns as well.
When a task has multiple outgoing edges, routing semantics on the dataflow decide if an output message is \emph{duplicated} onto all the edges, or just one downstream task is selected for delivery, either based on a \emph{round robin} behavior or using a \emph{hash function} on an attribute in the outgoing message to decide the target task.
Similarly, multiple incoming streams arriving at a task may be \emph{merged} into a single interleaved message stream for the task. Or alternatively, the messages coming on each incoming stream may be conjugated, based on order of arrival or an attribute exposed in each message, to form a \emph{joined} stream of messages. %be performed as a \texttt{Join}. However, \texttt{Joins} are infrequently supported in DSPS, and appear much more in DSMS. %While at the output end of a task, each message can be broadcasted to all the outgoing streams, called as \emph{Duplicate}, or the messages are sent to a particular stream based on control logic of the task, called as \emph{Split}.
%These dataflow patterns are more common as compared to other advanced dataflow patterns like BSP and iterative MapReduce which are discussed in paper~\cite{simmhan:tr:2014}.
%These patterns are not supported in common streaming frameworks. A possible reason is that most applications don't seem to require these.
Other custom DSPS routing semantics may exist too.
%\subsubsection{Dataflow Dimensions}
There are additional dimensions of the streaming dataflow that can determine its performance on a DSPS.
%
%Fig.~\ref{fig:semantics} mentions the selectivity ratio above each task. Operations like \texttt{Transform} have a uniform selectivity ratio of $1:1$ and \texttt{Filter} has a ratio $N:M$, where $M \le N$ and their values depend on the matches in the input data. \texttt{Aggregate} is applied on a windows of input tuples, and depending on whether it is a count or a time window, the selectivity of the task is fixed or variable, respectively. Also, looking at a coarser level, the dataflow application itself has a selectivity that is the ratio of the number of messages emitted at the sink tasks to the number of messages consumed at the source tasks.
%
Tasks may be \emph{data parallel}, in which case, it can be allocated concurrent resources (threads, cores) to process messages in parallel by different instances the task. This is typically possible for tasks that do not maintain state across multiple messages. The \emph{number of tasks} in the dataflow graph indicates the size of the streaming application. Tasks are mapped to computing resources, and depending of their degree of parallelism and resource usage, it determines the cores/VMs required for executing the application. % We can classify the size of applications and their benchmarks, based on this parameter
% It is to be noted that in a topology, if the user-logic is same for multiple tasks, then each occurrence of the task \emph{is counted}.
%
The \emph{length of the dataflow} is the latency of the critical (i.e., longest) path through the dataflow graph, if the graph does not have cycles. This gives an estimate of the expected latency for each message and also influences the number of network hops a message on the critical path has to take in the cluster. % Topology length is also defined in terms of time taken for a message to be processed along a critical path.
%
%\noindent \textbf{Width of the Dataflow} is the maximum number of vertex that can fall on a linear cut of the planar dataflow graph. This is an indication of the maximum degree of task parallelism present in the streaming dataflow, and can thus be exploited.
%
%Product of the two parameters (i.e. area) \emph{Length} and \emph{Width} of the topology gives an estimate of the other parameter \emph{Number of Tasks}.
%\tsnote{Discussion: The two parameters \emph{Number of Tasks} and \emph{Length of Topology} are closely related. How?}
%
% \noindent \textbf{Average Degree of the Dataflow} is, as the name suggests, the average number of incoming or outgoing edges at tasks of the dataflow graph. Tasks having a high \emph{out degree} of edges in the dataflow can indicate % This branching information about a dataflow can when compared against the edge degree of a task in it
% \emph{hot spots} in the application that will impact downstream tasks.
% % In particular, we can consider in-degree and out-degree of each task.
% For e.g., a task with a high out-degree whose out edges have \emph{Duplicate} semantics will increase load on the downstream tasks, as compared to a low out-degree task with \emph{Hash} semantics.
%
% %\emph{Degree of task parallelism} is defined as the maximum value for level-wise count of vertices starting from \emph{source tasks}.
% %Finding a \emph{max-cut} of the dataflow graph also indicates the \emph{degree of task parallelism} for the entire application. \tsnote{CHECK! max-cut may not always be correct. For example consider the ESC paper topology.}
% %\asnote{Better is to take, level wise or stage wise count of vertices and maximum of size of any stage is the maximum degree of task parallelism.}
% \noindent \textbf{Statefulness of a Task} refers to whether the output from a task depends only on the current input message that it is processing or also on state that it has collected from historic messages. Aggregate tasks are examples of stateful tasks that operate on a window of messages, and this fact is exposed to the DSPS. But there may be other tasks that implicitly maintain state as part of user logic, e.g., to calculate the running word count. Applications that have stateful tasks are deemed as stateful applications, and need DSPS that support state management.
%
%At the same time, elasticity of the Cloud can also be leverage by them. If a task required \emph{message order} to be maintained or if maintains a \emph{state} across multiple message executions (e.g. Aggregate), it is more difficult for DSPS to support data parallelism. % Data parallelism is not supported in those cases where processing of messages by tasks should have ordering constraints. It is supported when the tasks are \emph{stateless}.
%\emph{Degree of data parallelism}, is defined as the number of \emph{task instances} corresponding to each task. Each task can have different task instances, but, for simplification, we assume all the tasks to have same degree of parallelism and call it the degree of parallelism for the topology.
%Applying these dataflow dimensions to streaming applications can highlight its characteristics, that can further help optimize the application for effective performance or help choose an appropriate DSPS for its composition and execution. As an exemplar, we have performed this classification along these dimensions for several representative streaming applications found in literature, in Table~\ref{tbl:apps}. Several of these, such as power grid, transportation and crowd-sourcing, fall in the IoT domain.
\subsection{Data Stream Characteristics}
We list a few characteristics of the input data streams that impact the runtime performance of streaming applications, and help classify IoT message streams. %are relevant both to generic streaming applications as well as to IoT applications, and use these to identify representative real-world IoT datasets for our benchmarks.
The \emph{input throughput} in messages/sec is the cumulative frequency at which messages enter the source tasks of the dataflow. Input throughputs can vary by application domain, and are determined both by the number of streams of messages and their individual rates. This combined with the dataflow selectivity will impact the load on the dataflow and its individual tasks, and determine the output throughput.
\emph{Throughput distribution} captures the variation of input throughput over time. In real-world settings, the input data rate is usually not constant and DSPS need to adapt to this. There may be several common data rate distributions besides a \emph{uniform} one. There can be \emph{bursts} of data coming from a single sensor, or a coordinated set of sensors. A \emph{saw-tooth} behavior is seen in the ramp-up/-down before/after specific events. % such as the traffic near a stadium before the start of a game
A \emph{Normal} distribution can occur for diurnal (day vs. night) stream sources, with \emph{bi-modal} variations capturing peaks during the morning and evening periods of human activity. % that track human behavior. % various e-commerce sites, square-wave input pattern for offices where majority data gets generated during office hours say 8 AM to 8 PM, and so on.
Lastly, the \emph{message size} provides the average size of each message, in bytes. Often, the messages sizes remain constant for structured messages arriving from specific sensor or observation types, but may vary for free-text input streams or those that interleave messages of different types. This size help assess the communication cost of transferring messages in the dataflow.% It can be fixed or variable.
\section{Characteristics of IoT Applications and Streams}% \note{2.5pgs}}
\label{sec:features:iot}
Here, we categorize IoT tasks, applications and data streams used within DSPS, based on the domain requirements. These, together with the patterns and semantics discussed in the previous section, offer a search space for defining dataflows and workloads that meaningfully and comprehensively validate IoT applications on DSPS.
\subsection{Categories of IoT Tasks}
IoT covers a broad swathe of domains, many of which are rapidly developing. So, it is not possible to comprehensively capture all possible IoT application scenarios. However, DSPS have clear value in supporting the real-time processing, analytics, decision making and feedback that is intrinsic to most IoT domains. Here, we attempt to categorize these common processing and analytics tasks that are performed over real-time data streams.
\textbf{Parse.} Messages are encoded on the wire in a standard text-based or binary representation by the stream sources, and need to be parsed upon arrival at the application. Text formats in particular require string parsing by the tasks, and are also larger in size on the wire. The tasks within the application may themselves retain the incoming format in their streams, or switch to another format or data model, say, by projecting a subset of the fields. They may also annotate and extend the number of fields. Industry-standard formats that are popular for IoT domains include CSV, XML, SenML and JSON text formats, and EXI and CBOR binary formats. For e.g., IETF's \emph{SenML (Sensor Markup Language)}~\cite{senml} can define an array of entries, where each entry is an object object that encapsulates attributes and their values, such as the unique identifier for the sensor, the time of measurement, and the current value, with the ability to model repetitions, relative time, etc. SenML serializations into JSON, XML and EXI are possible.% Also tasks like projection for selecting values of particular attributes, annotation for adding metadata.%\ysnote{(JSON, XML, CBOR), Projection, Annotation}
\textbf{Filter.} Messages may require to be filtered based on specific attribute values present in them, for data quality checks, to route a subset of message types to a part of the dataflow graph, or as part of their application logic. % Value filters and bloom filters are examples of tasks in this category.
\emph{Value filters} such as min/max or band-pass filters check the numeric values of different observational fields from the sensors and can drop outliers. Since IoT event rates may be high, more efficient \emph{Bloom filters} are a probabilistic structure that can be used to process large sets of discrete values with low space complexity but a small fraction of false positives. It can be used to detect invalid sensors or users in an incoming data streams. Filtering over text or media streams is also possible, but requires consideration like using text or video processing libraries. % Value and band-pass filters that test an attribute's \emph{numerical value ranges} are common, and are both compact to model and fast to execute. % \ysnote{E.g. (Simple, Bloom)}
\textbf{Statistical Analytics.} Groups of messages within a sequential time or count window of a stream may require to be aggregated as part of the application. The aggregation function may be \emph{common mathematical operations} like average, count, minimum and maximum. They may also be \emph{higher order statistics} such finding outliers, quartiles, second and third order moments, and counts of distinct elements. Statistical \emph{data cleaning} like linear interpolation or denoising using Kalman filters are common for sensor-based data streams. Some tasks may maintain just local state for the window width (e.g., local average) while others may maintain state across windows (e.g., moving average). When the state size grows, here again approximate aggregation algorithms may be used. \emph{Distinct approximate count} is another such example of statistical tasks where we try to find approximate distinct values present in stream. % \ysnote{E.g. (Block Window Avg, Slide, Time, Distinct appox Count), Statistics (Moments, Kalman)}
\textbf{Predictive Analytics.} Predicting future behavior of the system based on past and current messages is an important part of IoT applications. Various statistical and machine-learning algorithms may be employed for predictive analytics over sensor streams. The \emph{predictions} may either use a recent window of messages to estimate the future values over a time or count horizon in future, or train models over streaming messages that are periodically used for predictions over the incoming messages.
Even simple techniques like \emph{interpolation} can be useful for replacing empty entries by interpolation over past values. \emph{Classification} algorithms like decision trees, neural networks and na\"{i}ve Bayes can be trained to map discrete values to a category, which may lead to specific actions taken on the IoT system. External packages like Weka or R may be used by such tasks.
%Other statistical operations like simple linear regression and Multi-variate linear regression can be used for prediction.
The \emph{training} itself can be an online task that is part of a DSPS dataflow. For e.g., ARIMA and linear regression use statistical methods to predict uni- or multi-variate attribute values, respectively. Also trained models can be updated on the fly within such forecasting tasks. %used to update the models at run time.
%\ysnote{e.g. (ARIMA, Interpolation, Linear regression) Classification (Weka Decision Tree testing, Naive Bayes, Perceptron). This also includes training of models}
\textbf{Pattern Detection.} Another class of tasks are those that identify patterns of behavior over several events. Unlike window aggregation which operate over static window sizes and perform a function over the values, pattern detection matches user-defined predicates on messages that may not be sequential or even span streams, and returns the matched messages. These are often modeled as \emph{state transition automata} or \emph{query graphs}. Common patterns include contiguous or non-contiguous sequence of messages with specific property on each message (e.g., high-low-high pattern over 3 messages), a join over two streams based on a common attribute value, or even semantic matching~\cite{zhou2016knowledge}. Complex Event Processing (CEP) engines like Siddhi~\cite{siddhi} may be embedded within the DSPS task to match such patterns.%\ysnote{(Siddhi CEP/SEQ, image recognition)}
\textbf{Visual Analytics.} Other than automated decision making, IoT applications often generate \emph{charts and animations} for consumption by end-users or system managers. These visual analytics may be performed at the client's browser using libraries like \texttt{D3.js}, in which case the processed data stream is aggregated and provided to the users. Alternatively, the streaming application may itself periodically generate such plots and visualizations as part of the dataflow, to be hosted on the web or pushed to the client. Charting and visualization libraries like \texttt{XChart}, \texttt{gnuplot} or \texttt{matplotlib} may be used for this purpose. % (JFreeChart?)
\textbf{IO Operations.} Lastly, the IoT dataflow may need to access external storage or messaging services to access/push data into/out of the application. These may be to store or load trained models, archive incoming data streams, access historic data for aggregation and comparison, and subscribe to message streams or publish actions back to the system. These require access to \emph{file storage, SQL and NoSQL databases,} and \emph{publish-subscribe messaging systems}. Often, these may be hosted as part of the Cloud platforms themselves like Azure Storage. This also include writing files to local or remote disk, and optionally compressing or uncompressing them. Each of them have their own characteristics in term of latency, peak rate supported and resource usage.% \ysnote{e.g. NoSQL/file/message ops (Blob, Table, MQTT publish)}
% \begin{itemize}
% \item Parsing (JSON, XML, CBOR), Projection
% \item Filter (Simple, Bloom)
% \item Aggregate (Block Window Avg, Slide, Time, Distinct appox Count)
% \item Pattern detection (Siddhi CEP/SEQ, image recognition)
% \item Prediction (ARIMA, Interpolation, Linear regression)
% \item Classification (Weka Decision Tree testing, Naive Bayes, Perceptron)
% \item Statistics (Moments, Kalman)
% \item NoSQL/file/message ops (Blob, Table, MQTT publish)
% \item Visualization (JFreeChart?)
% \end{itemize}
\subsection{Categories of IoT Applications}
The tasks from the above categories, along with other domain-specific tasks, are composed together to form streaming IoT dataflow applications. These domain dataflows themselves fall into specific classes based on common use-case scenarios, and loosely map to the Observe-Orient-Decide-Act (OODA) phases.
%\subsubsection{Common Categories of IoT Applications}
%\asnote{may need to include model training}\ysnote{add it as part of the classification, as a preprocessing. Note that it may be like a ``batch'' rather than stream}
%\ysnote{revise these general categories based of additional things we do}
\textbf{Extract-Transform-Load (ETL) and Archival} applications are front-line ``observation'' dataflows that receive and pre-process the data streams, and if necessary, archive a copy of the data offline. Pre-processing may perform data format transformations, normalize the units of observations, data quality checks to remove invalid data, interpolate missing data items, and temporally reorder messages arriving from different streams, annotate with the metadata. The pre-processed data may be archived to table storage, and passed onto subsequent dataflow for further analysis.
\textbf{Summarization and Visualization} applications perform statistical aggregation and analytics over the data streams to understand the behavior of the IoT system at a coarser granularity. Statistical analytics may include tasks such as finding approximate counts, identifying skewness in data distribution, and using linear regression for online trends. Such summarization can give the high-level pulse of the system, and help ``orient'' the decision making to the current situation. These tasks are often coupled with visualization tasks in the dataflow to present the summary status to end-users and decision makers.
\textbf{Prediction and Pattern Detection} applications use current information and historic models to help determine the future state of the IoT system, and ``decide'' if any reaction is required. They identify patterns of interest that may indicate the need for a correction, or forecasts based on current behavior that require preemptive actions. For e.g., a trend that indicates an unsustainably growing load on a smart power grid may cause a decision to preemptively shed load, or a detection that the heart-rate from a fitness watch is dangerously high may trigger a decision to reduce physical exertion. Model-based prediction applications are also coupled with batch or online dataflow applications that periodically re-train the models using observed data. %These also cover model based predictions where models can be updated at the run time.
\textbf{Classification and Notification} applications determine specific ``actions'' that are required and communicate them to the IoT system. Decisions may be mapped to specific actions, and the entities in the IoT system that can enact those are notified. These notifications can be delivered using SMS gateways, web service calls, or publish-subscribe brokers. For e.g., the need for load shedding in the power grid may map to notifying specific residents with a request for curtailment, or the need to reduce physical exertion may lead to a treadmill being notified to reduce the speed. The classification or case based reasoning systems may also require model training, like for predictive analytics.
\subsection{IoT Data Stream Characteristics}
IoT data streams are often generated by physical sensors that observe physical systems or the environment. As a result, they are typically time-series data that are generated periodically by the sensors. The sampling rate for these sensors may range from once a day to hundreds per second, depending on the domain. The number of sensors themselves may vary from a few hundred to millions as well. IoT applications like smart power grids can generate high frequency plug load data at thousands of messages/sec from a small cluster of residents, or low frequency data from a large set of sensors, such as in smart transportation or environmental sensing. As a result, we may encounter a wide range of input throughputs from $10^{-2}$ to $10^{5}$ messages/sec. In comparison, streaming web applications like Twitter deal with $6000$~tweets/sec from 300M users.
At the same time, this event rate itself may not be uniform across time. Sensors may also be configured to emit data only when there is a change in observed value, rather than unnecessarily transmitting data that has not changed. This helps conserve network bandwidth and power for constrained devices when the observations are slow changing. Further, if data freshness is not critical to the application, they may sample at high rate but transmit at low rates but in a burst mode. E.g. smart meters may collecting kWh data at 15~min intervals from millions of residents but report it to the utility only a few times a day, while the FitBit smart watch syncs with the Cloud every few minutes or hours even as data is recorded every few seconds.
%
Message variability also comes into play when human-related activity is being tracked. Diurnal or bimodal event rates are seen with single peaks in the afternoons, or dual peaks in the morning and evening. E.g., sensors at businesses may match the former while traffic flow sensors may match the latter.
There may also be a variety of observation types from the same sensor device, or different sensor devices generating messages. These may appear in the same message as different fields, or as different data streams. This will affect both the message rate and the message size. These sensors usually send well-formed messages rather than free-text messages, using standards like SenML. Hence their sizes are likely to be deterministic if the encoding format is not considered -- text formats tend to bloat the size and also introduce size variability when mapping numbers to strings. However, social media like tweets and crowd-sourced data are occasionally used by IoT applications, and these may have more variability in message sizes.
%Time-series data, events, CSV tuples, typed
%Uniform, uni and bi modal, burst rates
%Sometimes freeform text, social feeds
%Integrating different stream types
% like Los Angeles with 4M customers recording kWh every 15~mins. Besides spatial size, applications may also increase their sampling frequency to generate faster data. E.g. FitBit fitness tracker monitors health parameters once a second, per person. For benchmarking, we can simulate a larger population of sensors by multiplying the message rate by a fixed quantity so that the input rate distribution is maintained while the amplitude is increased.
% The characteristics discussed above are orthogonal to each other and different combinations will yield different scenarios for performing benchmarks.
% The main application characteristics that drive the motivation behind the experiments can be broadly classified with below characteristics:
% %\tsnote{Apr8: a) Application Overview from the table to be written here. 4 lines per app. For example:- used in which domain, Key distinguishing factor, LRB standard benchmark for continuous queries. \\
% %b) Refer these apps in later subsections. This will indirectly justify why you have included these apps.}
% \modc{We now define some candidate applications and provide insight as to how they are distinguished from each other. Detailed discussion about the application dimensions is in subsection~\ref{subsec:dataflow}.
% \emph{Lois} application~\cite{tang:tpds:2013} is taken from astronomy domain. Its topology structure has majority of tasks as white-box. Its length is larger as compared to width. It is a \emph{medium} sized application.
% \emph{Vwap} application~\cite{tang:tpds:2013} is the Volume-Weighted Average Price that can be used to detect bargain. It is of finance domain. It too has majority of the tasks as white-box. The topology is fat, i.e. width is higher than length. It is a \emph{medium} sized application.
% \emph{LinearRoad} application~\cite{tang:tpds:2013} is a variable tolling system on hypothetical highways. It works on real-time readings of positions from the different vehicles. It has majority of its tasks as black-box. The topology is long. It is a \emph{medium} sized application.
% \emph{PanSTARRS} application is used to detect hazardous objects in solar system. It is from astronomy domain. It is actually a batch-based application. But we would like to see this application as a real-time application. It is relatively small topology. But the tasks itself are compute-intensive and black-box. It is a \emph{small} sized application. However, the data rates consumed can be much larger.
% \emph{Citizen Journalism} application is used to broadcast video streams received from different locations and based on the locations, one can embed relevant twitter and other streams. It is from multimedia domain. The topology structure is a one-dimensional pipeline. It is a \emph{small} sized application.
% \emph{Smart Grid} application is used to analyze the smart-meter readings and take appropriate actions and demand forecast. It is taken from Power Grid domain. It is a balanced topology in terms of length and width. It is black-box. It is a \emph{small} sized application.
% \emph{Stock Correlation} application is used to find the most correlated stocks, based on the real-time readings from stock. It is a fat topology. It is taken from finance domain. It is a \emph{medium} sized application.
% }
\section{Performance Metrics}% \Note{1pg}}
%\asnote{1.5page put spark citation , put infosphere citation}
\label{sec:metrics}
We identify and formalize commonly-used quantitative performance measures for evaluating DSPS for the IoT workloads. % that can help identify a suitable DSPS for the target application.
%\begin{enumerate}[noitemsep,topsep=0pt,parsep=0pt,partopsep=0pt,leftmargin=0cm,itemindent=.5cm,labelwidth=\itemindent,labelsep=0cm,align=left]
\textbf{Latency.}
Latency for a message that is generated by a task is the time in seconds it took for that task to process one or more inputs to generate that message. If $\sigma=N:M$ is the selectivity for a task $T$, the time $\lambda^T_M$ it took to consume $N$ messages to \emph{causally produce} those $M$ output messages is the latency of the $M$ messages, with the \emph{average latency} per message given by $\overline{\lambda^T} = \frac{\lambda^T_M}{|M|}$. When we consider the average latency $\overline{\lambda}$ of the dataflow application, it is the average of the time difference between each message consumed at the source tasks and all its causally dependent messages generated at the sink tasks.
%It is measured in seconds or milliseconds. \[L_{o_k} = \frac{\sum_{t^o_k=1 \ldots m}{t^o_k - t^i}}{m} \mid i \rightarrow o_k \]
%gives the average latency times for a set of messages, $o_k$, that were output from the sink of the dataflow, and $|o_k| = m$. Here, $t^o_k$ is the timestamps at which the $o_k$ messages were \emph{causally generated} by an input message $i$ that entered the dataflow at time $t^k$. The average latency for the dataflow is the average of latencies of all output messages.
The latency per message may vary depending on the input rate, resources allocated to the task, and the type of message being processed. % Rather than consider the average latency over all input messages, it is better to examine the latency distribution across all messages to get a sense of quality of service, say, using box plots or using the $95^{th}$ percentile latency.
While this task latency is the inverse of the mean throughput, the \emph{end-to-end latency} for the task within a dataflow will also include the network and queuing time to receive a tuple and transmit it downstream.
\textbf{Throughput.}
The output throughput is the aggregated rate of output messages emitted out of the sink tasks, measured in messages per second.
% It is an indicator of amount of work done by the dataflow application.
The throughput of a dataflow depends on the input throughput and the selectivity of the dataflow, provided the resource allocation and performance of the DSPS are adequate. Ideally, the output throughput $\omega^o = \sigma \times \omega^i$, where $\omega^i$ is the input throughput for a dataflow with selectivity $\sigma$. It is also useful to measure the \emph{peak throughput} that can be supported by a given application, which is the maximum stable rate that can be processed using a fixed quanta of resources. % \Note{Peak throughput on a single thread, and multiple threads on a single core}
Both throughput and latency measurements are relevant only under \emph{stable conditions} when the DSPS can sustain a given input rate, i.e., when the latency per message and queue size on the input buffer remain constant and do not increase unsustainably.
\textbf{Jitter.} The ideal output throughput may deviate due to variable rate of the input streams, change in the paths taken by the input stream through the dataflow (e.g., at a \texttt{Hash} pattern), or performance variability of the DSPS. We use jitter to track the variation in the output throughput from the expected output throughput, defined for a time interval $t$ as,
\[ J_t = \frac{\omega^o - \sigma \times \omega^i}{\sigma \times \overline{\omega^i}} \] where the numerator is the observed difference between the expected and actual output rate during interval $t$, and the denominator is the expected long term average output rate given a long-term average input rate $\overline{\omega^i}$. In the case of an ideal DSPS, jitter will tend toward zero, even if there are instantaneous changes in the input rate.
\textbf{CPU and Memory Utilization.} Streaming IoT dataflows are expected to be resource intensive, and the ability of the DSPS to use the distributed resources efficiently with minimal overhead is important. This also affects the VM resources and consequent price to be paid to run the application using the given stream processing platform. % For dataflows that run within a single data center, the former is expected to be more of a bottleneck, and also impact the acquisition and release of VMs.
We track the CPU and memory utilization for the dataflow as the average of the CPU and memory utilization across all the VMs that are being used by the dataflow's tasks. The per-VM information can also help identify which VMs hosting which tasks are the potential bottlenecks and can benefit from data-parallel scale-out, and cases of over-allocation of resources. %fraction of is a dimensionless quantity. The metric shows us the bottleneck task in the topology. It says how much fraction of the time in last $\tau$ seconds the tasks spent time in processing the messages.
%\tsnote{Reference: http://stackoverflow.com/questions/20371073/how-to-tune-the-parallelism-hint-in-storm}
% Once the bottleneck task(s) are identified, parallelism in the form of \emph{task} or \emph{data} parallelism can be leveraged to distribute the excess load.
% \ysnote{This is like CPU utilization}
% \subsubsection{Price to Throughput Ratio} For a given output throughput, this ratio gives the price tat will be paid for executing a dataflow on a DSPS. This ratio helps evaluate the cost to value trade-off for the dataflow, and also helps us understand the benefits or shortcomings of scale-out/in. For weak scaling, the price increase per unit increase in throughput should be constant; and DSPS that support adaptivity can exhibit weak scaling even with variable input rates. % This measure is for elastic frameworks. Ideally, it should be a line with a slope of 1. If it is greater than 1, it will imply that the framework is not efficient. It also indicates the utilization.
% % Even in absence of a elastic framework,
% % Even for non-adaptive DSPS, this metric helps determine the number of VMs required to achieve an output throughput $\widehat{\tau^o}$. Also,
% Even without an adaptive DSPS, this ratio can help estimate the number of static VMs to provision for a particular throughput, and the maximum input rate beyond which the dataflow is unstable, i.e., the dataflow cannot keep up.
% %It addresses the \emph{weak-scaling}.
% %\end{enumerate}
% \subsection{Communication Overhead:}
% This metric helps us to measure communication overhead due to working in a distributed setting.
% For instance, for a given input rate \emph{r} what is the output rate \emph{with} communication overhead and other is without communication overhead. It aims to capture how much the network overhead causes drop in the output rate. This can be evaluated using a micro-benchmark of two tasks, one time taken on a single machine (i.e. using memory communication), other time both are on two different machines (i.e. using network communication).
%\tsnote{Apr8: Convert it to Efficiency metric. a) Constant machines what is the performance? \\
%b) To achieve rate r what is the number of machines?\\
%c) Given a input rate what is the output rate with overhead? Without communication overhead?\\
%d) WEAK Scaling\\
%e) Computation, Communication, and System Overhead, which platform is better in these terms?
%}
%Omitting below subsection as it needs significant signal processing domain knowledge.
%\subsection{Quiescence or transient and steady state response:} It can be used to measure how well the system adapts for the sudden bursts in the input rate. Reference: http://nptel.ac.in/courses/108103008/12
%\subsubsection{Weak Scaling} Can we increase the data rate with a proportional increase in number of VMs?
\section{\emph{RIoTBench} IoT Benchmark Suite}% \note{4.5pgs}}
\label{sec:benchmark}
We propose benchmark workloads to help evaluate the metrics discussed before for various DSPS. These benchmarks are in particular targeted for emerging IoT applications, to help them distinguish the capabilities of contemporary DSPS on Cloud computing infrastructure. The benchmarks themselves have two parts, the dataflow logic that is executed on the DSPS and the input data streams that they are executed for. We next discuss our choices for both.
% \emph{Message Content Type}, is the type of abstraction of a message with respect to a platform.
% Specifically, we use term \emph{messages} in those context where framework treats records as \emph{black-box}, and as \emph{tuples} where framework treats records as \emph{white-box}.
% Tuples have a schema or data fields of a message, which a platform is aware of. It is important to capture these classifications in benchmarks for two main reasons: firstly, the different data types for fields like boolean, numeric and string can have performance impact, and secondly, the way framework treats the records, i.e. white-box or black-box, can have performance impact.
%\tsnote{CHECK THIS: Citation problem, but as per url, it is in ICDE 2015. (http://dase.ecnu.edu.cn/Chronos/\#ct4)}
\subsection{IoT Micro-benchmarks}
%\ysnote{Classify the proposed benchmark tasks based on the dataflow dimensions}
We propose a suite of common IoT tasks that span various categories we have identified and different streaming task patterns. These tasks form independent micro-benchmarks, and are further composed into application benchmarks later. The goal of the micro-benchmarks is to evaluate the performance of the DSPS for individual IoT tasks, and we measure the \emph{peak input throughput} that they can sustain on a unit computing resource as the performance metric. This offers a baseline for comparison with other DSPS, and can also inform resource scheduling decisions for more complex application dataflows composed using these tasks. % when these tasks are used in application benchmarks with variable input rates.
\begin{table}[t]
\footnotesize
\centering
\caption{IoT Micro-benchmark Tasks with different IoT Categories and DSPS Patterns.}
\begin{tabular}{lccccc}
\hline
\textbf{Task Name} &~\textbf{Code}~&~\textbf{Category}~& \textbf{Pattern} &~\textbf{$\sigma$ Ratio}~&~\textbf{State}\\ \hline
\hline
Annotate & ANN & Parse & Transform &1:1&No\\
CsvToSenML & C2S & Parse & Transform &1:1&No\\
SenML Parsing~\cite{senml} & SML & Parse & Transform &1:1&No\\
XML Parsing & XML & Parse & Transform &1:1&No\\
\hline
Bloom Filter~\cite{bloom:acm:1970} & BLF & Filter & Filter &1:0/1&No\\
Range Filter & RGF & Filter & Filter &1:0/1 & No\\
\hline
Accumlator & ACC& Statistical & Aggregate &N:1&Yes\\
Average & AVG& Statistical & Aggregate &N:1&Yes\\
Distinct Appox. Count~\cite{durand:esa:2003} & DAC & Statistical & Transform &1:1&Yes\\
Kalman Filter~\cite{kalman:asme:1959} & KAL & Statistical & Transform &1:1&Yes\\
Second Order Moment~\cite{alon:stoc:1996} & SOM & Statistical & Transform &1:1&Yes\\
%Pi (Viete) &
\hline
Decision Tree Classify~\cite{quinlan:ml:1986} & DTC & Predictive & Transform &1:1&No\\
Decision Tree Train & DTT & Predictive & Aggregate &N:1&No\\
Interpolation & INP & Predictive & Transform &1:1&Yes\\
Multi-var. Linear Reg. & MLR& Predictive & Transform &1:1&No\\
Multi-var. Linear Reg. Train & MLT& Predictive & Aggregate &N:1&No\\
Sliding Linear Regression & SLR& Predictive & Flat Map &N:M&Yes\\
\hline
Azure Blob D/L & ABD& IO & Source/Transform &1:1&No\\
Azure Blob U/L & ABU& IO & Sink &1:1&No\\
Azure Table Lookup & ATL& IO & Source/Transform &1:1&No\\
Azure Table Range & ATR& IO & Source/Transform &1:1&No\\
Azure Table Insert & ATI& IO & Transform &1:1&No\\
MQTT Publish & MQP& IO & Sink &1:1&No\\
MQTT Subscribe & MQS& IO & Sink &1:1&No\\
Local Files Zip & LZP & IO & Sink &1:1&No\\
Remote Files Zip & RZP & IO & Sink &1:1&No\\
%zipping lib- https://docs.oracle.com/javase/7/docs/api/java/util/zip/ZipEntry.html
\hline
MultiLine Plot~\cite{xchart:lib} & PLT& Visualization & Transform &1:1&No\\
\hline\hline
\end{tabular}
\label{tbl:tasts}
%\ysnote{Verify that we also have online model training task? Any more citations?}
\end{table}
%\ysnote{these para below need to be updated based on changes to the table}
Table~\ref{tbl:tasts} lists the different micro-benchmark tasks, and their IoT categories, task patterns, and selectivity. These are grouped by their categories. The \emph{parse} category includes tasks that process standard text formats such as SenML and XML, and convert them to object formats, and also convert from a CSV format to a SenML form with additional semantics. The annotation task appends metadata content to an existing message based on an in-memory lookup for a unique ID present in the tuple. All these parse tasks transform messages from one form to another. % and SenML and XML parser as the representative parsing operation within our suite. % CsvToSenML for transforming the comma seperated message fromat to SenML format.
The Bloom filter finds practical use in the \emph{filter} category for processing a large, discrete data space. It is trained with a white-list of valid sensor IDs that it will permit. The simple value range filter is used filtering in messages with observation fields that fall within a fixed upper and lower bound.
We have several tasks in the \emph{statistical analytics} category that perform aggregations and transformations. Basic statistics include a simple average of a single attribute's values over a count window, and a generic accumulator that buffers incoming messages based on a count window for use by other tasks. The second order moments over time-series values is another common statistics we implement. Estimating the frequencies a large range of streaming values can be memory intensive, and distinct approximate count performs a probabilistic count over the incoming messages while conserving memory. Lastly, the Kalman filter we provide is a popular denoising algorithm used for smoothing sensor data values in a time-series.
\emph{Predictive analytics} uses the Weka library to implement several common Machine Learning tasks. A multi-variate linear regression is included to predict one attribute's numerical value based on the values of one or more other in the message. This has both online training and online prediction tasks. Similarly, the decision tree classifier is used for predict a class based on enumerated field values in the message, and also comes with a training and a classification task. Training for both these models happens over large, batched windows of messages. Interpolation and linear regression are standard techniques used for univariate time-series observation, and are available in the micro-benchmark suite. % model that is trained online are included. A decision tree machine learning model for discrete attribute values is also used for classification, based on offline training. Model training tasks for multi-variate linear regression and decision tree classification for training over batched datasets.
We have several \emph{IO} tasks for reading and writing to Microsoft Azure Cloud's file (blob) storage and NoSQL (table) storage. In addition, the common file operation of compressing a set of files is also available, with the source files being either on local disk or on the network. Publish/subscribe to/from an MQTT publish-subscribe broker for notifications and also included. Lastly, a single exemplar \emph{Visualization} task in the form of a Java XChart plotting library is present to accumulate and generate an image file. % for plotting the aggregated results. Due to limited space, we skip pattern matching task categories.
A micro-benchmark dataflow is composed for each of these tasks as a sequence of a source task, the benchmark task and a sink task. As can be seen, these tasks also capture different dataflow patterns such as transform, filter, aggregate, flat map, source and sink.
% \item Detection of sequence pattern using Siddhi CEP ???
% \item Interpolation
% \item Weka Naive Bayes Classification ???
% \item Table upload
% \item MQTT subscribe
% \item Visualize JFreeChart???
\subsection{IoT Application Benchmarks}
%\ysnote{Classify the proposed benchmark tasks based on the dataflow dimensions}
\begin{figure}[t]
\centering%~%
\includegraphics[width=0.4\textwidth]{figures/flow.pdf}
\caption{High-level logical interactions between different sensors, benchmark applications, platform services and users}
\label{fig:app:flow}
\end{figure}
Application benchmarks are valuable in understanding how non-trivial and meaningful IoT applications behave on DSPS. Application dataflows for a domain are most representative when they are constructed based on real or realistic application logic, rather than synthetic tasks. In case applications use highly-custom logic or proprietary libraries, this may not be feasible or reusable as a community benchmark. However, many of the common IoT tasks we have defined earlier are naturally composable into application benchmarks that satisfy the requirements of a OODA decision making loop. %the task of porting the application to DSPS may be non-trivial, and the effort to run complex application logic too cumbersome. , environment dependencies and security may be additional factors that impede this. As such, benchmarks should be reusable so that the overhead for running them on a new DSPS is low.
%
%An alternative to porting real applications is to mimic the behavior of common usecases in the domain, develop a dataflow composition that would achieve that usecase, but replace the application logic of the tasks with either simple versions of the real task or with micro-operations or spin-waits to simulate activity. The tasks would generate realistic message contents and selectivity, though the values themselves may not be correct from a domain perspective. This has the advantage of testing a realistic dataflow pattern with real values of selectivity and message content, that tests the DSPS, while being simpler to implement on multiple platforms. Such efforts have also been used in Big Data benchmarks where model applications like TeraSort and K-means clustering are used~\cite{hibench}.
%
Fig.~\ref{fig:app:flow} shows a high-level use case of such an IoT scenario that is generalizable to many domains such as smart power, transportation and fitness. This is achieved by the interaction between four different application dataflows we propose. Here, input streams from sensors in the domain arrive at a \emph{Extract-Transform-Load (ETL)} dataflow that performs data pre-processing and cleaning on the observations and archives it to Cloud table storage. Further, one output stream is published to the MQTT message broker so that clients interested in real-time monitoring can subscribe to it, while another copy is forked to the second dataflow which performs \emph{Statistical Summarization (STATS)}. This application does higher order aggregation and plotting operations, and stores the generated plots to Cloud blob file storage, from where web-pages can load the visualization files on browsers.
Concurrently, two dataflows support predictive analytics. \emph{Model Training (TRAIN)} periodically loads the archived data from the Cloud table store and trains forecasting models that are stored in the Cloud, and notifies the MQTT broker of an updated model being available. The \emph{Predictive Analytics (PRED)} dataflow subscribes to the broker and downloads the new models from the Cloud, and continuously operates over the pre-processed data stream from ETL to make predictions and classifications that can indicate actions to be taken on the domain. It then notifies the message broker of the predictions, that can independently be subscribed to by a user or device for action.% applications for training, as shown in Fig.~\ref{fig:app}.
% \note{Fig~\ref{fig:app:flow} shows high level interaction between incoming data from sensors, processing and data flow among applications. The data fetched from sensors is first preprocessed by ~\emph{ETL} application (parsing, filtering etc) for archiving and published to MQTT which can be subscribed by users or application. The output of ETL is then used for online statistical or predictive analyses using~\emph{STATS and PRED} applications followed by publishing and storing the results as per user requirements. The archived data from ETL can be used for batched online training by ~\emph{TRAIN} application which can be then used to update the models for PRED application, at fixed intervals.}
More specifically, \textbf{ETL} (Fig.~\ref{fig:app-etl}) ingests incoming data streams in SenML format, performs data filtering of outliers on individual observation types using a Range and Bloom filter, and subsequently interpolates missing values. It then annotates additional meta-data into the observed fields of the message and then inserts the resulting tuples into Azure table storage, while also converting the data back to SenML and publishing it to MQTT. % The SenML output of this preprocessing phase can be used as source for PRED and STATS applications. It is a commonly used application in different domains as streaming data needs to be archived for historical purposes.
A dummy sink task shown is used for logging purposes.
The \textbf{STATS} dataflow (Fig.~\ref{fig:app-stats}) parses the input messages that arrive in SenML format -- typically from the ETL, but kept separate here for modularity. It then performs three types of statistical analytics in parallel on individual observation fields present in the message: an average over a $10$ message window, % for all datasets (e.g. $\sim15$~mins native time window for TAXI),
Kalman filtering to smooth the observation fields followed by a sliding window linear regression, and an approximate count of distinct values that arrive. These three output streams are then grouped for each sensor IDs, plotted and the resulting image files zipped. These three tasks are tightly coupled and we combine them into a single meta-task for manageability, as is common. and the output file is written to Cloud storage for hosting on a portal.
%The outcomes from these statistics are then accumulated seperately for each sensor/taxi/meter IDs for plotting, then these plots are zipped for uploading to Azure Blob storage.
The \textbf{TRAIN} (Fig.~\ref{fig:app-train}) application uses a timer to periodically (e.g., for every minute) trigger a model training run. Each run fetches data from the Azure table available since the last run and uses ti to train a Linear Regression model. In addition, these fetched tuples are also annotated to allow a Decision Tree classifier to be trained. % For creating Decision Tree model, first a class is annotated dynamically based on quartiles of current training batch and then model is trained.
Both these trained model files are then uploaded to Azure blob storage and their files URLs are published to the MQTT broker. %These published messages can be parallely used by MQTT subscriber task of PRED application for model updation at run-time.
The \textbf{PRED} (Fig.~\ref{fig:app-pred}) application subscribes to these notifications and fetches the new model files from the blob store, and updates the downstream prediction tasks. Meanwhile, the dataflow also consumes pre-processed messages streaming in, say from the ETL dataflow, and after parsing it % , captures the lifecycle of online prediction and classification to drive decision making for IoT applications. It parses incoming messages and
forks it to the decision tree classifier and the multi-variate regression tasks. % decision tree uses a trained model to
The classifier assigns messages into classes, such as good, average or poor, based on one or more of their field values, while linear regression predicts a numerical attribute value in the message using several others. The regression task also compares the predicted values against a moving average and estimates the residual error % $\frac{|p-o|}{\overline{o}}$
between them. % value, normalized by the sliding average of the observations.
The predicted classes, values and errors are published to the MQTT broker. % , which can separately be subscribed to and visualized on a client browser or a mobile app. Apart from classifying and predicting new values, we are also doing an online update of models being used. The output of TRAIN topology is used as input in this. As the TRAIN topology reads and creates model, it sends a model update message to a topic through MQTT. MQTT subscribe in PRED receives the model update message with model name to Blob Read which reads the blob uploaded by TRAIN and updates the model files being used by Decision tree and Multi-linear regression tasks at the run time.
The Appendix lists configuration parameters and attributes used for relevant tasks in the dataflows for different workloads we benchmark them on.
\begin{figure}[t]
\centering%~%
% \subfloat[Pre-processing \& statistical summarization dataflow (STATS)]{%
% \includegraphics[width=0.5\textwidth]{figures/app-stats.pdf}%
% \label{fig:app-stats}
% }\\
% \subfloat[Predictive Analytics dataflow (PRED)]{%
% \includegraphics[width=0.5\textwidth]{figures/app-pred.pdf}
% \label{fig:app-pred}
% }\\
\subfloat[Extraction, Transform and Load (ETL)]{%
\includegraphics[width=0.7\textwidth]{figures/ETL.pdf}%
\label{fig:app-etl}
}\\
\subfloat[Statistical Summarization (STATS)]{%
\includegraphics[width=0.7\textwidth]{figures/stats.pdf}%
\label{fig:app-stats}
}\\
\subfloat[Model Training (TRAIN)]{%
\includegraphics[width=0.7\textwidth]{figures/Train.pdf}%
\label{fig:app-train}
}\\
\subfloat[Predictive Analytics (PRED)]{%
\includegraphics[width=0.7\textwidth]{figures/pred.pdf}%
\label{fig:app-pred}
}
%\asnote{sorry for not catching it beforehand : In STATS app (1)bloom in micro-benchmark table 1:0/1 here N:M , which one should i choose (2) Avg in STATS is 90/10:1 (city/taxi) and not 1:1 }
\caption{Application benchmarks composed using the micro-benchmark tasks.}% \ysnote{rename spout as source}\asnote{sel. of Group and Viz is 120:1}
\label{fig:app}
\end{figure}
As such, these applications leverage many of the compositional capabilities of DSPS. The dataflows include \emph{single and dual sources}; tasks that are composed \emph{sequentially, task-parallel} and as \emph{combined} meta-tasks; \emph{stateful and stateless} tasks; and \emph{data parallel tasks} allowing for concurrent instances. Each message in the data streams contains multiple observation fields, but several of these tasks are applicable only on univariate streams and some are meaningful only from time-series data from individual sources. Thus, the initial parse task for ETL and STATS uses a \emph{flat map} pattern ($\sigma=1:N$, where N is number of observational fields) to create observation-specific streams early on. These streams are further passed to task instances, grouped by their observation type and optionally their sensor/meter ID using a \emph{hash pattern}.
% data:taxi, Brian Donovan and Daniel B. Work. “Using coarse GPS data to quantify city-scale transportation system resilience to extreme events.” to appear, Transportation Research Board 94th Annual Meeting, August 2014.
\subsection{IoT Input Stream Workloads}
We have identified four real-world IoT data streams available in the public domain as candidates for our benchmarking workload. These correspond to domains within smart cities, which is a major contributor to the growth of IoT, taxi cab services, and personal fitness. Their features are shown in Table~\ref{tbl:datasets} and their message rate distribution is in Fig.~\ref{fig:data-distribution}.
%\begin{enumerate}[noitemsep,topsep=0pt,parsep=0pt,partopsep=0pt,leftmargin=0cm,itemindent=.5cm,labelwidth=\itemindent,labelsep=0cm,align=left]
\begin{table}[t]
\centering
\footnotesize
\caption{Characteristics of Smart Cities data stream workloads used in benchmarks, with temporal and spatial scaling.}
\resizebox{\linewidth}{!}{%
\begin{tabular}{c|*{4}c|*{3}c|cc}
\hline
& \multicolumn{4}{c|}{\textbf{Raw Workload}}&\multicolumn{3}{c|}{\textbf{Scaling Factor}}&\multicolumn{2}{c}{\textbf{Effective Workload}}\\
\textbf{Name}&\emph{Sensors}&\emph{Attrib.$^*$} & \emph{Size} $(bytes)$ &\textbf{Distribution} &\emph{Temporal}&\emph{Spatial}&\emph{Effective} & \emph{Peak Rate} $(msg/sec)$ & \emph{Sensors} \\ \hline\hline
\textbf{CITY}~\cite{data:city} & 90 & 9 & 380 & Uniform & $30\times$ & $30\times$ & $900\times$ & 5,000 & 2,700 \\ \hline
\textbf{FIT}~\cite{data:fit} & 10 & 26 & 1,024 & Uniform & $1\times$ & $1\times$ & $1\times$ & 500 & 10 \\ \hline
\textbf{GRID}~\cite{data:grid} & 6,435 & 3 & 130 & Normal & $1\times$ & $500\times$ & $500\times$ & 10,000 & 32,17,500 \\ \hline
%\textbf{PLUG}~\cite{data:plug} & 8 & 29 & 1660 & $10 \times$ & Normal \\ \hline
\textbf{TAXI}~\cite{data:taxi} & 20,355 & 17 & 191 & Bimodal & $1,000\times$ & $1\times$ & $1,000\times$ & 4,000 & 20,355 \\ \hline \hline
\multicolumn{10}{c}{\emph{$^*$ Every dataset has a minimum of three attributes: sensorId, timestamp and one (or more) observational field(s).}}\\
%? & ? & Email & ? & ? \\
\end{tabular}}
\label{tbl:datasets}
\end{table}
\textbf{Sense your City (CITY)~\cite{data:city}.}
%\scnote{city data originally had 90 unique sensors 2670 unique sensor ids generated by folding 30 days data to single day} .
This is an \emph{urban environmental monitoring} project~\footnote{\texttt{http://map.datacanvas.org}} that has used crowd-sourcing to deploy sensors at $7$ cities across $3$ continents in 2015, with about $12$ sensors per city. Five timestamped observations, outdoor temperature, humidity, ambient light, dust and air quality, are reported every minute by each sensor along with metadata on sensor ID and geolocation. Besides urban sensing, this also captures the vagaries of using crowd-sourcing for large IoT deployments. Data from over $2$ months is available.
%
%We have done spatial scaling of $30\times$ by folding 30 days worth data into single day to simulate resulting in increase in number of sensors from 90 to 2700.
%
We use a single logical stream that combines the global data from all unique sensors provided in the dataset. % Since a practical deployments of environmental sensing can easily extend to thousands of such sensors per city, we use a temporal scaling as well as spatial scaling on actual data. A spatial scaling of $30\times$ is done by folding 30 days worth data into a single day which simulate a larger deployment of $2700$ unique sensors. We have also done a temporal scaling of $30\times$ for running 6 hours data into 12 min experimental benchmark run.
Fig.~\ref{fig:data:sys} shows a narrow distribution of the message rate, with the peak frequency centered at $5,000$~msg/sec.
%% old text used in tpctc paper --commented out and reused parts of this with new scaling ---by shilpa
%We use a single logical stream that combines the global data from all $90$ sensors. Since a practical deployments of environmental sensing can easily extend to thousands of such sensors per city, we use a temporal scaling of $1000\times$ the native input rate to simulate a larger deployment of $90,000$ sensors. Fig.~\ref{fig:data:sys} shows a narrow normal distribution of the event rate centered at $6,400$~msg/sec with a peak that reaches $7,000$ msg/sec. We use 7 days of data from 27-Jan-2015 to 2-Jan-2015 for our benchmark runs.%
%\texttt{timestamp,source,longitude,latitude,temperature,humidity,light,dust,airqualityraw}\\
%\texttt{2015-01-27T00:00:00.000Z,ci4ue1845000102w7ni64j7pl,-71.106167,42.372802,-0.1,65.3,0,367.38,26}
% \subsubsection{Smart Plug (PLUG)~\cite{data:plug}}
% This dataset represents the \emph{smart power grid} domain~\footnote{\texttt{http://www.cse.iitb.ac.in/debs2014/?page\_id=42}} and captures the plug-loads and power usage for $2125$ power outlets at $40$ houses in 2013. Messages are of either of these types. The load data is sampled ans sent periodically while the power data is sent only when $1$~kWh is accumulated, and hence we see a wider normal distribution of event rates centered at a mean of $1660$~msg/sec. We use the message native rate for this workload in our benchmarks. %from within It has data from different devices of households. The sensor readings emit periodically the load of the devices. The data arrival rate is steady.
\textbf{NYC Taxi cab (TAXI)~\cite{data:taxi}.} %\ysnote{Add proper URL to download data to this bibtex entry}
This offers a stream of \emph{smart transportation} messages that arrive from $2M$ trips taken in 2013 on $20,355$ New York city taxis equipped with GPS. A message is generated when a taxi completes a single trip, and provides the taxi and license details, the start and end coordinates and timestamp, the distance traveled, and the cost, including the taxes and tolls paid. Other similar transportation datasets are also available~\footnote{\texttt{https://github.com/fivethirtyeight/uber-tlc-foil-response}}, though we chose ours based on the richness of the fields.
%
This data has a bi-modal event rate distribution that reflects the morning and evening commutes, with peaks at $300$ and $3,200$~events/sec. We use 7~days of data from 14-Jan-2013 to 20-Jan-2013 for our benchmark runs. % an average of $\sim 400$ msg/sec.
% Sample Dataset - \\\{taxiId,hackLicense,pickupDatetime,timestamp,triptimeInSecs,tripDistance,pickupLong,pickupLat,\\dropoffLong,dropoffLat,paymentType,fareAmount,surcharge,mtaTax,tipAmount,tollsAmount,totalAmount\}\\
% \{024BE2,9F5FE5,2013-01-14 03:57:00,2013-01-14 04:23:00,1560,19.36,-73.953178,40.776016,\\-73.779190,40.645145,CRD,52.00,0.00,0.50,13.00,4.80,70.30\}\\
\textbf{Energy dataset (GRID)~\cite{data:grid}.}
This is a univariate dataset that reports the energy consumption for each smart meter in a pilot smart grid deployment in Ireland. The actual dataset had $6,435$ unique sensors and emits a reading every half an hour. % We have done a spatial scaling of $500 \times $ by folding $500$ days data into a single stream simulating $500 \times$ sensors resulting in $32,17,500$ uniques sensors in folded data.
Data is available from over 500~days of observations. It shows a normal distribution of data around each half an hour timestamp. The final peak rate of dataset used in the experimental runs is $10,000$~events/sec.
\textbf{MHealth dataset (FIT)~\cite{data:fit}.}
The MHEALTH (Mobile HEALTH) dataset consists of body motion and vital signs recordings for ten volunteers of diverse profiles collected when performing physical activities. Sensors measure in different parts of the subject's body % placed on the subject's chest, right wrist and left ankle are used to measure the motion experienced by diverse body parts, namely,
collect acceleration, rate of turn, magnetic field, and ECG data, among others, at a constant rate of $50~Hz$. % orientation. The sensor positioned on the chest also provides 2-lead ECG measurements, which can be potentially used for basic heart monitoring, checking for various arrhythmias or looking at the effects of exercise on the ECG.
% We have kept the scaling same as original dataset with readings taken at $50Hz$.
We the merge 10 subjects data into a single global stream, with messages having the subject ID as sensor ID. It has a constant rate of $500$~events/sec as shown in Fig.~\ref{fig:data:health}. %The number of observations provided are 26 which includes 24 different readings taken at 50Hz interval, subjectId and timestamp.
\begin{figure}[t]
\centering%~%
\subfloat[CITY]{%
\includegraphics[width=0.3\textwidth]{plots/dataset/sys-data-histogram.pdf}%
\label{fig:data:sys}%
}
\subfloat[TAXI]{%
\includegraphics[width=0.3\textwidth]{plots/dataset/taxi-distribution.pdf}%
\label{fig:data:taxi}%
}\\
\subfloat[GRID]{%
\includegraphics[width=0.3\textwidth]{plots/dataset/grid-data-histogram.pdf}%
\label{fig:data:energy}%
}
\subfloat[FIT ]{%
\includegraphics[width=0.3\textwidth]{plots/dataset/fit-data-histogram.pdf}%
\label{fig:data:health}%
}
\caption{Frequency distribution of input throughput for the four workloads, with the temporal and spatial scaling used for the benchmark runs.}
%\ysnote{Add a line on rate without temporal scaling?}
% \ysnote{Add a plot on the number of (spatially scaled) sensors in each dataset}}
\label{fig:data-distribution}
\end{figure}
%\ysnote{Discuss about spatial and temporal scaling here, in general, before using them in the text below. What are they? Why do we do them? How? What is the impact on distribution, number of sensors?}
While these datasets correspond to real values collected from the domain, they are representative samples from even larger datasets that are typically proprietary. In order to capture the real scale of these data streams, we make use of temporal and spatial scaling. % We are supporting spatial and temporal scaling on real dataset as shown in table~\ref{tbl:datasets}.
\emph{Temporal scaling} allows us to accelerate the data rate by time-compressing messages that were generated over a longer interval into a smaller one. % run a larger time frame data in a smaller time frame and
% affects the rate of events present in actual distibution.
For e.g., when the CITY data is temporally scaled by $30\times$, its original rate per sensor goes from an average of $6$~msg/min to $180$~msg/min, and $7$ days of wall-clock time get reduces to $336$~mins of benchmark time. This causes the shape of the distribution in Fig.~\ref{fig:data-distribution} to be retained, but widens the X Axis. Temporal scaling is relevant when the raw workload data that is available is not representative of the sampling rates that are expected in contemporary IoT sensors and domains. Considering that GPS sensors placed in taxis report their location each second for navigation and monitoring rather than only at the end of the trip, we use a temporal scaling factor of $1000\times$ for the TAXI workload. % ran with $10\times$ temporal scaling will give $5,000$~msg/sec. However, it does not affects the distribution of data like if the actual distribution is normal, it will still be a normal distribution in the scaled version.
% Rather than change existing data or duplicate them, we instead fold data from different time periods into their previous
% We perform temporal In TAXI we have done $1,000\times$ scaling to convert 7 days worth data into 12 min of wall clock time of experimental run.
\emph{Spatial scaling}, on the other hand, allow us to simulate a larger number of sensors than available in the raw data. % scale deployment of sensors from limited sensors present in actual dataset.
This is necessary when data streams are available only from a small sample of sensors. Here, we consider data streams from the same sensor but during different time windows (e.g., days) to act as if they are from different sensors but at a precious time. This too does not affect the shape of the message rate distribution, but expands the Y Axis. For e.g., tn the GRID data, a spatial scaling of $500\times$ increases the $6,435$ smart meters present in original dataset to $3,217,500$ unique meters, which is more representative of a city-scale deployment. % in the scaled version. This allow us to simulate a city level or larger deployment from a limited data.
Similarly, a $30\times$ spatial scaling in CITY (in addition to the temporal scaling) causes the 336~mins of benchmark time to further reduce to 12~mins of benchmark time, while % . we are doing both a $30 \times$ spatial scaling to
increasing the sensor count from $90$ to $2,700$. % followed by a temporal scaling of $30 \times$ to run 6hours of data in approximately 12 min of experimental run.
These two scaling factors are also shown in Table~\ref{tbl:datasets}, along with the effective number of sensors and the peak rate after applying these factors. As we can see, using scaling to create workloads that are representative of real-world scenarios also achieves diversity in the event rate distribution profiles for the input streams, and also offers peak rates that span from 500 to 10,000~msg/sec.
% \item \emph{Email} It is an Email classification benchmark application for online spam detection. It is designed by IBM Infosphere Streams.
%\end{enumerate}
% histogram of time distance between events to their frequency.
% histogram of event rate to the frequency for that event rate.
%\sout{We propose application benchmarks that capture two common IoT scenarios: a \emph{Data pre-processing and Statistical summarization (STATS)} application and a \emph{Predictive Analytics (PRED)} application. STATS (Fig.~\ref{fig:app-stats}) ingests incoming data streams, performs data filtering of outliers on individual observation types using a Bloom filter, and then does three concurrent types of statistical analytics on observations from individual sensor/taxi IDs: sliding Average over a $90/10$ event window for CITY/TAXI ($\sim15$~mins native time window), Kalman filter for smoothing followed by a sliding window linear regression, and an approximate count of distinct readings. The outcomes from these statistics are published by an MQTT task, which can separately be subscribed to and visualized on a client browser or a mobile app. The dummy sink task is used for logging.}
%\\
%\sout{The PRED dataflow captures the lifecycle of online prediction and classification to drive visualization and decision making for IoT applications. It parses incoming messages and forks it to a decision tree classifier and a multi-variate regression task. The decision tree uses a trained model to classify messages into classes, such as good, average or poor air quality, based on one or more of their attribute values. The linear regression uses a trained model to predict an attribute value in the message using several others. It then estimates the error $\frac{|p-o|}{\overline{o}}$ between the predicted and observed value, normalized by the sliding average of the observations. These outputs are then grouped and plotted, and the output file written to Cloud storage for hosting on a portal. One realistic addition is the use of a separate stream to periodically download newly trained classification and regression models from Cloud storage, and push them to the prediction tasks.}
% We categorize dataflow benchmarks for evaluating DSPS into two major types, \emph{micro-pattern} benchmarks and \emph{application} benchmarks. The former is a general purpose benchmark for DSPS while the latter is customized for IoT applications.
%Classification of different types of benchmarks based on underlying application logic: synthetic (or random topology) Vs real-world domain applications.
%The paper~2013} considers three categories for benchmarking viz., microbenchmarks, synthetic, and application benchmarks. The experiments are performed on System S.
% \subsubsection{Micro-Pattern Benchmarks}
% The dataflow patterns we identified in \S~\ref{sec:semantics} and Fig.~\ref{fig:semantics} are frequently occurring ones, and full applications are composed as a combinations of these. Micro-pattern benchmarks target at measuring the behavior of a DSPS for each of these design patterns, using a practical and common functional as the user logic within these patterns. Micro-patterns vary in their selectivity, edge degrees and statefulness, but often have just a couple of tasks at most, besides the source and the sink, and are of short lenghts and widths. The emphasis is on avoiding noise that may appear when varying many different dimensions, and offer a baseline for comparison across DSPS on the stream workloads for the proposed metrics.
% We take this latter approach and define two ``light'' but realistic application dataflows from the IoT domain, and suiting the real input dataset we have identified. First is a pipeline for \emph{Real-Time forecasting and storage of observational data (FCAST)} that is inspired by our prior work on the \emph{Los Angeles Smart Grid project}~\cite{escience2012}. The second is a streaming dataflow for calculating the \emph{Toll and revenue collection (TOLL)} for smart transportation networks, which is based on the popular \emph{Linear Road Benchmark} using in DSMS~\cite{lrb}.
% The dataflow for \textbf{FCAST} application is shown in Fig.~\ref{fig:app-obs}, and is applicable to CITY and PLUG streams. Here, a task first \emph{parses} and transforms the input message strings having time-series observations, and performs data quality \emph{validation} based on range bounds as a filter. It sends a \emph{duplicate} copy of the valid messages to a \emph{sequence} of forecasting and data store segments. The forecast tasks first perform an \emph{interpolation} to plug missing values, for which we use a selectivity of $5$:$6$ (i.e. $16.6\%$ missing values), and then performs a \emph{linear regression} on each of \emph{P} observational values to generate $10$ intervals of predictions as a single message, for each observation. The store sequence checks if a backend data store already contains similar messages, and if so \emph{fetches} and \emph{updates} them, or otherwise just \emph{inserts} the message. These map to file I/O operations.
% \ysnote{TOLL if we're able to implement and test.}
% These are candidate applications taken from a real-world domain. These type of benchmarks can be run in two manners, first, which we term as \emph{simulation benchmarks}, is by taking the topology structure as it is from a real-world application, but assigning a particular distribution to the latency at each processor, and second, is to run the job with the same topology structure and real production code for each processor.
% Both these types of benchmarks are one step ahead of synthetic benchmarks, and are easier to run. Simulation benchmarks may not capture all the necessary characteristics, like using NOP as task logic may not give a good indicator of the CPU utilization.
% The first set of benchmarks produce \emph{representative} results while second set of benchmarks produce the \emph{actual} results, exposing a lot more features. But the latter, comes with more overhead, licensing, security and privacy issues.
% Ideally, we want these benchmarks to be common but not trivial and should be easy to port.
% HiBench benchmarking suite too have included different real-world applications. For example, it includes machine learning applications like bayesian classification and K-Means clustering algorithms.
% As seen from the table~\ref{tbl:apps}, there are multiple S and M apps so we come up with an informal rule to classify an app as S/M if majority (out of the three fields) are of that class.
% The table~\ref{tbl:smlclassify} shows the classification guideline to be applied.
% \begin{table*}[th]
% \centering
% \caption{S/M/L Classification}
% \begin{tabular}{|c|c|c|c|p{3.7cm}|}
% \hline
% Dimension & S & M & L & Remarks \\ \hline
% NBolts & O(10) & O(100) & O(1000) & Approx. Equal \\ \hline
% Width & O(10) & O(100) & - & Less Than \\ \hline
% Length & O(10) & O(100) & - & Less Than \\ \hline
% Makespan & O(10) & O(100) & O(1000) & Less Than; Unit sec or ms?; Indirectly calc by inverting throughputs? \\
% \hline
% \end{tabular}
% \label{tbl:smlclassify}
% \end{table*}
% Finally, we bracket down to take one of the application from each of the S / M / L categories. Out of these we choose SmartGrid as S, VWAP (\tsnote{Need to send email regarding individual latencies of the tasks in the topology structure of this, and also look for other VWAP applications on internet} as M, and a synthetic topology for L category $(>100)~Num~Bolts$. The design of the synthetic topology can be with length 40 and first part of the graph with larger width of 30 with Fork/Join semantics and second part of the graph with smaller width of 10 tasks connected in Map/Reduce pattern. \tsnote{Apply Duplicate or Split Semantics here?}.
% Real-world application benchmarks are much better indicators of metrics whenever their is a match of application characteristics with a corresponding proposed benchmark. Similarly, when a particular metric is to be evaluated for a platform then micro-benchmarks are best suited. Micro-benchmarks help us to evaluate all of the evaluation metrics. However, the \emph{synthetic} and \emph{real-world-simulation} benchmarks can be used when both these category of application benchmarks do not meet an end-users requirements.
% \modc{Rationale behind including this type of benchmark suite is to fill the gaps when moving from \emph{Micro Application Benchmarks} to \emph{Real-World Application Benchmarks}. For a typical use-case, synthetically generated application benchmark can help to perform benchmarks that are as close to the real-world use-case within a short frame of \emph{time} and \emph{effort}.}
% \modc{
% Since, historical application traces are easily available for many large-scale real applications,
% one can make use of these traces to extract various parameters about data and run the synthetic benchmarks. This is a \emph{faster} means to see and compare the results, rather than deploying the real-production code on different DSPs.
% Thus, synthetic application benchmarks are aimed at filling the gap for those users that cannot directly find a proposed real-world application benchmark as per their requirements.
% }
%that when integrated together in different combinations, can reveal some class of real-world applications.
%These are well suited in situations where for applications, where historical traces can reveal lot more insight about the application.
%One may randomly generate the topology structure as well as the latency incurred at each of the processor.
%For example, in this paper, reverse tree dataflow is used in which leaf of the tree is considered to be the source operator in the dataflow, and root is the sink operator. This represents a scenario where there is a high degree of sharing of operators among the threads. Also, Random graph topology, are constructed by varying the sample ratio parameter which indicates the degree of overlapping of operators between threads.
% Other key idea is to take real-world scientific workflow which generally process for hours or days, and each processor has a latency of the order of minutes, and scale these down by a factor such that each processor attains a latency of the order of milliseconds to seconds. One such example is PanSTARRS ~\cite{ramakrishnan:workflow:2010} workflow.
% That is, we can identify various batch-mode applications and run them in continuous mode after applying certain constraints. For example, Map-Reduce incremental version can be one such candidate application. We can transform any DAG to a continuous DAG as an execution semantic.
%\tsnote{Apr8: a) Emphasize that since we have synthetic benchmarks we can have complete \emph{real-application coverage}.\\
%b) For complete \emph{Performance Metrics} coverage we have microbenchmarks [Best-case scenario]. Similar argument for \emph{Barebone benchmarks}.\\
%c) Create a table with rows as performance metrics and columns with the app characteristics like long/medium/short chains of topologies, etc.\\
%d) Coverage perspective to be thought for each performance metric and app characteristics. Why this benchmark? Which perf metric is evaluated by which benchmark?\\
%e) Look for more papers from EDBT example Paolo et al. http://openproceedings.org/EDBT/2014/paper\_199.pdf\\
%http://dl.acm.org/citation.cfm?id=2320789\\ http://ieeexplore.ieee.org/xpls/abs\_all.jsp?arnumber=6678504\&tag=1\\
%http://grids.ucs.indiana.edu/ptliupages/publications/survey\_stream\_processing.pdf}
%1) Email classification for online spam detection [3] on both IBM-INFO. and Storm - IBM - INFOSPHERE
%
%2) Microbenchmark -to estimate the impact of tuple transfer on CPU utilisation and thruput w.r.t.
%
%3) Barebone benchmark - removing the application logic and pushing raw tuples through Storm, i.e. the topology graph remains the same but the bolts are identity (no-op).
\section{Evaluation of Proposed Benchmarks}% \note{4.5pgs}}
\label{sec:results}
\subsection{Benchmark Implementation}
We implement the 27 micro-benchmarks as generic Java tasks that can consume and produce objects. These tasks are building blocks that can be composed into micro-dataflows and the ETL, STATS, PRED and TRAIN dataflows using any DSPS that is being benchmarked. To validate our proposed benchmark, we compose these dataflows on the Apache Storm open source DSPS, popular for fast-data processing, using its Java APIs. We then run these for the four stream workloads and evaluate them based on the metrics we have defined. The benchmark is available online at \texttt{\url{https://github.com/dream-lab/riot-bench}}.
% We use Storm's Java API to compose the dataflow graphs explicitly for the benchmarks and all tasks extend from the \texttt{BaseRichBolt}, with the logic present in the \texttt{emit} method that is called once per input message by the DSPS.
In Storm, each task logic is wrapped by a \emph{bolt} that invokes the task for each incoming tuple and emits zero or more response tuples downstream. % Some of the micro-benchmarks like the ones in statistical category and Moving Linear Regression are stateful in nature while others are stateless.
The dataflow is composed as a \emph{topology} that defines the edges between the bolts, and the \emph{groupings} which determine duplicate or hash semantics. %The logic for each task in the \emph{STATS} and \emph{PRED} applications is re-used from the API created from micro- benchmarks.\\
We have implemented a scalable data-parallel event generator that acts as a source task (\emph{spout}). It loads time-series tuples from a given SenML file and replays them as an input stream to the dataflow. While the spatial scaling of the workloads is performed offline as a pre-processing step, our generator can perform temporal scaling online, as it emits the message. % optionally scaled to go faster or slower.
% -- at a constant rate, at the maximum possible rate, or at intervals determined by the timestamps.
We generate random integers as tuples at maximum rate for the micro-benchmarks, and replay the original datasets by scaling their native rates as in table~\ref{tbl:datasets} for the application benchmarks, following the earlier distribution. % scaling for CITY and TAXI datasets for the application benchmarks made available to the topologies. This simulator library was called from within the source task logic of Storm.
\subsection{Experimental Setup}
We use Apache Storm $1.0.1$ running on OpenJDK 1.7, and hosted on \emph{Ubuntu 14.0} Virtual Machines (VMs) in the Southeast Asia data center of Microsoft Azure public cloud. For the micro-benchmarks, Storm runs the task being benchmarked on one exclusive \texttt{D1} size VM ($1$ Intel Xeon E5-2660 core at 2.2~GHz, $3.5$~GiB RAM, $50$~GiB SSD), while the supporting source and sink tasks and the master service run on a \texttt{D4} size VM ($8$ Intel Xeon E5-2660 core at 2.2~GHz cores, $28$~GiB RAM, $400$~GiB SSD). % , \emph{with one worker slot per core}\ysnote{do the spout and sink get only one worker instance? Do they scale?}\asnote{No they can have any number of worker instances depending on the input rate needed . YES they are scalable. }.
The larger VM for the supporting tasks and services ensures that they are not the bottleneck, and helps benchmark the peak rate supported by the micro-benchmark task on a single core VM.
For the ETL, STATS, TRAIN and PRED application benchmarks, we use \texttt{D3}~VMs ($4$ Intel Xeon E5-2660 core at 2.2~GHz cores, $14$~GiB RAM, $200$~GiB SSD) for all the tasks of the dataflow, while reserving additional \texttt{D4}~VMs to exclusively run the source and sink tasks, and the Storm master service. Storm requires the users to explicitly assign the data parallelism per task, and the total number of resources in the cluster. We determine the number of cores and data parallelism required by each task using a simple resource allocation algorithm, as follows.
We have the peak rate supported by the single-threaded task on a single core as given by the micro-benchmarks, and the peak rate seen for that task for a given application and stream workload by examining the dataflow and selectivity. For tasks where the expected rate in the dataflow is less than its peak rate supported on one core, we assign it an exclusive core and two threads. In cases that are I/O bound (e.g., MQTT, Azure storage) rather than CPU bound, we require multiple task instances on a single core to leverage data parallelism, and sometimes multiple cores as well. Table ~\ref{table:slots} shows the number of cores and VMs assigned for running the experiments with the applications and stream workloads.
\begin{table}[t]
\centering
\footnotesize
\caption{The number of resources assigned, given as ``cores, VMs'', for each application benchmark and workload. Each VM has 4 cores.}% \ysnote{Use same table style as Tbl 2}}
\label{table:slots}
\begin{tabular}{c|c|c|c|c}
%\begin{tabular}{c|p{1.95cm}p{1.95cm}p{1.95cm}p{1.95cm}}
\hline
\textbf{App.}&\qquad \textbf{CITY}\qquad\qquad &\qquad \textbf{FIT}\qquad\qquad &\qquad \textbf{GRID}\qquad\qquad &\qquad \textbf{TAXI}\qquad\qquad \\ \hline\hline
\textbf{ETL} & 11, 3 & 8, 2 & 14, 3 & 10, 3 \\ \hline
\textbf{STATS} & 27, 7 & 10, 3 & 11, 3 & 32, 8 \\ \hline
\textbf{TRAIN} & 7, 2 & 7, 2 & N/A$^\dagger$ & 7, 2 \\ \hline
\textbf{PRED} & 10, 3 & 9, 3 & N/A$^\dagger$ & 9, 3 \\ \hline
\hline
\multicolumn{5}{c}{\emph{$^\dagger$ Benchmarks are not done for the particular applications with the GRID dataset as}}\\
\multicolumn{5}{c}{\emph{it is univariate and DTC and MLR tasks require multiple fields.}}
\end{tabular}
\end{table}
We log the ID and timestamp for each message at the source and the sink tasks in-memory to calculate the latency, throughput and jitter metrics. %\ysnote{Source and sink on same VM, right?}\asnote{YES, in all cases source and sink are on same VM.} % Given the distributed setup, we also calculate the time skew between the machines, both before and after the experiments, and apply a skew correction to the timestamp to allow them to be compared at the gross level. These are used in the latency metrics.
We also sample the CPU and memory usage on all VMs every 5~secs to plot the utilization metrics. Each experiment runs for $\sim 10$~mins of wallclock time.% that translates to about $7$~days of event data for the TAXI datasets at $1000\times$ scaling~\footnote{Application runtime = $\frac{7~days \times 24~hours \times 60~mins \times 60~secs}{1000\times~scaling}secs = 10.08~mins$}, $6$~hours of CITY dataset with $30\times$ scaling and $12~min$ of FIT and GRID dataset with $1\times$ scaling as mentioned in Table~\ref{tbl:datasets}.
% \ysnote{We pre-process these raw datasets to generate SenML}
% \ysnote{More details and tables required on the configuration of the benchmark properties, scheduling, VM setup, etc.}
% \ysnote{Have a 2D matrix with dataset and topo as rows and cols, and ``slots,VMs'' as cell values}
\subsection{Micro-benchmark Results}
Fig.~\ref{fig:storm:micro:bm} shows plots of the different metrics evaluated for the micro-benchmark tasks on Storm when running at their peak input rate supported on a single \texttt{D1} VM with one thread. The \emph{peak sustained throughput} per task is shown in Fig.~\ref{fig:storm:micro:peakthru} in \emph{log-scale}. We see that most tasks can support $3,000$~msg/sec or higher rate, going up to $68,000$~msg/sec for ANN, BLF, RGF, ACC, DAC and KAL. XML parsing is highly CPU bound and has a peak throughput of only $310~msg/sec$. SML parse supports much higher rate than XML with less CPU usage, indicating that it is a better fit for streaming IoT applications than the XML format. % We use json notation for SenML which is lightweight compared to XML parsing.
DTT and MLT uses WEKA library for model training and supports only $50$ and $70~msg/sec$ rate, CPU being the bottleneck. PLT uses the XChart~\cite{xchart:lib} Java charting library and supports only $25~msg/sec$ rate as it is CPU intensive around $70\%$ as shown in~\ref{fig:storm:micro:cpu} at peak rate.
The Azure operations are I/O bound on the Cloud service and slow due to the web service latency. ATR supports only $1~msg/min$, as the task has to scan the full table on Azure with, e.g., $753,382$ records for the Taxi dataset, to query over non-key attributes on single Azure table partition. Better input rates can be achieved by storing Azure table on multiple partitions with query attributes as partition or row-key. RZP supports $300~msg/sec$ while LZP supports $3,000~msg/sec$ -- RZP has to write the Zip file to a remote shared directory while LZP uses a local disk.
The inverse of the peak sustained throughput gives the \emph{mean latency}, and we do not explicitly plot it. However, it is interesting to examine the \emph{end-to-end latency}, calculated as the time taken between emitting a message from the source, having it pass through the benchmarked task, and arrive at the sink task. This is the effective time contributed to the total tuple latency by this task running within Storm, including framework overheads. We see that while the mean latencies should be in sub-milliseconds for the observed throughputs, the box plot for end-to-end latency (Fig.~\ref{fig:storm:micro:latency}) varies widely up to $2,600~ms$ for Q3, except ACC and INP task. This wide variability could be because of non-uniform task execution times due to which slow executions queue up incoming messages that suffer higher queuing time, such as for DTC and MLR that both use the WEKA library. Or tasks supporting a high input rate in the order of $10,000~msg/sec$, such as DAC and KAL, may be more sensitive to even small per-tuple overhead of the framework, say, caused by thread contention between the Storm system and worker threads, or queue synchronization.
The Azure tasks that have a lower throughput also have a higher end-to-end latency, but much of which is attributable directly to the task latency. ATR has a latency of $1~min$ due to scanning of the large table. ACC shows wide distribution of latency due to variability in the complexity of operation performed on it. Events associated with a single sensor ID are stored in a time-ordered queue until the threshold count is reached, upon which it extracts all the accumulated values and passes it downstream. MQS shows latency of $1,900~ms$ with no whiskers as the task logic just polls a local queue of messages being populated by the subscribed messages arriving from the broker.
%As observed for each of the log scaled value of peak input rate on Y-axis, CPU is the bottleneck (except for azure tasks) thus verifying the results for peak rates. Since Azure operations are having low CPU for the peak rates so we tried to find the variation of peak throughput with thread parallelism .Impact of multithreading on peak rate Plots ~\ref{fig:peakrate:multithreading} shows the importance of assigning optimal number of threads of different task logics. As seen in case of Float-Pi and XML tasks ,increasing the number of threads have reverse impact on performance in terms of peak throughput .\\
% 1. Thruput gives the avg latency for just the micro benchmark task.
% 2. End to end latency includes network and queuing time, in addition to the execution time. This value is high, relative to exec time for most tasks. Thus, we can see this as the overhead imposed by the DSPS. This could be due to thread contention, queueing delays, etc. We see that often, the latency time variation is higher for higher peak thruput, showing the senstivity of the framework to high rates due to thread context switching, etc. The box could also be wide if the latency per tuple is non-uniform. For e.g., weka seems to have this issue. Tuning the buffer sizes in storm may help improve this performance, but that is outside scope.
% 3. For azure, the e2e latency and avg latencies are similar, showing that the f/w overhead is less.
%The box plot for latencies (Fig.~\ref{fig:storm:micro:latency}) at the peak rate supported for 1 thread/parallelism hint on 1 core \texttt{D1} VM .Median latency values are high for all the Azure operations due to network requests for each message for fetching results. Also XML and Second order moment have higher latencies as CPU usage at peak rate is higher for them than in case of others. Wider box plots for DTC and SLR are due to specific implementation using WEKA-library .
\begin{figure}[t]
\centering
\subfloat[Peak Throughput]{% \emph{Note: ATR is able to support only $1~msg/min$.}]{
\includegraphics[width=0.40\textwidth]{plots/microBenchmark/peakRateBarplot_updated.pdf}
\label{fig:storm:micro:peakthru}
}\\
\subfloat[End-to-end latency]{
\includegraphics[width=0.40\textwidth]{plots/microBenchmark/Final-FullBoxPlot-Latency-updated.pdf}
\label{fig:storm:micro:latency}
}
\subfloat[Jitter]{
\includegraphics[width=0.40\textwidth]{plots/microBenchmark/FullBoxPlotJitter-updated.pdf}
\label{fig:storm:micro:jitter}
}\\
\subfloat[CPU\%]{
\includegraphics[width=0.40\textwidth]{plots/microBenchmark/FullBoxPlotCPU_updated.pdf}
\label{fig:storm:micro:cpu}
}
\subfloat[MEM\%]{
\includegraphics[width=0.40\textwidth]{plots/microBenchmark/FullBoxPlotMEM_updated.pdf}
\label{fig:storm:micro:mem}
}
\caption{Performance of micro-benchmark tasks for integer input stream at peak rate.}
\label{fig:storm:micro:bm}
\end{figure}
The box-plot for \emph{jitter} (Fig.~\ref{fig:storm:micro:jitter}) shows values close to zero in all cases. This indicates the long-term stability of Storm in processing the tasks at peak rate, without unsustainable queuing of input messages. The wider whiskers indicate the occasional mismatch between the expected and observed output rates. ATR again has a high range for the whiskers as its rate is very low at $1~msg/min$; thus even minor variation in rate shows high jitter values.% input rate is very high as it can be observed from the respective peak rate bar plot for them.In case the input rate is highly variable the jitter box-plot will justify if variation output rate is in correspondence to input rate or not.
The box plots for CPU utilization (Fig.~\ref{fig:storm:micro:cpu}) shows the single-core VM effectively used at $70\%$ or above in all cases except for the SML, MQS and Azure tasks that are I/O bound. MQS is bounded by the number of threads as single thread is busy in polling the message queue which is not CPU intensive. SML is having low CPU of $\approx{30\%}$, the reason being that as we are using a JSON representation for SenML which is less CPU intensive as compared to XML.
The memory utilization (Fig.~\ref{fig:storm:micro:mem}) appears to be higher for tasks that support a high throughput, potentially indicating the memory consumed by messages waiting in queue rather than consumed by the task logic itself. MQS shows a high memory usage ($\approx{50\%}$) even for a low rate due to buffering of incoming messages from the broker in a queue that is asynchronously being polled.
Similarly, memory for DTT and MLT is $\approx{45\%}$ because a batch of nearly thousand rows is stored in memory for model training triggered by every incoming input message.
%Memory usages in case of BLF,DAC,KAL tasks are high as we need to keep states in memory across the messages and it can be controlled by the parameter size to the logic.In case of MLR memory is high due to slow garbage collection for high input rate .
% \ysnote{Except for the azure table and blob, other rates and CPU are high. We also know that azure ops can benefit from multi threading. It may be useful to show the multi threaded plots for just these 3: X Axis no. of threads, Y axis peak rate. As a contrast, we can also show that XML and Pi, though they have peaks of 100's per sec (but lower than the others except Az), do not benefit from multi-threading. Generalize in future work.}
% \ysnote{Both weka bolts (LR and DTree) have a wider latency box. I can imagine DTree has different tree heights and may finish faster for some messages but LR should be deterministic? Something weka specific? To verify.}
% \ysnote{Pi has a wider box, wonder why. It should be very deterministic. To verify. Wonder what the jitter will look like for these.}
% \ysnote{Mem is high for distinct cont and Kalman. That is something we control with the size param? Mem is medium for blob and parse. That may be due to large objects and slower garbage collection. But strangely, SLR also has high mem. Not clear why that should happen. Maybe due to high rate and hence high mem due to slow GC? To verify. }
\subsection{Application Results}
\begin{figure}[t]
\centering
% \subfloat[Latency for STATS]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-latency-boxplot-STATS.pdf}
% \label{fig:storm:stats:latency}
% }
% \subfloat[Latency for PRED]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-latency-boxplot-PRED.pdf}
% \label{fig:storm:pred:latency}
% }
\subfloat[ETL]{
\includegraphics[width=0.17\textwidth]{plots/apps/latency-boxplot-etl.pdf}
\label{fig:storm:etl:latency}
}
\subfloat[STATS]{
\includegraphics[width=0.17\textwidth]{plots/apps/stats-latency-boxplot.pdf}
\label{fig:storm:stats:latency}
}
\subfloat[TRAIN]{
\includegraphics[width=0.17\textwidth]{plots/apps/train-latency-boxplot.pdf}
\label{fig:storm:train:latency}
}
\subfloat[PRED]{
\includegraphics[width=0.17\textwidth]{plots/apps/pred-latency-boxplot.pdf}
\label{fig:storm:pred:latency}
}
\caption{End-to-end latency plots for application benchmarks on workloads. ETL and PRED are in millisec and STATS and TRAIN are in sec. *TRAIN and PRED are not run for GRID workload as it has only the target field, and no additional field to predict upon.}
% \subfloat[Latency for ETL- Grid , Fit]{
% \includegraphics[width=0.35\textwidth]{plots/apps/app-latency-boxplot-ETL-grid-fit.pdf}
% \label{fig:storm:etl:sys:cpu}
% }
%\subfloat[Latency for STATS]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-latency-boxplot-updatedSTATS.pdf}
% \label{fig:storm:stats:city:latency}
% }
\end{figure}
\begin{figure}[t]
\centering
\subfloat[ETL]{
\includegraphics[width=0.25\textwidth]{plots/apps/app-jitter-etl-boxplot.pdf}
\label{fig:storm:app:etl:jitter}
}
\subfloat[STATS]{
\includegraphics[width=0.25\textwidth]{plots/apps/app-jitter-stats-boxplot.pdf}
\label{fig:storm:app:stats:jitter}
}
\subfloat[TRAIN]{
\includegraphics[width=0.25\textwidth]{plots/apps/app-jitter-train-boxplot.pdf}
\label{fig:storm:app:train:jitter}
}
\subfloat[PRED]{
\includegraphics[width=0.25\textwidth]{plots/apps/app-jitter-pred-boxplot.pdf}
\label{fig:storm:app:pred:jitter}
}
\caption{Jitter plots for application benchmarks on workloads. *TRAIN and PRED are not run for GRID workload as it has only the target field, and no additional field to predict upon.}
%\label{fig:apps:jitter}
% \subfloat[STATS:CITY]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-STATS-sys-cpu-mem.pdf}
% \label{fig:storm:stats:city:cpu}
% }
% \subfloat[STATS:TAXI]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-STATS-taxi-cpu-mem.pdf}
% \label{fig:storm:stats:taxi:cpu}
% }
% \subfloat[PRED:CITY]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-PRED-sys-cpu-mem.pdf}
% \label{fig:storm:pred:city:cpu}
% }
% \\
% \subfloat[PRED:TAXI]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-PRED-taxi-cpu-mem.pdf}
% \label{fig:storm:pred:taxi:cpu}
% }
\label{fig:storm:app:jitter}
\end{figure}
\begin{figure}[t]
\subfloat[CITY]{
\includegraphics[width=0.235\textwidth]{plots/apps/etl-sys-cpu.pdf}
\label{fig:storm:etl:city:cpu}
}
\subfloat[FIT]{
\includegraphics[width=0.235\textwidth]{plots/apps/etl-fit-cpu.pdf}
\label{fig:storm:etl:fit:cpu}
}
\subfloat[GRID]{
\includegraphics[width=0.235\textwidth]{plots/apps/etl-grid-cpu.pdf}
\label{fig:storm:etl:grid:cpu}
}
\subfloat[TAXI]{
\includegraphics[width=0.235\textwidth]{plots/apps/etl-taxi-cpu.pdf}
\label{fig:storm:etl:taxi:cpu}
}
\caption{CPU and Memory utilization plots for \emph{ETL} application benchmark on all workloads.}
\label{fig:storm:etl:cpu}
\end{figure}
\begin{figure}[t]
\subfloat[CITY]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-stats-sys-cpu-mem.pdf}
\label{fig:storm:stats:city:cpu}
}
\subfloat[FIT]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-stats-fit-cpu-mem.pdf}
\label{fig:storm:stats:fit:cpu}
}
\subfloat[GRID]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-stats-grid-cpu-mem.pdf}
\label{fig:storm:stats:grid:cpu}
}
\subfloat[TAXI]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-stats-taxi-cpu-mem.pdf}
\label{fig:storm:stats:taxi:cpu}
}
\caption{CPU and Memory utilization plots for \emph{STATS} application benchmarks on all workloads.}
\label{fig:storm:stats:cpu}
\end{figure}
\begin{figure}[t]
\centering
\subfloat[CITY]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-train-city-cpu-mem.pdf}
\label{fig:storm:train:city:cpu}
}
\subfloat[FIT]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-train-fit-cpu-mem.pdf}
\label{fig:storm:train:fit:cpu}
}
\subfloat[TAXI]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-train-taxi-cpu-mem.pdf}
\label{fig:storm:train:taxi:cpu}
}
\caption{CPU and Memory utilization plots for \emph{TRAIN} application benchmarks three workloads, CITY FIT and TAXI. GRID workload is not used as it has only the target field, and no additional field to predict upon.}% on all workloads.~\emph{NOTE: GRID dataset have only one observation field and MLR, DTC requires multiple fields for training.} }
\label{fig:storm:train:cpu}
\end{figure}
\begin{figure}[t]
\centering
\subfloat[CITY]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-pred-city-cpu-mem.pdf}
\label{fig:storm:pred:city:cpu}
}
\subfloat[FIT]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-pred-fit-cpu-mem.pdf}
\label{fig:storm:pred:fit:cpu}
}
\subfloat[TAXI]{
\includegraphics[width=0.235\textwidth]{plots/apps/app-pred-taxi-cpu-mem.pdf}
\label{fig:storm:pred:taxi:cpu}
}
\caption{CPU and Memory utilization plots for \emph{PRED} application benchmarks on three workloads, CITY FIT and TAXI. GRID workload is not used as it has only the target field, and no additional field to predict upon.}
\label{fig:storm:pred:cpu}
\end{figure}
% \subfloat[STATS:CITY]{
% \includegraphics[width=0.235\textwidth]{plots/apps/app-updatedStats-sys-cpu-mem.pdf}
% \label{fig:storm:stats:city:cpu}
% }
The ETL and STATS application benchmarks are run for the CITY, FIT, GRID and TAXI stream workloads. TRAIN and PRED are run for CITY, FIT and TAXI datasets and not for GRID because it has only one observation field, and prediction tasks such Decision Tree and Multivariate Linear Classifier uses a combination of fields to predict or classify an observational field. The input rate is as per scaling discussed in table ~\ref{tbl:datasets} for each dataset.
The \emph{end-to-end latencies} of the applications depend on the sum of the end-to-end latencies of each task in the critical path of the dataflow.
For the ETL application, latency values in Fig.~\ref{fig:storm:etl:latency} remain the same $30~millisec$ for CITY, FIT and TAXI datasets. GRID has a higher variation in latency than others because of its normal distribution of messages over timestamp. The median latency for all the datasets are nearly comparable, with GRID having median latency $50~millisec$ and CITY, TAXI and FIT around $30~millisec$.
The STATS dataflow has latency values in the range of $10-40~millisec$, shown in Fig.~\ref{fig:storm:stats:latency}, which is higher than ETL and PRED. This is mainly due to the GroupViZ meta-task which batches messages, forming a time-series for plotting, and then accumulating the plots to create a zipped file. Also, its median latency values are highly variable depending on the dataset. The reason is that the accumulation and plotting are done separately for every distinct sensor ID until a fixed count is reached, and hence the latency for the meta-task depends on the content of input messages received.
The TRAIN dataflow's timer source task simulates the model training trigger every $2~hours$ of original time for CITY, every day for TAXI, and every minute for the FIT dataset. This translates to a benchmark time period of $2-5$~mins between two source events. % with temporal scaling for $\approx{15~minutes}$ experiment.
The latency values for TRAIN are understandably higher than other applications since it is a batch processing dataflow encoded as a streaming dataflow. The key reason is the Azure Table Range task that scans the full table to fetch rows that were inserted since the last training time. Also, the latency for the CITY dataset in Fig.~\ref{fig:storm:train:latency} is larger at $300~sec$ than FIT and TAXI datasets that are at $50~sec$ due to the difference in the table sizes. CITY has $3,629,428$ rows in its table while TAXI has $753,382$ rows inserted. %entries in their corresponding Azure tables used in the experiment. %Although Table size is small for FIT datset than TAXI but number of fields for training are more for FIT than TAXI this can be reason why latency value for FIT is not less than TAXI.
The PRED topology's latencies (Fig.~\ref{fig:storm:pred:latency}) also remain close together at $20~millisec$ for all the datasets. The large range of whiskers for all datasets in PRED is due to DTC and MLR tasks, which exhibit significant variations in their runtimes even for the micro-benchmarks.
The \emph{jitter} is also close to zero in all cases (Fig.~\ref{fig:storm:app:jitter}), indicating a sustainable performance for the application benchmarks. The whiskers for STATS are not visible as the total number of messages at the sink tasks are comparatively fewer than the input messages since the % due to plotting of multiple messages and zipping them at
GroupViZ task accumulates many of the inputs in singleton outputs per sensor ID.
Similarly, the whiskers for TRAIN are larger as few messages are emitted from source (max $10~msg/sec$ for FIT) in total, and thus most of the time variation is observed between source and sink rate.
The number of \emph{cores and VMs} required for the same application varies with the workload used (Tbl.~\ref{table:slots}). This is due to the difference in input rate that is processed by tasks for the respective workload, thus requiring different number of cores per task. We also see that the resource allocation strategy is generally liberal, and resources are under-utilized.
The CPU utilization for STATS is higher at $20-80\%$ than other applications (Fig.~\ref{fig:storm:stats:cpu}). This is due to AVG, DAC and GroupViZ tasks requiring higher CPU\%, matches with the CPU\% required for microbenchmarks. Also memory usage is higher for STATS in comparison to others due to GroupViZ task accumulating the messages %till a fixed count of messages for same sensorId/meterId are obtained
and plots in memory before zipping (Fig.~\ref{fig:storm:stats:cpu}).
The CPU utilisation for TRAIN is fairly small due to the low message rate, and the memory usage is comparatively high at $20\%$ as the large batch of table rows is stored in memory for model training.
The CPU utilization for the FIT workload is the least for all the application benchmarks due to the fact that it has the least rate at $500~msg/sec$, and we have assigned exclusive an core to each of its tasks. TAXI has a low CPU usage, mostly at a $5\%$ median, with a wide box (Figs.~\ref{fig:storm:etl:taxi:cpu},~\ref{fig:storm:pred:taxi:cpu}~\ref{fig:storm:stats:taxi:cpu}~\ref{fig:storm:train:taxi:cpu}) -- this is due to its bi-modal distribution with low input rates at nights, with lower utilization, and high in the day with higher utilization.
In general, we see that such a resource under-utilization strongly motivates the need for robust resource allocation strategies for IoT applications on DSPS.
%\sout{The STATS and PRED application benchmarks are run for the CITY and TAXI workloads at $1000\times$ their native rates, and the performance plots shown in Fig.~\ref{fig:apps}. The end-to-end latencies of the applications depend on the sum of the end-to-end latencies of each task in the critical path of the dataflow. The peak rates supported by the tasks in STATS is much higher than the input rates of CITY and TAXI. So the latency box plot for STATS is tightly bound (Fig.~\ref{fig:storm:stats:latency}) and its median much lower at $20$~ms compared to the end-to-end latency of the tasks at their peak rates. The jitter is also close to zero in all cases. So Storm can comfortably support STATS for CITY and TAXI on 7 and 5 VMs, respectively. The distribution of VM CPU utilization is also modest for STATS. CITY has a $35\%$ median with a narrow box (Fig.~\ref{fig:storm:stats:city:cpu}), while TAXI has a low $5\%$ median with a wide box (Fig.~\ref{fig:storm:stats:taxi:cpu}) -- this is due to its bi-modal distribution with low input rates at nights, with lower utilization, and high in the day with higher utilization.}
%\sout{For the PRED application, we see that the latency box plot is much wider, and the median end-to-end latency is between $500-700$~ms for CITY and TAXI (Fig.~\ref{fig:storm:pred:latency}). This reflects the variability in task execution times for the WEKA tasks, DTC and MLR, which was observed in the micro-benchmarks too. The Azure blob upload also adds to the absolute increase in the end-to-end time. The jitter however remains close to zero, indicating sustainable performance. The CPU utilization is also higher for PRED, reflecting the more complex task logic present in this application relative to STATS.}
% \ysnote{I'm not sure if the resource allocation is right. PRED seems to have a higher latency and that could be because of the fewer resources allocated to it relative to STATS. Something to think about for the next paper}
% \asnote{Box for Latency is wider in prediction due to Decision tree and MLR (used in DAG-app) as seen in microbenchmarks too. Jitter is more whiskers in case rate is high and latency is also low as in STATS-SYS.PREDiction apps have high latency so variation is not seen much in the jitter plots . CPU is wider in case of TAXI due to variation in input rate within 24 hours , during daytime peak rate is seen giving high value of CPU and in night time CPU is low as rate is low . }
%As an empirical validation of our proposed benchmarks, we implement the proposed micro-patterns and IoT application benchmarks on \emph{Apache Storm} and \emph{Apache Spark}, two popular DSPS in the open source community. We then run these for the three stream workloads and evaluate them based on the metrics we have defined.
%
%
%\subsection{Benchmark Implementation}
%The \emph{8 micro-patterns} are basic building block patterns for DSPS and naturally fit with the patterns provided by both Storm and Spark. For Storm, we use their Java API for composing the dataflow graphs explicitly for the micro-patterns and all tasks extend from the \texttt{BaseRichBolt}, with the logic present in the \texttt{emit} method that is called once per input message by the DSPS. Only the AV pattern has a stateful aggregator task while others are stateless. The routing is defined by a \texttt{Topology} class as an edge list between the bolts, and \emph{Groupings} to distinguish duplicate and hash semantics.
%
%Spark, in the other hand, provides a higher level functional construct, also in Java. Here, we do not explicit define the wiring and routing between the tasks but instead define a function logic that is called on each \emph{(discretized) stream}. These functions extend from one of several built in functions, such as \texttt{map}, \texttt{flatMap} and \texttt{reduceByKey}, that also determine the routing semantics and selectivity. The micro-patterns themselves, given their compact size, are mapped to this functional model. The AV task uses a built-in window function that is exposed for associative functions like sum and count, which we use to achieve average. \ysnote{check}.
%
%The logic for each task in the \emph{FCAST application} was implemented in Java, and similar logic used for both Storm and Spark. The application dataflow composition itself was straight forward in Storm with the proposed pipeline naturally fitting its graph model. However, with Spark, the dataflow composition using its functional model was non-trivial. There were three key reasons for this. One, Spark does not allow \emph{state} to be persisted easily, instead treating state as a new stream with a function to accumulated stateful accumulation. This makes it cumbersome to perform stateful tasks like interpolation and regression. This is unlike Storm where the state is intuitively present as a field in the bolt. Secondly, all reduce or accumulation operations in Spark require to be \emph{associative}. This again is a challenge for the above two statistics tasks which are not so. Lastly, the window operation within Spark can only be applied over a time window and not a \emph{count window}. This further means that it is not easy to, say, use the last 10 observations to perform a linear regression, but instead time windows have to be accumulated and their tuple count brought to the required value. As such, the FCAST dataflow that was composed in Spark was an approximation of the intended application.
%
%%One distinction in composing the application
%We also implemented an \emph{event generator} harness in Java that, when provided a CSV file with time-series of events, replays these events at the same time intervals as the original events. A scaling factor can be provided to replay the events faster or slower than their native speeds. We use $100 \times$ throughput scaling for CITY and TAXI, and native rate ($10 \times$) for PLUG. This simulator library was called from within the source task logic of both Spark and Storm.
%
%\subsection{Experimental Setup}
%\asnote{To be Modified As per rigel}
%We use Apache Storm $0.9.4$ (released on 25 Mar, 2015) and Apache Spark $1.3.0$ (13 Mar, 2015) for our DSPS evaluation, both running on OpenJDK 1.7. The DSPS are each installed on $8$ Amazon EC2 \texttt{m3.medium} VMs (One Intel Xeon E5-2670 v2 core $2.5$~GHz, $7.5$~GiB RAM, $8$~GiB SSD disks) in their Singapore data center. Of these $7$ VMs are used as the worker nodes running the application tasks, with one worker slot per machine, and $1$ VM is used as the master for coordination.
%
%We log the timestamp and id for each message emitted by the event generator in a RAM disk, causally trace the id to downstream messages, and similarly log the id and timestamp of output messages after the sink. These help calculate the throughput and jitter metrics. Given the distributed setup, we also calculate the time skew between the machines, both before and after the experiments, and apply a skew correction to the timestamp to allow them to be compared at the gross level. These are used in the latency metrics. We also sample the resource usage on all machines every 5~secs and log them in RAM disk, and these are used for the CPU utilization metrics.
%
%Each of the benchmark experiments were run for $12$~mins of wall-clock time and fed the input stream for that duration. For CITY and TAXI, that run at $100 \times$ time scaling, these translate to about $20$~hours of event data, while PLUG \asnote{Update needed}runs at native rate.
%
%\subsection{Micro-pattern Results}
%
%%\subsubsection{Throughput}
%Figs.~\ref{fig:sys:100x:bm:thruput},\ref{fig:plug:10x:bm:thruput} and \ref{fig:taxi:100x:bm:thruput} show the observed input and output throughputs for the $8$ micro-patterns when run on Storm and Spark for the CITY, PLUG and TAXI datasets. Related to it, we have the box plot for the jitter for the each of these experiments shown in Fig.~\ref{fig:100x:bm:turbu}. We can make several observations from these for each dataset.
%
%CITY data's input data rate is more or less steady at $600$~msg/sec, reflecting the narrow rate distribution that it showed. The output throughput for Spark, however, shows a more spiky trend for most benchmarks. This is a manifestation of their discretized stream model (DStream) where stream are processed in windows of $1$~sec width, and this causes sharp jumps when the output rate catches up in batches with the input. We also see a visible jitter value that while centered at $0$ does show outliers of $\pm5\%$. Storm on the other hand has a steady output rate with negligible jitter. For both Spark and Storm, the output rates match the selectivity of the pattern, e.g. with the rates overlapping for ID, PJ, and SQ having $1$:$1$ selectivity, while FM shows thrice the output rate, based on $1$:$3$ selectivity.
%% - Spark output spiky one in a while due to it's dicretized model. Flat map and data gen particularly so since they have higher selctivity.
%% - output rates match with expected selectivity
%\asnote{Outliers in jitter plot is due to extra latency in start of experiment.}
%\ysnote{SM outrate for spark is lower than storm?}\asnote{Thruput plot for SM shows continuous message count diff also due to it Jitter plot go down NEEDS CODE DEBUGGING and EXPERI REDO}
%\ysnote{AV outrate for spark is tiny in all cases?}\asnote{number of messages in batches are highly variable so total count of agg. values i.e. output messages are diff then expected for finding selectivity.\\
% Can discuss i have some Extra plots .\\Redo experiment for SM for bothe CITY and TAXI spark . }
%
%
%% Jitter is negligible for CITY storm. Q1-Q3 are on 0 line.
%% There is visible jitter for spark. Q1-Q3 are on 0 line but more outliers.
%
%
%PLUG shows a more variable input rate, with the output rate for both Storm and Spark not in perfect harmony with the input rate. This is visible through a wider jitter for the $25$--$75$ percentiles. This variability is magnified even more for patterns where the selectivity is higher, such as DG where the mean output rates is greater than $1,00,000$~msg/sec.
%% For longer chains like SQ, Storm appears to have wider outliers in the jitter too.
%% PLUG 1x thruput MP
%% - more spiky input and output rates for both storm, spark.
%\asnote{High spikes in spark thruput are due to variability in MSG count in time based batch size. Spark is able to average out the spikes in input rates to get smoother output rates. Storm seems to keep up in real-time.}
%
%Lastly, TAXI shows the bi-modal event rate distribution with the two peaks in event rates. Both Storm and Spark are able to keep up the output rate as the input rate varies due to a smaller absolute rate of $<400$~msg/sec.
%% - Both appear to be able to keep up due to smaller relative rate.
%\ysnote{- AV output small for spark?}\asnote{since input batch size is not fixed (follows sinusoidal pattern) so agg over the particular batch in spark(for every 30 to 300 msgs variable) will not give that many values as in STORM (for every 60 msgs fixed),setting higher value for agg in storm may help.}
%\ysnote{Tight jitter for storm but QC is outlier with faster than estimated output rate. Why? }\asnote{Continously less messages are received at sink for QC in both spark and storm - Needs re-run and debugging -}
%% Wider jitter Q1-Q3 for spark but fewer outliers.
%% =====================
%
%As such, while there is more jitter in Spark due to the discretization, both Storm and Spark appear to keep up with the event rates for all the micro-patterns and workloads evaluated.
%
%
%\asnote{
%e-Plug spark scatter plot shows Latency is increasing with time - needs discussion.\\
%Value range of Y-ticks in spark Latency have been updated /}
%\end{figure*}
%\asnote{In general, Latency in case of spark is always more than the storm ,spark being batch processing model.
%For CITY ,Storm being tuple processing have negligible latency value in most cases , outliers for DG are due to extra latency in first 30 sec of topology .While in SPARK for AV extra latency is due to violation of assumption that "batch processing time should be less than the batch interval " which needs extra tuning like changing batch interval thus changing window duration or including more resources .\\
%For PLUG, Latency for DG and SM in storm is large due to less resources as shown in FIG-5 .
%while for spark latency is large due to Large Processing Latency(due to large batch) than Batch duration.Link for explanation- http://spark.apache.org/docs/latest/streaming-programming-guide.html\#setting-the-right-batch-interval \\
%For TAXI , Outliers in case of Storm PJ and FM are due to latency by first 30 sec of messages as in Figure -6,7 while in spark Outliers in Identity need Expr. REDO .Latency for AV is large due to high processing latency .
%}
%\asnote{For the same data rate and resources allocated ,CPU utilization for SPARK is less than the storm in all cases due to grouping of messages in batches making low aggregated message handling cost .\\
%For CITY storm have high CPU utilisation for DG due to high message rate of 60,00 msg/sec produced within the topology while in Spark SQ and AV and FM have high CPU utilisation due to large computations . NEED MORE explanation for CPU usage .}
%\iffalse
%\begin{figure*}[th]
%\centering%~%
% \subfloat[CITY $100\times$, \emph{Storm}]{%
% \includegraphics[width=0.33\textwidth]{plots/storm/sys-100x-app-thruput.pdf}%
% \label{fig:storm:sys:100x:app:thruput}%
% }
% \subfloat[PLUG $10\times$, \emph{Storm}]{%
% \includegraphics[width=0.33\textwidth]{plots/storm/plug-10x-app-thruput.pdf}%
% \label{fig:storm:plug:10x:app:thruput}%
% }
% \subfloat[TAXI $100\times$, \emph{Storm}]{%
% \includegraphics[width=0.33\textwidth]{plots/storm/taxi-100x-app-thruput.pdf}%
% \label{fig:storm:taxi:100x:app:thruput}%
% }\\
% \subfloat[CITY $100\times$, \emph{Spark}]{%
% \includegraphics[width=0.33\textwidth]{plots/spark/sys-100x-app-thruput.pdf}%
% \label{fig:spark:sys:100x:app:thruput}%
% }
% \subfloat[PLUG $10\times$, \emph{Spark}]{%
% \includegraphics[width=0.33\textwidth]{plots/spark/plug-10x-app-thruput.pdf}%
% \label{fig:spark:plug:10x:app:thruput}%
% }
% \subfloat[TAXI $100\times$, \emph{Spark}]{%
% \includegraphics[width=0.33\textwidth]{plots/spark/taxi-100x-app-thruput.pdf}%
% \label{fig:spark:taxi:100x:app:thruput}%
% }
%
%\caption{Throughput for application benchmarks for three datasets. Top row (\ref{fig:storm:sys:100x:app:thruput}--\ref{fig:storm:taxi:100x:app:thruput}) is \textbf{Storm}, bottom row (\ref{fig:spark:sys:100x:app:thruput}--\ref{fig:spark:sys:100x:app:thruput}) is \textbf{Spark}.}
%\label{fig:100x:app:thruput}
%\end{figure*}
%
%\fi
%\ysnote{Cloud sustained 24hrs test, if we can}
%Links to the datasets \\
%\texttt{http://www.ucd.ie/issda/data/commissionforenergyregulationcer/}\\
%\texttt{https://archive.ics.uci.edu/ml/datasets/MHEALTH+Dataset}
\section{Conclusion}% \note{2pgs}}
\label{sec:conclusion}
%Such benchmarks also show us how to eke out more performance from DSPS, e.g. we can pump up Az bolts since we see low thruput, low cpu.
In this paper, we have proposed \emph{RIoTBench}, a novel benchmark suite for evaluating distributed stream processing systems for Internet of Things applications, which encompasses several emerging domains. Fast data platforms like DSPS are integral for the rapid decision making needs of IoT applications. Our proposed micro and application benchmarks help evaluate their efficacy using common tasks found in IoT domains, as well as fully-functional dataflows for pre-processing, statistical summarization and predictive analytics. These applications naturally fit into the OODA interaction model found in many IoT domains. These benchmarks are combined with four real-world data streams from Smart Grid, Smart Transportation, Urban Sensing and personal fitness domains of IoT, that are further spatially and temporally scaled to recreate the stream profiles of contemporary IoT deployments. The proposed benchmark has been validated for the highly-popular Apache Storm DSPS, and the performance metrics reported.
% \sout{As future work, we propose to add further depth to some of the IoT task categories such as parsing and analytics, and also add two further applications on archiving real-time data and detecting online patterns. We also plan to include more data stream workloads having different temporal distributions and from other IoT domains, with a possible generalization of the distributions to allow for synthetic data generation.}
As future work, we would like to add event pattern detection and notifications as tasks to our benchmark suite to complete the representative categories. The benchmark can also be used to evaluate other popular DSPS such as Apache Spark Streaming and Flink. Incidentally, these tasks and applications we have provided have real and accurate business logic. Thus, they form a valuable library of tasks that can be used in both generic and IoT streaming applications. We are currently in the process of integrating customized versions of these benchmark applications into the IISc Smart Campus IoT project for smart water and power management~\footnote{IISc Smart Campus Project, \url{http://smartx.cds.iisc.ac.in}}.
%
\section*{Acknowledgments}
%
We acknowledge detailed inputs provided by Tarun Sharma of NVIDIA Corp. and formerly from IISc in preparing this paper. The experiments on Microsoft Azure were supported through a grant from Azure for Research. We thank the reviewers of the Technology Conference on Performance Evaluation \& Benchmarking (TPCTC), 2016, for their valuable comments to improve the benchmark suite.
%\note{\section{Discarded}}
% \subsection*{Energy Dataset details }
%\label{sec:datasets }
%There are total 6 files present in the datasets. Here are few key points related to this dataset. \\
%\textbf{Format of each data file - 3 columns corresponding to these 3 fields :} \\
%\hspace*{10pt} Meter ID \\
%\hspace*{10pt} Five digit code: \\
%\hspace*{20pt} Day code: digits 1-3 (day 1 = 1st January 2009) \\
%\hspace*{20pt} Time code: digits 4-5 (1-48 for each 30 minutes with 1= 00:00:00 – 00:29:59) \\
%\hspace*{20pt} Electricity consumed during 30 minute interval (in kWh) \\
%\textbf{Number of unique metres in each file} \\
%\hspace*{10pt}file1- 1000 \\
%\hspace*{10pt}file2- 998 \\
%\hspace*{10pt}file3- 1000 \\
%\hspace*{10pt}file4- 1000 \\
%\hspace*{10pt}file5- 1000 \\
%\hspace*{10pt}file6- 1443\\
%Total unique meter ids present - 6435 \\
%Total Unique meter id present after folding = $3,087,011$\\
% trigger a \newpage just before the given reference
% number - used to balance the columns on the last page
% adjust value as needed - may need to be readjusted if
% the document is modified later
%\IEEEtriggeratref{8}
% The "triggered" command can be changed if desired:
%\IEEEtriggercmd{\enlargethispage{-5in}}
% references section
\bibliographystyle{plain}
% argument is your BibTeX string definitions and bibliography database(s)
\footnotesize{
\bibliography{main}
}
% references section
%\bibliographystyle{splncs03}
% argument is your BibTeX string definitions and bibliography database(s)
%\footnotesize{
%\clearpage
%\section*{References}
%\bibliography{main}
%%}
\clearpage
\appendix
\section{Configurations used in Application Dataflows}
\begin{table}[h]
\centering
\footnotesize
\caption{Attributes Used in Tasks of the ETL Application}
%\ysnote{Use same table style as Tbl 2}\ysnote{Use $p\{\}$ for attribute columns}
\label{tab:etl:field}
\resizebox{\linewidth}{!}{%
\begin{tabular}{c|p{2.95cm}p{2.95cm}p{2.95cm}p{3.95cm}}
%\begin{tabular}{l l l l l}
\hline
\textbf{Task} &~\textbf{CITY} &~\textbf{FIT} &~\textbf{GRID} &~\textbf{TAXI} \\ \hline \hline
\textbf{ANN}$^*$ & location, sensor type & age, gender & tariff allocation, sme allocation, stimulus allocation & driver name, city, company \\\hline
\textbf{BLF} & source & N/A~$^\dagger$ & meterid & taxi identifier \\\hline
\textbf{INP} & temperature, humidity, light, dust, airquality raw & acceleration chest, arm, ankle X/Y/Z, ecg 1 & energyConsumed & N/A~$^\ddagger$ \\ \hline
\textbf{RGF} & temperature, humidity, light, dust, airquality raw & acceleration chest, arm, ankle X/Y/Z, ecg 1 & energyConsumed & trip time in sec, trip distance, fare amount, surcharge, mta tax, tip amount, tolls amount, total amount \\ \hline
\hline
\multicolumn{5}{l}{\emph{$^*$ Annotation attributes that are added to the dataset by ANN, either provided with the dataset or synthetically}}\\
\multicolumn{5}{l}{\emph{$^\dagger$ No fields were used for the particular task with the dataset because the number of unique subjects is very less (10) for FIT thus not requires BLF.}}\\
\multicolumn{5}{l}{\emph{$^\ddagger$ Interpolation of values over different Taxi trips is not meaningful.}}\\
\end{tabular}}
\end{table}
\begin{table}[h]
\centering
\footnotesize
%\caption{Dataset Fields Used For Stats Topology,~\emph{N/A means that no fields were used for the paricular task with the dataset.}}
\caption{Attributes Used in Tasks of the STATS Application}
\label{tab:stats:field}
\resizebox{\linewidth}{!}{%
\begin{tabular}{c|p{2.95cm}p{2.95cm}p{2.95cm}p{3.95cm}}
\hline
\textbf{Task} &~\textbf{CITY} &~\textbf{FIT} &~\textbf{GRID} &~\textbf{TAXI} \\ \hline \hline
\textbf{AVG} &temperature, humidity, light, dust, airquality raw & acceleration chest, arm, ankle X/Y/Z, ecg 1/2 & energyConsumed & trip time in sec, trip distance, fare amount, surcharge, mta tax, tip amount, tolls amount, total amount\\\hline
\textbf{DAC} & temperature & ecg 1 & energyConsumed & N/A~$^\dagger$ \\ \hline
\textbf{SLR}&temperature, humidity, light, dust, airquality raw & acceleration chest, arm, ankle X/Y/Z, ecg 1/2 & energyConsumed & trip time in sec, trip distance, fare amount, surcharge, mta tax, tip amount, tolls amount, total amount \\ \hline
\hline
\multicolumn{5}{l}{\emph{$^\dagger$ No fields were used for the particular task with the dataset because DAC over individual Taxi trips is not meaningful.}}
\end{tabular}}
\end{table}
\begin{table}[h]
\centering
\footnotesize
% \caption{Dataset Fields Used For Prediction Topology, \emph{(C1/C2/C3/C4)$^*$ are the classes predicted by DTC task.}}
\caption{Attributes Used in Tasks of the PRED Application}
\label{tab:pred:field}
\resizebox{\linewidth}{!}{%
\begin{tabular}{c|p{2.95cm}p{2.95cm}p{2.95cm}p{3.95cm}}
\hline
\textbf{Task} &~\textbf{CITY} &~\textbf{FIT} &~\textbf{GRID} &~\textbf{TAXI} \\ \hline \hline
\textbf{AVG} & airquality raw & ecg 1 & N/A~$^\dagger$ & fare amount \\ \hline
\textbf{DTC} & $\mathcal{F}$(temperature, humidity, light, dust, airquality raw) $\rightarrow \{C1 \mid C2 \mid C3 \mid C4 \}^*$ & $\mathcal{F}$(acceleration chest, arm, ankle X/Y/Z, ecg 1) $\rightarrow \{C1 \mid C2 \mid C3 \mid C4 \}^*$ & N/A~$^\dagger$ & $\mathcal{F}$(trip time in sec, trip distance, fare amount) $\rightarrow \{C1 \mid C2 \mid C3 \mid C4 \}^*$ \\ \hline
\textbf{MLR} & $\mathcal{F}$(temperature, humidity, light) $\rightarrow$ airquality raw & (acceleration chest, arm, ankle X/Y/Z) $\rightarrow$ ecg 1 & N/A~$^\dagger$ & $\mathcal{F}$(trip time in secs, trip distance)$\rightarrow$ fare amount \\ \hline
\hline
\multicolumn{5}{l}{\emph{$^*$ Classses used for prediction by DTC task}}\\
\multicolumn{5}{l}{\emph{$^\dagger$ No fields were used for the particular task with the dataset as GRID is univariate whereas DTC and MLR tasks require multiple fields.}}
\end{tabular}
}
\end{table}
\begin{table}[h]
\centering
\footnotesize
%\caption{Dataset Fields Used For Train Topology\ysnote{Fix table similar to ~\ref{tab:etl:field}}}
\caption{Attributes Used in Tasks of the TRAIN Application}
\label{tab:train:field}
\resizebox{\linewidth}{!}{%
\begin{tabular}{c|p{2.95cm}p{2.95cm}p{2.95cm}p{3.95cm}}
\hline
\textbf{Task} &~\textbf{CITY} &~\textbf{FIT} &~\textbf{GRID} &~\textbf{TAXI} \\ \hline \hline
\textbf{DTT} & temperature, humidity, light, dust, airquality raw & acceleration chest, arm, ankle X/Y/Z, ecg 1 & N/A~$^\dagger$ & trip time in sec, trip distance, fare amount \\ \hline
\textbf{MLT} & temperature, humidity, light, dust, airquality raw & acceleration chest, arm, ankle X/Y/Z, ecg 1 & N/A~$^\dagger$ & trip time in sec, trip distance,fare amount \\ \hline
\hline
\multicolumn{5}{l}{\emph{$^\dagger$ No fields were used for the particular task with the dataset as GRID is univariate whereas DTT and MLT tasks require multiple fields.}}
\end{tabular}}
\end{table}
% that's all folks
\end{document}
\begin{thebibliography}{10}
\bibitem{data:city}
Sense your city: Data art challenge.
\newblock Technical report, Data Canvas, 2015.
\bibitem{xchart:lib}
Xchart java library.
\newblock \url{http://knowm.org/open-source/xchart/}, April 2015.
\bibitem{agrawal:sparkbench:2015}
Dakshi Agrawal, Ali Butt, Kshitij Doshi, Josep-L Larriba-Pey, Min Li,
Frederick~R Reiss, Francois Raab, Berni Schiefer, Toyotaro Suzumura, and
Yinglong Xia.
\newblock Sparkbench--a spark performance testing suite.
\newblock In {\em TPCTC}, pages 26--44. Springer, 2015.
\bibitem{ali:citybench:2015}
Muhammad~Intizar Ali, Feng Gao, and Alessandra Mileo.
\newblock Citybench: a configurable benchmark to evaluate rsp engines using
smart city datasets.
\newblock In {\em ISWC}, 2015.
\bibitem{alon:stoc:1996}
Noga Alon, Yossi Matias, and Mario Szegedy.
\newblock The space complexity of approximating the frequency moments.
\newblock In {\em ACM symposium on Theory of computing}, pages 20--29. ACM,
1996.
\bibitem{arasu:vldb:2004}
Arvind Arasu, Mitch Cherniack, Eduardo Galvez, David Maier, Anurag~S Maskey,
Esther Ryvkina, Michael Stonebraker, and Richard Tibbetts.
\newblock Linear road: a stream data management benchmark.
\newblock In {\em VLDB}, 2004.
\bibitem{arlitt:icpe:2015}
Martin Arlitt, Manish Marwah, Gowtham Bellala, Amip Shah, Jeff Healey, and Ben
Vandiver.
\newblock Iotabench: An internet of things analytics benchmark.
\newblock In {\em ICPE}, 2015.
\bibitem{babu:record:2001}
Shivnath Babu and Jennifer Widom.
\newblock Continuous queries over data streams.
\newblock {\em ACM Sigmod Record}, 30(3):109--120, 2001.
\bibitem{balazinska:tods:2008}
Magdalena Balazinska, Hari Balakrishnan, Samuel~R Madden, and Michael
Stonebraker.
\newblock Fault-tolerance in the borealis distributed stream processing system.
\newblock {\em ACM TODS}, 2008.
\bibitem{data:fit}
Oresti Banos, Rafael Garcia, Juan~A Holgado-Terriza, Miguel Damas, Hector
Pomares, Ignacio Rojas, Alejandro Saez, and Claudia Villalonga.
\newblock mhealthdroid: a novel framework for agile development of mobile
health applications.
\newblock In {\em International Workshop on Ambient Assisted Living}, pages
91--98. Springer, 2014.
\bibitem{biem:sigmod:2010}
Alain Biem, Eric Bouillet, Hanhua Feng, Anand Ranganathan, Anton Riabov,
Olivier Verscheure, Haris Koutsopoulos, and Carlos Moran.
\newblock Ibm infosphere streams for scalable, real-time, intelligent
transportation services.
\newblock In {\em ACM SIGMOD}, 2010.
\bibitem{bloom:acm:1970}
Burton~H. Bloom.
\newblock Space/time trade-offs in hash coding with allowable errors.
\newblock {\em Commun. ACM}, 13(7):422--426, July 1970.
\bibitem{carbone2015flink}
Paris Carbone, Stephan Ewen, Seif Haridi, Asterios Katsifodimos, Volker Markl,
and Kostas Tzoumas.
\newblock Apache flink: Stream and batch processing in a single engine.
\newblock {\em IEEE Data Engineering Bulletin}, page~28, 2015.
\bibitem{carney:vldb:2002}
Don Carney, UG'ur {\c{C}}etintemel, Mitch Cherniack, Christian Convey, Sangdon
Lee, Greg Seidman, Michael Stonebraker, Nesime Tatbul, and Stan Zdonik.
\newblock Monitoring streams: a new class of data management applications.
\newblock In {\em VLDB}, pages 215--226, 2002.
\bibitem{ericsson-report}
Patrik Cerwall.
\newblock Ericsson mobility report.
\newblock Technical report, Ericsson, 2016.
\bibitem{aurora}
Ugur Cetintemel, Daniel Abadi, Yanif Ahmad, Hari Balakrishnan, M~Balazinska,
M~Cherniack, J~Hwang, W~Lindner, S~Madden, A~Maskey, et~al.
\newblock The aurora and borealis stream processing engines.
\newblock {\em Data Stream Management: Processing High-Speed Data Streams,,
Springer-Verlag}, pages 1--23, 2006.
\bibitem{chandrasekaran:cidr:2003}
Sirish Chandrasekaran, Owen Cooper, Amol Deshpande, Michael~J. Franklin,
Joseph~M. Hellerstein, Wei Hong, Sailesh Krishnamurthy, Samuel Madden,
Vijayshankar Raman, Frederick Reiss, and Mehul~A. Shah.
\newblock Telegraphcq: Continuous dataflow processing for an uncertain world.
\newblock In {\em CIDR}, 2003.
\bibitem{chen:sigmod:2000}
Jianjun Chen, David~J DeWitt, Feng Tian, and Yuan Wang.
\newblock Niagaracq: A scalable continuous query system for internet databases.
\newblock {\em ACM SIGMOD Record}, 29(2):379--390, 2000.
\bibitem{cugola:csur:2012}
Gianpaolo Cugola and Alessandro Margara.
\newblock Processing flows of information: From data stream to complex event
processing.
\newblock {\em ACM Comput. Surv.}, 2012.
\bibitem{data:taxi}
Brian Donovan and Daniel~B. Work.
\newblock Using coarse gps data to quantify city-scale transportation system
resilience to extreme events.
\newblock In {\em Transportation Research Board 94th Annual Meeting}, 2014.
\bibitem{durand:esa:2003}
Marianne Durand and Philippe Flajolet.
\newblock Loglog counting of large cardinalities.
\newblock In {\em European Symposium on Algorithms}, pages 605--617. Springer,
2003.
\bibitem{gao:bigdatabench:2013}
Wanling Gao, Yuqing Zhu, Zhen Jia, Chunjie Luo, Lei Wang, Zhiguo Li, Jianfeng
Zhan, Yong Qi, Yongqiang He, Shiming Gong, et~al.
\newblock Bigdatabench: a big data benchmark suite from web search engines.
\newblock {\em arXiv preprint arXiv:1307.0320}, 2013.
\bibitem{ghazal:acm:2013}
Ahmad Ghazal, Tilmann Rabl, Minqing Hu, Francois Raab, Meikel Poess, Alain
Crolotte, and Hans-Arno Jacobsen.
\newblock Bigbench: towards an industry standard benchmark for big data
analytics.
\newblock In {\em ACM SIGMOD}, 2013.
\bibitem{gu:chronos:2015}
Ling Gu, Minqi Zhou, Zhenjie Zhang, Ming-Chien Shan, Aoying Zhou, and Marianne
Winslett.
\newblock Chronos: An elastic parallel framework for stream benchmark
generation and simulation.
\newblock In {\em ICDE}, 2015.
\bibitem{huang:hibench:2010}
Shengsheng Huang, Jie Huang, Jinquan Dai, Tao Xie, and Bo~Huang.
\newblock The hibench benchmark suite: Characterization of the mapreduce-based
data analysis.
\newblock In {\em IEEE ICDEW}, 2010.
\bibitem{senml}
C.~Jennings, Z.~Shelby, J.~Arkko, A.~Keranen, and C.~Bormann.
\newblock Media types for sensor measurement lists (senml).
\newblock Technical Report draft-ietf-core-senml-04, Internet Engineering Task
Force (IETF), 2016.
\bibitem{kalman:asme:1959}
Rudolph~Emil Kalman.
\newblock A new approach to linear filtering and prediction problems.
\newblock {\em Transactions of the ASME--Journal of Basic Engineering},
82(Series D):35--45, 1960.
\bibitem{li:tpctc:2014}
Chunhui Li and Robert Berry.
\newblock Cepben: a benchmark for complex event processing systems.
\newblock In {\em TPCTC}, pages 125--142. Springer, 2013.
\bibitem{lu:ucc:2014}
Ruirui Lu, Gang Wu, Bin Xie, and Jingtong Hu.
\newblock Stream bench: Towards benchmarking modern distributed stream
computing frameworks.
\newblock In {\em IEEE/ACM UCC}, 2014.
\bibitem{nabi:streams:2014}
Zubair Nabi, Eric Bouillet, Andrew Bainbridge, and Chris Thomas.
\newblock Of streams and storms.
\newblock Technical report, IBM, 2014.
\bibitem{nambiar:vldb:2006}
Raghunath~Othayoth Nambiar and Meikel Poess.
\newblock The making of tpc-ds.
\newblock In {\em VLDB}, 2006.
\bibitem{neumeyer:icdmw:2010}
Leonardo Neumeyer, Bruce Robbins, Anish Nair, and Anand Kesari.
\newblock S4: Distributed stream computing platform.
\newblock In {\em IEEE ICDMW}, 2010.
\bibitem{perera:ieee:2014}
Charith Perera, Arkady Zaslavsky, Peter Christen, and Dimitrios Georgakopoulos.
\newblock Context aware computing for the internet of things: A survey.
\newblock {\em Communications Surveys \& Tutorials}, 16(1):414--454, 2014.
\bibitem{quinlan:ml:1986}
J.R. Quinlan.
\newblock Induction of decision trees.
\newblock {\em Machine Learning}, 1(1):81--106, March 1986.
\bibitem{shukla:tpctc:2016}
Anshu Shukla and Yogesh Simmhan.
\newblock Benchmarking distributed stream processing platforms for iot
applications.
\newblock In {\em Technology Conference on Performance Evaluation \&
Benchmarking (TPCTC)}, 2016.
\bibitem{simmhan:cise:2013}
Yogesh Simmhan, Saima Aman, Alok Kumbhare, Rongyang Liu, Sam Stevens, Qunzhi
Zhou, and Viktor Prasanna.
\newblock Cloud-based software platform for data-driven smart grid management.
\newblock {\em IEEE/AIP Computing in Science and Engineering},
July/August:1--11, 2013.
\bibitem{siddhi}
Sriskandarajah Suhothayan, Kasun Gajasinghe, Isuru Loku~Narangoda, Subash
Chaturanga, Srinath Perera, and Vishaka Nanayakkara.
\newblock Siddhi: A second look at complex event processing architectures.
\newblock In {\em ACM Workshop on Gateway Computing Environments}, 2011.
\bibitem{data:grid}
{The Research Perspective Ltd}.
\newblock Cer smart metering project.
\newblock Technical report, Commission for Energy Regulation, Ireland, 2012.
\bibitem{toshniwal:sigmod:2014}
Ankit Toshniwal, Siddarth Taneja, Amit Shukla, Karthik Ramasamy, Jignesh~M
Patel, Sanjeev Kulkarni, Jason Jackson, Krishna Gade, Maosong Fu, Jake
Donham, et~al.
\newblock Storm@ twitter.
\newblock In {\em ACM SIGMOD}, pages 147--156, 2014.
\bibitem{iotconnect}
Mark Wallis and Brent Wilson.
\newblock Iot-connect: An industry-standard benchmarks for embedded systems.
\newblock Technical report, EEMBC, 2017.
\bibitem{fitness-iot}
Gary Wolf.
\newblock The data-driven life.
\newblock The New York Times Magazine, 2010.
\bibitem{zaharia:usenix:2012}
Matei Zaharia, Tathagata Das, Haoyuan Li, Scott Shenker, and Ion Stoica.
\newblock Discretized streams: an efficient and fault-tolerant model for stream
processing on large clusters.
\newblock In {\em USENIX Hot Cloud}, 2012.
\bibitem{zaslavsky2013internet}
A~Zaslavsky.
\newblock Internet of things and ubiquitous sensing.
\newblock {\em Computing Now}, 2013.
\bibitem{zhou2016knowledge}
Qunzhi Zhou, Yogesh Simmhan, and Viktor Prasanna.
\newblock Knowledge-infused and consistent complex event processing over
real-time and persistent streams.
\newblock {\em Future Generation Computer Systems}, 2016.
\end{thebibliography}