14.3.1 Big Compute Versus Big Data

Simply processing large datasets is typically not considered to be big data. Groups like Conseil Européen pour la Recherche Nucléaire (CERN) and Transnational Research In Oncology (TRIO) have been using High Performance Computing systems and scalable software to analyze very large datasets. However, this is considered compute-centric processing. Typically, a mathematical algorithm is used to generate results from the data that forms the input. This is true for computational fluid dynamics, image processing, and traditional genome analysis.

From a components perspective, little differentiates a big data machine from a supercomputer; however, the philosophy of design and interaction of components places these systems in two classes. Table 14.1 outlines the salient differences between big compute and big data systems.

Computers, large and small, are very efficient and quick at processing decimal numbers and mathematical equations. While not always true, typically compute-centric processing takes as input small quantities of data and generates large quantities of data while processing or as output. As the power and thermal limits of silicon were approached, multiple processors were ganged together to parallelize and speed up processing. Compute-centric applications typically parallelize the processing by distributing the data and the instruction to multiple processors that can be as close as being located on the same chip, up to being geographically remote and only accessible over the Internet.

Taking traditional high performance computers (HPC) as an exemplar system, processing elements (PE) are typically located across a tightly coupled local area network operating at network speeds between 1 to 100 Gbps. The biggest bottleneck to a compute-centric parallel program on an HPC system is the network. Complex programs are not able to scale to bigger systems due to large global communication operations. Input data therefore cannot be distributed in the same manner as the instruction set, from one system across the entire parallel machine. A dedicate fast shared storage device is used to make the data available to all processing cores. As all the data is clumped together, even if the data domain is partitioned within the algorithm, each PE still needs to traverse the entire dataset to find the required subset. First and foremost, this is a waste of compute cycles, and secondly it causes unsustainable loads on the storage device (e.g., 50–100 PEs synchronously accessing 1 storage server to find the relevant 1 GB of data to process). For example, if we consider a CFD job with a detailed model (>1 GB) of the object then each PE will request that file and then only load into memory the chunk of the model it must process. When designing a typical HPC system (for CFD or FEA workloads) the key metrics in order of importance is processor clock speed, processor density, network interconnect, processing power and network capability of centralized storage, and finally memory per PE.

The main distinction of the big data computing paradigm is that the processing algorithms and systems are usually designed with data centricity in mind. Data centric systems devote the majority of the running time to performing data manipulation and I/O over numerical operations. The optimization of algorithm is secondary to the data management. Within the MapReduce ecosystem (discussed further down) data locality is a major component to the concept of data centricity. Unlike the HPC systems above the data is not stored at a single point. At ingestion time it is in fact divided and replicated across a distributed compute system, and stored local to the processing elements. That way each PE only needs to deal with its local subset of data without needing to search for it. The PEs also do not need to compete with each other for bandwidth over the network. This creative distribution of data ensures that processing instructions only go to the PEs that have direct access to the relevant data. The replication of the data subsets ensures further and much improved parallelization. If two instructions require access to the same subset of data, the replication ensures that two separate PEs can run both instructions simultaneously without any contention for network or disk. When designing a big data system, the metrics to keep in mind are size of PE level storage, speed of PE level storage, memory per PE, processor clock speed, and network interconnect. This is another reason for the penetration of big data within the commercial sector. HPC systems have large power overheads – leading to high processor density, and require highly specialized networking components to deliver performance. Conversely big data systems can be very efficient using high quality workstations loosely coupled with commodity networking equipment. This also makes big data systems easy to deliver over cloud computing infrastructure, lowering the financial barrier to entry.

14.3.2 Data Intensive Applications

There are a variety of computation types that can be considered data intensive. Such applications often use massive quantities of input data to derive some important value from it. Indeed, increasingly a company's success is driven by their ability analyze and draw conclusions from enormous quantities of disperse data sources. The need to derive new information from data has led to the emergence of several sub categories of data intensive applications including:

•  Data Mining – Data-mining algorithms are a broad class algorithms used to extract certain key features or metrics from a given data. The types of algorithms used and features extracted from data vary upon the domain the data belongs to. The algorithms used to mine astronomy data, for example, would differ from those used to mine financial market data [16].

•  Machine Learning – Machine learning algorithms are a varied set used broadly to study and learn from datasets with goals including pattern recognition and future prediction. Machine learning is closely related to the data-mining field and recent improvements in algorithms and the power of modern hardware have accelerated progress in the field. The use of larger and more varied training data sets have increased accuracy of any predictions created [42].

•  Real Time Processing – The introduction of modern software and hardware has increasingly made the real time processing of massive datasets a reality. Such real time analysis is often named stream processing [42]. In stream processing model, data is processed, queried and analyzed immediately as it arrives into the system. Increasingly a data stream from a generation source is sent directly to a compute resource for processing before then being archived for latter access. This is in contrast to the batch processing method, where data is first stored and processed in bulk at a later date.

•  Graph Processing – Many domains can naturally be represented as graphs as they can capture the inherent interconnecting nature of real world phenomena [8]. Such domains include social networks, semantic web, protein networks and computer networks among others. Due to this, there have emerged a class of algorithms designed specifically to extract metrics and topological features from graph objects, giving valuable and unique insight into the dataset.

14.3.3 Data Intensive Frameworks

Data intensive frameworks are a class of software designed to enable the efficient creation and running of data intensive applications. Data intensive applications pose interesting and unique demands on the underlying hardware as data transfer, not processor speeds, limits their performance. As such, data intensive frameworks make important considerations and compromises to optimize for data processing in their architecture design and implementation [68]. The need to process massive quantities of data being generated called for a paradigm shift in the mind-set of system and application designers. This shift called for systems that excel at ingesting, moving, manipulating, and retrieving data on an unprecedented scale. This poses real challenges for both hardware and software. Many of the resulting data intensive frameworks emerged from Internet-focused companies and research institutes, for example MapReduce from Google and Dryad from Microsoft. Data intensive applications prioritize input/output (IO) operations, specifically disk and memory access, over CPU based computation [66]. Both compute and data intensive computing are performed of distributed clusters, usually with a shared-nothing architecture. Although the famous Moore's Law has observed that CPU performance has doubled, due to progress made in the miniaturization of transistors, approximately every two years, IO has experienced a much slower increase in performance. This has led to the notion that compute is considered cheap and data IO is expensive. As data intensive computing is usually performance upon a distributed system, the bandwidth and latency of the network are also an important factor in performance, as large quantities of data needs to be transferred across it. However, the movement of massive datasets across the network has proven to be exceptionally time consuming. For example, a system with a 10 GB network running at 80% utilization, would take approximately 11 days to transfer a dataset 1 petabyte in size [37].

Due to these demands, data intensive frameworks are designed to provide such optimizations as data locality and fault tolerance to optimize the available hardware. There has been a general shift in the management of the complexities introduced by fault tolerance and data locality from hardware to software in the form of data intensive frameworks. In addition, data intensive frameworks mask the complexity of the underlying hardware from the application developer and automatically handle data partitioning, scheduling, and parallelization [33].

14.3.4 MapReduce and GFS

In 2003 and 2004 Google introduced two key concepts to the research community; both would become some of the cornerstones of the data Intensive computing research landscape. Firstly, Google introduced the concept of MapReduce, which it had developed internally as a conceptually simple, yet extremely powerful new programming paradigm for processing massive datasets in parallel. The second paper introduced was the Google File System (GFS), a distributed and fault resilient file system. The GFS system allows for large files to be split into smaller blocks, which can then be distributed across all nodes in a cluster. These file blocks allow for easy resilience against hardware failure, as each block can be replicated across a range of separate machines. The following sections will give an overview of the design and functionality of both GFS and MapReduce and explain why this model has become so successful for Data Intensive Computing.

14.3.4.2 MapReduce

MapReduce is both a powerful programming paradigm and a distributed data processing engine, designed to run on large clusters comprised of commodity hardware originally introduced by Google via a 2004 paper [20]. MapReduce was specifically designed as a new way of processing the massive quantities of data required by a company like Google. Its programming model takes inspiration from functional programming and allows users to easily create scalable data parallel applications, whilst the processing engine ensures fault tolerance, data locality and scheduling automatically. MapReduce is not designed as a replacement for traditional parallel processing frameworks such as MPI; instead it is a response to the new class of applications demanded by the big data phenomenon. When Google originally designed the MapReduce system, the following assumptions and principals guided its development [64]:

•  MapReduce was designed to be deployed on low-cost and unreliable commodity hardware.

•  This hardware was loosely coupled and configured as a Redundant Array of Independent Nodes (RAIN).

•  Nodes from the RAIN were assumed to fail and thus could be removed at any time. These failures should have no impact on any running jobs or result in any data loss.

•  The MapReduce paradigm was designed to be highly parallel, yet abstract enough to allow for fast and easy algorithm development.

In MapReduce, the compute engine and the distributed file-system are designed together and are tightly coupled. The system utilizes this tightly coupled nature to create the key performance driver of MapReduce – data locality. Data locality ensures that the required computation is moved to the data as the node that holds the data will process it [27]. This is an advantage in a modern compute cluster environment, as data transfer is often the bottleneck in application performance and bringing the compute to the data will remove the need for a costly network transfer.

From a conceptual point of view, MapReduce can be considered as just two distinct phases: Map and Reduce [44]. In order to achieve some of its fault tolerance and scalability goals, the MapReduce system places some limitations on the way end users create their applications. Perhaps the most challenging, from an end user's perspective, is that Map tasks must be written in such a way that they can operate completely independently, and in isolation, on a single chunk of the overall larger dataset. Any operations that require some form of communication must be performed in the Reduce phase, which can aggregate the required result from data passed to it by a series of Mappers. To create a MapReduce application an end user must be able to express the logic required by their algorithms in these two phases, although chaining multiple MapReduce iterations together can accommodate more complicated tasks. It is also possible to create Map only jobs for tasks that do not require any sort of accumulations, such as some data cleaning or validation tasks. From an end user's perspective, one of the key strengths of the MapReduce paradigm is that their applications are completely removed from the often-challenging tasks usually associated with parallel computing. The backend system of MapReduce handles the data distribution, fault tolerance and scheduling for the end user's automatically [64]. This frees users to just focus upon the creation of new algorithms and the parallelization is handled automatically. This lowers the complexity of writing algorithms massively and helps democratize the creation of parallel programs so nonspecialists can harness the power of modern compute clusters [20].

The MapReduce system uses key/value pairs as the input and output for both of the stages. The input data is presented to the Map function as key/value pairs and, after processing, the output is stored as another set of key/value pairs. In between the Map and Reduce phases, common keys in the Map output are grouped together so all the associated values are available for processing in the same Reduce task. The final processing and result from the Reduce task are again output as key/value pairs [20]. One of the key performance drivers of MapReduce is that the Map phase is highly parallel. By default, if the input data resides in m blocks, then m Map tasks will be spawned. As GFS ensures that blocks are distributed across the entire cluster, the Map tasks will be executed on many nodes simultaneously. This is the simplistic model of MapReduce and gives a good representation of how data will flow through an application but it does not discuss some key behind the scenes operations performed by the system.

Greater insight into the operation and intricacies of MapReduce can be gained through the analysis of the word count application, also known as the hello world of MapReduce. For this example application, the input to the program will be a collection of text documents stored on a GFS-like file system and will be completed in a single MapReduce phase. The collection of documents would be split into m 64 MB chunks automatically by the GFS. Once the MapReduce program was launched m Map tasks would be created, wherever possible, upon the nodes containing the relevant file chunks. In the word count application, the role of the Map task is to split the text data contained in the block, using whitespace, into a sequence of individual words. The Map function would then emit its series of intermediate key/value pair with each word located being the key and the value being an integer value of one. Pseudocode representing this process is shown as Code 1.

An example output from the Map phase would be $(w1,1),(w2,1),...,(wn,1)$. The transfer of data between the Map and Reduce phases is handled by a process called shuffle and sort. The role of the shuffle and sort phase is to collect and sort the values associated with a specific key so that they are all presented to a single Reduce task. This phase can be a performance bottleneck, as all the intermediate data from the Mappers is first written back to disk before then being transferred over the network to the nodes that will run the Reduce task. In the word count application an example input to the Reduce function would be ($w1,[11111]$), for a word that had five instances in the original input file. The Reduce function would then simply sum the integer values for all keys and emit the total, along with the original word as the final output of the application – $(w1,5)$.

Whilst MapReduce has proven to be extremely powerful and popular, it is not without fault and has received some criticism within academic literature [21,56]. The main arguments against MapReduce centers around a few key areas including the comparatively reduced functionality, its lack of suitability for certain computation tasks, a still relatively low-level API and its need for a dedicated cluster resource. When contrasted with other data management and query systems, such as SQL, MapReduce can appear to offer limited functionally. Simple standard relation database operations such as joins are complicated in MapReduce and often require sophisticated solutions. Although several strategies, including Map-side, Reduce-side and Cascade joins, have emerged to enable the functionality, the framework was clearly not designed with workflows involving numerous complicated joins in mind [1]. MapReduce completely removes the data schema used by traditional databases so all data is stored without structure or an index, meaning that it is unable to utilize the possible optimizations offered by structured and semistructured data [21]. The lack of an index means that the entire dataset must be traversed to search for a specific portion of the data, which can be costly, especially with massive datasets. In its original incarnation there is no higher-level language for MapReduce, and users must write their applications using the still low-level API. When compared with writing SQL queries, for example, the MapReduce API has a greater level of complexity and requires more lines of code. In addition, MapReduce code is often less portable and tends to be very data-specific [44]. For certain data processing tasks, particularly those that require many iterations over the same dataset, the MapReduce paradigm is unsuitable. Unfortunately, many state-of-the-art machine learning and graph processing algorithms display exactly these very characteristics [78]. From a system point of view, MapReduce is often deployed on its own dedicated hardware, as the system does not lend itself to resource sharing with competing frameworks such as MPI. This can increase costs for an organization as it potentially must purchase and maintain two clusters if the requirement for both systems is present within the organization.

Despite these limitations, MapReduce has proved to be extremely popular in both industry and academia. Although it should be clear now that MapReduce is best suited for use in a specific class of application, when a massive amount of data needs processing and when the required processing fits well within the data parallel model. The original implementation of MapReduce cannot be, nor was it designed to be, a replacement for parallel processing engines such as MPI or structured database systems such as SQL. Instead it compliments these systems and fills the void when the required workload doesn't fit into either paradigm.

14.4.1 Hadoop V1

Apache Hadoop is an open-source project first developed by researchers at Yahoo [4]. It was designed to replicate the functionality of both Google's GFS and MapReduce system. As such, Hadoop originally launched with two components; the Hadoop Distributed File System (HDFS), designed to replicate GFS, and its own implementation of the MapReduce runtime and programming model [67]. Since the code was first made publicly available in 2007, Hadoop has grown to become the de facto standard for big data processing. Since it is a replication of the original Google ideas, Hadoop inherits the same set of strengths and weakness as was previously discussed. The only major differences between them are that Hadoop is written in the Java programming language, compared with the $\mathrm{C}++$ of the original Google system, and that Hadoop is open-source, thus it can be used and modified free of charge.

Hadoop is designed to run on a distributed shared nothing cluster of commodity machines. The original Hadoop implementation runs as a series of daemon processes, with nodes in the cluster taking one of two roles, master or slave. Each of the four separate daemon processes runs within a Java Virtual Machine (JVM). These daemon processes (shown in Fig. 14.1) are:

•  JobTracker – Master process for the MapReduce component that controls the submission and scheduling of jobs on the cluster. Runs on the master node.

•  NameNode – Master process for the HDFS that keeps track of how data has been distributed across the available DataNode. Runs on the master node.

•  TaskTracker – These processes are the worker components of the MapReduce system. The TaskTracker demons themselves do not perform the computation; instead they control the spawning of separate JVMs for each MapReduce job. These are run upon the slave nodes.

•  DataNode – These processes control the data stored in the HDFS. These are also run upon the slave nodes.

14.4.2 Hadoop 2.0

In Hadoop V1, the management and resource negotiation is controlled via the MapReduce runtime. As such, any additional system wishing to access data stored in HDFS must be able to translate itself down to a series of MapReduce tasks. This places a massive restriction of the type of computation that can be performed via Hadoop. To address this, Hadoop 2.0 incorporated a new component called Yet Another Resource Negotiator (YARN) into the software stack [72]. YARN's key advancement is that it separates scheduling decisions from the data processing layer. This means that workflows no longer need to be constructed in terms of MapReduce tasks, potentially manipulated or misused, to achieve completion of work not suited to a MapReduce framework. Examples of this include map-only tasks written for Hadoop 1.0 that would spawn services such as web servers. One of the drawbacks with Hadoop V1 was the single point of failure that the JobTracker daemon represented. If a JobTracker were to fail, the resulting situation would be one where all running jobs within a system would be lost, there was no mechanism for automatic recovery from such a situation and all users would have to manually resubmit jobs for completion once the system was restored. YARN solves this key problem by removing cluster management duties from the JobTracker [72].

The YARN system architecture comprises three separate components, the ResourceManager (RM), the NodeManager (NM), and the ApplicationManager (AM) [72]:

•  ResourceManager – The RM removes responsibility for scheduling of workflows from the JobTracker and serves containers for processing, which are logical bundles of available resources. This process is responsible for global system view collected from communication with the NM and AM, along with overall management of the system through the servicing resource requests made by user job submissions and the AMs. There is also a mechanism whereby the RM can request the return of resources from an AM, if it is deemed to be oversubscribed on the system, or forcibly claw back resources by instructing a NM to terminate containers if it deems too much time has elapsed following a request to an AM. In order to mitigate a single point of failure situation there is a standby RM that can take over responsibility of the system and spawn a new standby RM in the eventuality of a master RM failure. If there is a situation where a node, or NM, fails the RM will detect this, update system global state, report the failure to all running AMs and restart any AMs that were lost due to the failure of the NM.

•  NodeManager – The NM launches and manages all containers, including AMs, configuring the environment as appropriate from AM, and user, requests received via the RM. Container requirements, including resources and environment, are described through a Container Launch Context (CLC) that are sent with every request for container creation. The NM is also responsible for reporting actual hardware utilization to the RM for maintenance of the global cluster state view. Upon creation all containers are issued a lease to use the requested resources this enables the RM to make better scheduling decisions based on a known lifetime of existing processes. Further, the NM cleans up all processes and files when a container exits, the NM is not aware is a container has exited cleanly or not. Handling of container failure is left to running application framework, i.e., it is up to the AM to know that a container has not exited cleanly and to request a new container via the RM.

•  ApplicationManager – Bootstrap process, running within a NM spawned container, responsible for the execution of a “job” within the cluster, whether it be set of processes, a logical description of work, or even a long-running service. It requests required resources from the RM and executes code within other available containers. Also responsible for local job optimization, management for loss of resources (node failure, other than node running AM) and reporting of job based metrics.

14.5.1 Resilient Distributed Datasets

The core of Spark is the Resilient Distributed Dataset (RDD) abstraction. An RDD is a read-only collection of data that can be partitioned across a subset of Spark cluster machines and form the main working component [77]. RDDs are so integral to the function of Spark that the entire Spark API can be considered to be a collection of operations to create, transform, and export RDDs. Every algorithm implemented in Spark is effectively a series of transformative operations performed upon data represented as an RDD. The key performance driver of Spark is that an RDD can be cached in memory of the Spark cluster compute nodes and thus can be reused by many iterative tasks. The input data that forms a RDD is partitioned into chunks and distributed across all the nodes in the Spark cluster, with each node then performing computation in parallel upon its own set of chucks. Physically an RDD is a Scala object and can be constructed from a variety of data sources, including files from HDFS or other file system, directly from Scala arrays or from a range of transformations that can be performed upon an existing RDD [77]. A key feature of RDDs is that they can be reconstructed if a RDD partition is lost using a concept called lineage and thus can be considered to be fault tolerant. Spark keeps a record of the lineage of an RDD but tracking the transformation that have been performed to create it. If any part of an RDD is lost then Spark will utilize this linage record to quickly and efficiently re-compute the RDD using the identical operations that created the original version [77]. This method of lineage recomputation removes the need for costly data replication strategies used by other methods for abstracting in-memory storage across a compute cluster. However, if the lineage chain reaches a large enough size, users can manually flag a specific RDD to be check-pointed. Check-pointed RDDs are written to disk or HDFS to avoid the recomputation of long lineage chains. RDDs are not for storing or archiving final result data; this is still handled via HDFS or other file system.

There is a range of parallel operations that can be performed upon a RDD using the Spark Core API. These operations fall into two distinct categories, Transformations and Actions [77]. Transformations contain functions that will create new RDDs from existing data sources, whilst actions trigger computations to calculate a return value from the data or write the data out to external storage. The Transformation operations are lazy, so computation will not occur when these are called. An Action must be called to force the computation to happen. This lazy execution style has several advantages in the optimization of storage and performance for Spark. Spark will inspect a complete sequence of Transformations before a user's application is physically run upon the cluster. This allows for an overall picture of the data lineage to be created so that Spark can evaluate the complete transformation chain. Knowing this complete chain allows Spark to atomically optimize before runtime by only computing with the data required for the final result [9].

The RDD concept has further been expanded via the introduction of Spark DataFrames, [7]. A Spark DataFrame arranges the distributed collection of data into labeled columns similar to a traditional relational database. DataFrames are at a higher level of abstraction then RDDs and include a schema, allowing Spark to perform more automatic optimization at run time by exploiting the structured nature [7]. In addition to the performance improvements, the DataFrames API contains more domain specific functionality not offered by the RDD API, including joins and aggregations familiar to many data scientists.

14.5.2 Data Flow and Programming With Spark

To create a program with Spark is very similar to existing data flow languages. End users are only required to write a single class that acts as a high-level program driver. This is in contrast to MapReduce, where 3 classes are required to create a typical program, the driver, mapper and reducer. In Spark, the user created driver program defines how the input data will be transformed to create the desired output. Users can create their applications in Java, Scala, Python or R using the provided APIs. However, Spark is a Scala first platform, meaning that new features in the API are added first in the Scala language. Scala is a functional language, and so can be a bit of a learning curve for users more familiar with OOP. Users of other languages are forced to wait for the latest Spark features to be pushed down steam. The program is submitted on the Spark cluster via the head node, and is then automatically distributed across the cluster so that it can be executed upon the partition of the RDD stored upon each worker node.

The data flow of a Spark program can be considered as follows: Firstly, the driver creates a single or multiple RDDs from a Hadoop or other data source. It then will apply transformations upon the RDD that will create new RDDs. Finally, some number of actions will be invoked upon the transformed RDDs to create the final output from the program. Perhaps the best way to envisage how Spark functions is to consider a typical example application. In this example we will explore how to create a simple log-processing program that will take a log file as input and count the number of times a keyword appears. The overall plan for this application is to take a log file stored on a HDFS, extract only the required lines, load it into memory and count the number of stored lines. To create this application in Spark would require the following steps; firstly, an RDD would be created from the log file stored on the HDFS. The second RDD would be created via the filter transformation; this would filter the original file and only select the lines starting with the string “ERROR.” A third RDD would be created using the map transformation to split the string on white space to remove the “ERROR” sub-string from the beginning and load it into memory. Finally, the number of elements would be counted using the inbuilt count function. It may seem inefficient to create three RDDs for such a simple application but as the computation is only triggered when the count Action is called, Spark's execution planning will ensure that only the required data is processed.

Users must still be careful when designing their algorithms to best make use of the Spark system. Spark can still require a fair amount of tuning to achieve optimal performance when running on a cluster.

14.5.3 Spark Processing Engines

In addition to the Spark Core API for creating data flow programs using the RDD abstract, Spark is integrated with a number of additional higher-level APIs that include functionality for specific classes of compute and data problems. As of Spark 2.0.1, the additional APIs are GraphX, MLlib, Spark SQL, and Spark Streaming. These APIs replicate functionality that, if using Hadoop, would require numerous additional domain specific packages to be deployed alongside the base Hadoop. As well as the standard libraries included with Spark, a large number of third-party libraries and APIs are maintained in an online repository and can easily be included by a user when creating their applications.

GraphX is a system that allows users to process graph data using Spark [75]. Its API includes some fundamental operations such as subgraph creation and a variant of Google's Pregel API, the first of the Think Like A Vertex (TLAV) systems [49]. The TLAV paradigm has emerged as the most widely used method for processing massive graphs on distributed systems [51]. GraphX also includes implementations of many of the most commonly used algorithms for graph analysis including PageRank, connected components and triangle counting. As many graph-based algorithms are inherently iterative processes, often values have to be recomputed on the vertices until a particular threshold is reached, they are an ideal class of problems to be performed via Spark. GraphX's performance has been shown to beat other TLAV style frameworks based on Hadoop, as well as other dedicated graph processing systems [8]. A detailed comparison of some of the major graph processing platforms, performed by Batarfi et al. [8], shows that GraphX is often the faster platform at performing the computation, while also consuming fewer resources and generating less network traffic to do so.

MLlib is Spark's distributed machine learning library and provides fast implementations of several key algorithms from the field [52]. Using the DataFrame API, users can create a machine-learning pipeline, mixing several algorithms together to produce the required result. MLlib includes many of the fundamental algorithms of machine learning including linear support vector machines (SVM), logistical regression, random forest and several clustering methods along others [52]. MLlib is shown to be faster and more scalable then its main, Hadoop based, rival Mahout [52].

The Spark SQL library allows for Spark to process and query structured data via standard SQL syntax [7]. Spark SQL can load and query data taken from a variety of sources, and due to sharing some components, is particularly well integrated with Hive. Data can be loaded from mutual sources into a DataFrame, upon which a query can be run. Spark SQL supports all the main SQL data types including bools, integers, floats and strings. Perhaps the most interesting aspect to Spark SQL is that it allows users to combine relational with procedural processing within the same application [7]. Users can, for example, feed the results from an SQL query into functions from the RDD API to create complicated data flows. Spark SQL simultaneously makes Spark more accessible to new users and introduces key optimizations, in the form of DataFrames, for existing ones [7].

The last of the integrated Spark processing engines is Spark Streaming, which allows users to create applications that process data in near real-time [79]. The API has been designed such that writing a stream application is very similar to creating a batch processing application. Users can reuse code from batch processing and even integrate data from historical sources, via the use of join, for example, as the streaming application is running. An application created using Spark Streaming can take input from a variety of sources in including HDFS, Twitter Flume and ZeroMQ. Spark Streaming utilizes the DStreams abstract to represent input stream data as a series of RDDs [79]. Due to the process of creating the RDD's Spark streaming is not real time but can be considered as a micro-batch system.

The DStreams abstraction partitions the input stream into a series of time blocks. Each block contains a few 100 milliseconds of the stream and is stored in an RDD, upon which the standard Spark batch functions can be performed. The performance of Spark Streaming has been shown to scale nearly linearly, and can achieve a throughput rate of 20 MB per second on each node of the Spark cluster [79].

These additional integrated libraries included with Spark are another of its key advantages and set it apart from the Hadoop Ecosystem. Firstly, users need only learn one programming language and set of system APIs. This also benefits system administrators who only have to deploy and configure one software package, versus the many specialized systems that it would require to replicate the same functionality. Secondly, the integration and interoperability enabled between different computing paradigms via the unified Spark stack, allows users to create new classes of applications. This new class of big data applications will be made possible via the integration of previously separate systems. One such example application enabled by this integration is real-time fraud detection for the financial sector. Creating this application in Spark, the user could train the MLlib module on historical data and then use the developed model to detect fraud happening in real-time using the Spark Streaming module. This is just one of the many new applications that could be created using the integrated platform offered via the Spark framework.

14.5.4 Hadoop Ecosystem Taxonomy

Table 14.2 gives a summary of some of the key differences between, Hadoop V1, Hadoop 2.0, Spark, and Flink. These frameworks are compared on a number of metrics and capabilities.

14.7.1 Big Data Applications

The use of big data techniques and processing has already been successful in revolutionizing several fields. Areas such as social network analysis, the advertising industry and the management of large computer systems have already benefited. However, big data is poised to be successful in many more including:

•  Healthcare – There are numerous ways that big data could revolutionize the Healthcare industry. One such area that has been explored is error checking and anomaly detection in medical research datasets [10], although the possibilities for advances across the whole field are endless [59].

•  Governance – Extending the concept of creating holistic user models, governments are attempting to adopt a person-centric approach to governance. This will lead to customized interactions between the citizenry and local authorities [40]. Through analytics, modeling and simulations, urban authorities hope to be proactive in dealing with potential difficulties rather than reactive [48].

•  Smart Cities – With the automation of many core operations across urban environments and the deployment of different kinds of sensors where possible, the next step is the integration of these systems. Complex models are being developed that integrate real-time data from the various systems and sensors with simulated environments and humans-in-the-loop to streamline city wide operations and mitigate knock on affects that are caused by problems within one or more systems [19,62].

•  Industry 4.0 – The next revolution in the industrial production processes is the development of agile and flexible systems that are driven by analytics and simulation to respond quickly to problems and variations within the production environment. Intelligent Manufacturing Systems (IMS), using virtual, fractal, bionic, and holonic manufacturing systems, are being developed to provide production control. The systems not only factor in the factory floor but also the wider supply chain within its decision models [25,41].

•  The Internet of Things (IoT) – The IoT revolution will be driven by integration of Internet functionality into a range of products and sensors, allowing them to record and transmit data to be sent for analysis [26]. It can be argued that IoT will be the technological backbone used to drive the new class of intelligent Smart Cities and Industry 4.0 applications [26].

To meet the demands of these fields, future big data applications will need ever-greater integration with machine learning to drive intelligent conclusions from the wealth of available data. Many big data applications have traditionally been batch processes, where data was first collected and stored, before being processed at a later date. However, IoT, Healthcare, and Smart Cities applications will need to process data in real-time as they will require instantaneous answers drawn from the stream of data. In addition, all areas will begin to demand subsecond responses to queries on existing data sources.

14.7.2 Big Data Frameworks and Hardware

The evolution of data intensive frameworks has been driven by the roles demanded of them by data intensive applications. The development of the original MapReduce was in response to a new type of workload. Spark was created, in part, as a way to perform many iterative computations over the same working dataset. As such, the role of data intensive frameworks has been evolving since the introduction of MapReduce in 2004, driven by new classes of data intensive problems. Some of the current key roles of a data intensive framework include:

•  Provide fault tolerant storage for massive quantities of data from disperse sources.

•  Provide a common and expressive set of APIs, written in a variety of languages, enabling users to access and process massive quantities of data.

•  Allow users to focus on application development by abstracting away the traditional complexities of parallel computing.

•  Automatically schedule jobs to run in parallel across the underlying compute resource, whilst considering aspects such as data locality.

The evolution of data intensive frameworks will continue and will be focused towards increasing the speed of applications, creating more scalable applications and creating easier to use APIs with greater breadth of functionalities. More specifically, data intensive framework evolution will focus on these key areas:

•  Continue to increase the possible application performance to meet the demand of sub-second and real-time data analytics by more optimal use of resources and by allowing the applications to scale across larger compute clusters.

•  Allow an increasingly diverse range of compute paradigms to interoperate and share access to data and allow the same physical resource to be shared between compute and data centric frameworks.

•  Continue to reduce the complexity of creating big data applications with the introduction of more generic and higher-level APIs for developers, along with creating easy deployment methods for data intensive frameworks.

14.7.3 Big Data on the Road to Exascale

The increase in available data inputs led to the creation of data centric processing frameworks. Traditional large computer systems were geared to scale the processing elements but not increasing input data. This trend does not appear to be changing. In its 2015 Hype Cycle for Emerging Technology, Gartner Inc. put IoT, Advanced Analytics and Machine Learning at the “Peak of Inflated Expectations” with 5–10 years still to go before plateauing out as productive technologies [23]. Over the next 10 years and beyond the size and scope of data will grow considerably and computing paradigms will also have to adapt.

On the technological front, exascale has captured the imagination of experts in the field. Exascale has two facets, the first and the one that gets the most hype is exascale computer systems. Computers capable of calculating 10¹⁸ floating-point operations a second, or an exaflop, is the next challenge for computing in general and high performance computing in particular. This scale of computing comes with a whole host of challenges: from limitations of the silicon substrate, to the power and thermal limits, and I/O limitations (disk, network, etc.) breaking the exa barrier will be challenging. On the software side as well, compilers and program workflows have not fully addressed the challenges of petascale (10¹⁵) computing, and so exascale remains a goal that keeps getting pushed back.

On the data front, reaching exascale is right on the horizon. Datasets that are petabytes and terabytes in size are currently being processed by teraflop and gigaflop capable systems. With the step changes coming, through the phenomenon of Internet of things and person-centric services, workloads that are 1 exabyte (EB) in size will quickly become a reality. Exabyte's of data will be thrust upon businesses and this will be a real challenge to deal with. At that scale the noise ratio becomes much higher and the cost trade-offs also need to be revaluated.

The challenges facing the high performance computing community are the same that will shape the direction of big data. Technologically, scaling big data frameworks to exaflop and petaflop systems will see the same compiler, networking, power, and cost issues as in HPC. Considerable development in both hardware and software is required to deal with the oncoming data deluge.