Dr. O. Aly
Computer Science

Abstract

The purpose of this project is to discuss how data can be handled before Hadoop can take action on breaking data into manageable sizes.  The discussion begins with an overview of Hadoop providing a brief history of Hadoop and the difference between Hadoop 1.x and Hadoop 2.x. The discussion involves the Big Data Analytics process using Hadoop which involves six significant steps including the pre-processing data and ETL process where the data must be converted and cleaned before processing it.  Before data processing, some consideration must be taken for data preprocessing, modeling and schema design in Hadoop for better processing and data retrieval as it will affect how data can be split among various nodes in the distributed environment because not all tools can split the data.  This consideration begins with the data storage format, followed by Hadoop file types consideration and XML and JSON format challenges in Hadoop.  The compression of the data must be considered carefully because not all compression types are “splittable.” The discussion also involves the schema design consideration for HDFS and HBase since they are used often in the Hadoop ecosystem. 

Keywords: Big Data Analytics; Hadoop; Data Modelling in Hadoop; Schema Design in Hadoop.

Introduction

In the age of Big Data, dealing with large datasets in terabytes and petabytes is a reality and requires specific technology as the traditional technology was found inappropriate for it (Dittrich & Quiané-Ruiz, 2012).  Hadoop is developed to store, and process such large datasets efficiently.  Hadoop is becoming a data processing engine for Big Data (Dittrich & Quiané-Ruiz, 2012).  One of the significant advantages of Hadoop MapReduce is allowing non-expert users to run easily analytical tasks over Big Data (Dittrich & Quiané-Ruiz, 2012). However, before the analytical process takes place, some schema design and data modeling consideration must be taken for Hadoop so that the data process can be efficient (Grover, Malaska, Seidman, & Shapira, 2015).  Hadoop requires splitting the data. Some tools can split the data while others cannot split the data natively and requires integration (Grover et al., 2015). 

This project discusses these considerations to ensure the appropriate schema design for Hadoop and its components of HDFS, HBase where the data gets stored in a distributed environment.   The discussion begins with an overview of Hadoop first, followed by the data analytics process and ends with the data modeling techniques and consideration for Hadoop which can assist in splitting the data appropriately for better data processing performance and better data retrieval.

Overview of Hadoop

            Google published and disclosed its MapReduce technique and implementation early around 2004 (Karanth, 2014).  It also introduced the Google File System (GFS) which is associated with MapReduce implementation.  The MapReduce, since then, has become the most common technique to process massive data sets in parallel and distributed settings across many companies (Karanth, 2014).  In 2008, Yahoo released Hadoop as an open-source implementation of the MapReduce framework (Karanth, 2014; sas.com, 2018). Hadoop and its file system HDFS are inspired by Google’s MapReduce and GFS (Ankam, 2016; Karanth, 2014).  

The Apache Hadoop is the parent project for all subsequence projects of Hadoop (Karanth, 2014).  It contains three essential branches 0.20.1 branch, 0.20.2 branch, and 0.21 branch.  The 0.20.2 branch is often termed MapReduce v2.0, MRv2, or Hadoop 2.0.  Two additional releases for Hadoop involves the Hadoop-0.20-append and Hadoop-0.20-Security, introducing HDFS append and security-related features into Hadoop respectively.  The timeline for Hadoop technology is outlined in Figure 1.

Figure 1.  Hadoop Timeline from 2003 until 2013 (Karanth, 2014).

Hadoop version 1.0 was the inception and evolution of Hadoop as a simple MapReduce job-processing framework (Karanth, 2014).  It exceeded its expectations with wide adoption of massive data processing.  The stable version of the 1.x release includes features such as append and security.  Hadoop version 2.0 release came out in 2013 to increase efficiency and mileage from existing Hadoop clusters in enterprises.  Hadoop is becoming a common cluster-computing and storage platform from being limited to MapReduce only, because it has been moving faster than MapReduce to stay leading in massive scale data processing with the challenge of being backward compatible (Karanth, 2014). 

            In Hadoop 1.x, the JobTracker was responsible for the resource allocation and job execution (Karanth, 2014).  MapReduce was the only supported model since the computing model was tied to the resources in the cluster. The yet another resource negotiator (YARN) was developed to separate concerns relating to resource management and application execution, which enables other application paradigms to be added into Hadoop computing cluster. The support for diverse applications result in the efficient and effective utilization of the resources and integrates well with the infrastructure of the business (Karanth, 2014).  YARN maintains backward compatibility with Hadoop version 1.x APIs  (Karanth, 2014).  Thus, the old MapReduce program can still execute in YARN with no code changes, but it has to be recompiled (Karanth, 2014).

            YARN abstracts out the resource management functions to form a platform layer called ResourceManager (RM) (Karanth, 2014).  Every cluster must have RM to keep track of cluster resource usage and activity.  RM is also responsible for allocation of the resources and resolving contentions among resource seekers in the cluster.  RM utilizes a generalized resource model and is agnostic to application-specific resource needs.  RM does not need to know the resources corresponding to a single Map or Reduce slot (Karanth, 2014). Figure 2 shows Hadoop 1.x and Hadoop 2.x with YARN layer.   

Figure 2. Hadoop 1.x vs. Hadoop 2.x (Karanth, 2014).

Hadoop 2.x involves various enhancement at the storage layer as well.   These enhancements include the high availability feature to have a hot standby of NameNode (Karanth, 2014), when the active NameNode fails, the standby can become active NameNode in a matter of minutes.  The Zookeeper or any other HA monitoring service can be utilized to track NameNode failure (Karanth, 2014).  The failover process to promote the hot standby as the active NameNode is triggered with the assistance of the Zookeeper.  The HDFS federation is another enhancement in Hadoop 2.x, which is a more generalized storage model, where the block storage has been generalized and separated from the filesystem layer (Karanth, 2014).  The HDFS snapshots is another enhancement to the Hadoop 2.x which provides a read-only image of the entire or a particular subset of a filesystem to protect against user errors, backup, and disaster recovery.   Other enhancements added in Hadoop 2.x include the Protocol Buffers (Karanth, 2014). The wire protocol for RPCs within Hadoop is based on Protocol Buffers.  Hadoop 2.x is aware of the type of storage and expose this information to the application, to optimize data fetch and placement strategies (Karanth, 2014).  HDFS append support has been another enhancement in Hadoop 2.x.

Hadoop is regarded to be the de facto open-source framework for dealing with large-scale, massively parallel, and distributed data processing (Karanth, 2014).  The framework of Hadoop includes two layers for computation and data layer (Karanth, 2014).  The computation layer is used for parallel and distributed computation processing, while the data layer is used for a highly fault-tolerant data storage layer which is associated with the computation layer.  These two layers run on commodity hardware, which is not expensive, readily available, and compatible with other similar hardware (Karanth, 2014).

Hadoop Architecture

Apache Hadoop has four projects: Hadoop Common, Hadoop Distributed File System, Yet Another Resource Negotiator (YARN), and MapReduce (Ankam, 2016).  The HDFS is used to store data, MapReduce is used to process data, and YARN is used to manage the resources such as CPU and memory of the cluster and common utilities that support Hadoop framework (Ankam, 2016; Karanth, 2014).  Apache Hadoop integrates with other tools such as Avro, Hive, Pig, HBase, Zookeeper, and Apache Spark (Ankam, 2016; Karanth, 2014).

            Hadoop three significant components for Big Data Analytics.  The HDFS is a framework for reliable distributed data storage (Ankam, 2016; Karanth, 2014).  Some considerations must be taken when storing data into HDFS (Grover et al., 2015).  The multiple frameworks for parallel processing of data include MapReduce, Crunch, Cascading, Hive, Tez, Impala, Pig, Mahout, Spark, and Giraph (Ankam, 2016; Karanth, 2014). The Hadoop architecture includes NameNodes and DataNodes.  It also includes Oozie for workflow, Pig for scripting, Mahout for machine learning, Hive for the data warehouse.  Sqoop for data exchange, and Flume for log collection.  YARN is in Hadoop 2.0 as discussed earlier for distributed computing, while HCatalog for Hadoop metadata management.  HBase is for columnar database and Zookeeper for coordination (Alguliyev & Imamverdiyev, 2014).  Figure 3 shows the Hadoop ecosystem components.

Figure 3.  Hadoop Architecture (Alguliyev & Imamverdiyev, 2014).

Big Data Analytics Process Using Hadoop

The process of Big Data Analytics involves six essential steps (Ankam, 2016). The identification of the business problem and outcomes is the first step.  Examples of business problems include sales are going down, or shopping carts are abandoned by customers, a sudden rise in the call volumes, and so forth.  Examples of the outcome include improving the buying rate by 10%, decreasing shopping cart abandonment by 50%, and reducing call volume by 50% by next quarter while keeping customers happy.  The required data must be identified where data sources can be data warehouse using online analytical processing, application database using online transactional processing, log files from servers, documents from the internet, sensor-generated data, and so forth, based on the case and the problem.  Data collection is the third step in analyzing the Big Data (Ankam, 2016).  Sqoop tool can be used to collect data from the relational database, and Flume can be used for stream data.  Apache Kafka can be used for reliable intermediate storage.  The data collection and design should be implemented using the fault tolerance strategy (Ankam, 2016).  The preprocessing data and ETL process is the fourth step in the analytical process.  The collected data comes in various formats, and the data quality can be an issue. Thus, before processing it, it needs to be converted to the required format and cleaned from inconsistent, invalid or corrupted data.  Apache Hive, Apache Pig, and Spark SQL can be used for preprocessing massive amounts of data.  The analytics implementation is the fifth steps which should be in order to answer the business questions and problems. The analytical process requires understanding the data and relationships between data points.  The types of data analytics include descriptive and diagnostic analytics to present the past and current views of the data, to answer questions such as what and why happened.  The predictive analytics is performed to answer questions such as what would happen based on a hypothesis. Apache Hive, Pig, Impala, Drill, Tez, Apache Spark, and HBase can be used for data analytics in batch processing mode.  Real-time analytics tools including Impala, Tez, Drill, and Spark SQL can be integrated into the traditional business intelligence (BI) using any of BI tools such as Tableau, QlikView, and others for interactive analytics. The last step in this process involves the visualization of the data to present the analytics output in a graphical or pictorial format to understand the analysis better for decision making.  The finished data is exported from Hadoop to a relational database using Sqoop, for integration into visualization systems or visualizing systems are directly integrated into tools such as Tableau, QlikView, Excel, and so forth.  Web-based notebooks such as Jupyter, Zeppelin, and Data bricks cloud are also used to visualize data by integrating Hadoop and Spark components (Ankam, 2016). 

Data Preprocessing, Modeling and Design Consideration in Hadoop

            Before processing any data, and before collecting any data for storage, some considerations must be taken for data modeling and design in Hadoop for better processing and better retrieval (Grover et al., 2015).  The traditional data management system is referred to as Schema-on-Write system which requires the definition of the schema of the data store before the data is loaded (Grover et al., 2015).  This traditional data management system results in long analysis cycles, data modeling, data transformation loading, testing, and so forth before the data can be accessed (Grover et al., 2015).   In addition to this long analysis cycle, if anything changes or wrong decision was made, the cycle must start from the beginning which will take longer time for processing (Grover et al., 2015).   This section addresses various types of consideration before processing the data from Hadoop for analytical purpose.

Data Pre-Processing Consideration

The dataset may have various levels of quality regarding noise, redundancy, and consistency (Hu, Wen, Chua, & Li, 2014).  Preprocessing techniques must be used to improve data quality should be in place in Big Data systems (Hu et al., 2014; Lublinsky, Smith, & Yakubovich, 2013).  The data pre-processing involves three techniques: data integration, data cleansing, and redundancy elimination.

The data integration techniques are used to combine data residing in different sources and provide users with a unified view of the data (Hu et al., 2014).  The traditional database approach has well-established data integration system including the data warehouse method, and the data federation method (Hu et al., 2014).  The data warehouse approach is also known as ETL consisting of extraction, transformation, and loading (Hu et al., 2014).  The extraction step involves the connection to the source systems and selecting and collecting the required data to be processed for analytical purposes.  The transformation step involves the application of a series of rules to the extracted data to convert it into a standard format.  The load step involves importing extracted and transformed data into a target storage infrastructure (Hu et al., 2014).  The federation approach creates a virtual database to query and aggregate data from various sources (Hu et al., 2014).  The virtual database contains information or metadata about the actual data, and its location and does not contain data itself (Hu et al., 2014).  These two data pre-processing are called store-and-pull techniques which is not appropriate for Big Data processing, with high computation and high streaming, and dynamic nature (Hu et al., 2014).  

The data cleansing process is a vital process to keep the data consistent and updated to get widely used in many fields such as banking, insurance, and retailing (Hu et al., 2014).  The cleansing process is required to determine the incomplete, inaccurate, or unreasonable data and then remove these data to improve the quality of the data (Hu et al., 2014). The data cleansing process includes five steps (Hu et al., 2014).  The first step is to define and determine the error types.  The second step is to search and identify error instances.  The third step is to correct the errors, and then document error instances and error types. The last step is to modify data entry procedures to reduce future errors.  Various types of checks must be done at the cleansing process, including the format checks, completeness checks, reasonableness checks, and limit checks (Hu et al., 2014).  The process of data cleansing is required to improve the accuracy of the analysis (Hu et al., 2014).  The data cleansing process depends on the complex relationship model, and it has extra computation and delay overhead (Hu et al., 2014).  Organizations must seek a balance between the complexity of the data-cleansing model and the resulting improvement in the accuracy analysis (Hu et al., 2014). 

The data redundancy is the third data pre-processing step where data is repeated increasing the overhead of the data transmission and causes limitawtions for storage systems, including wasted space, inconsistency of the data, corruption of the dta, and reduced reliability (Hu et al., 2014).  Various redundancy reduction methods include redundancy detection and data compression (Hu et al., 2014).  The data compression method poses an extra computation burden in the data compression and decompression processes (Hu et al., 2014).

Data Modeling and Design Consideration

Schema-on-Write system is used when the application or structure is well understood and frequently accessed through queries and reports on high-value data (Grover et al., 2015).        The term Schema-on-Read is used in the context of Hadoop data management system (Ankam, 2016; Grover et al., 2015). This term refers to the raw data, that is not processed and can be loaded to Hadoop using the required structure at processing time based on the requirement of the processing application (Ankam, 2016; Grover et al., 2015).  The Schema-on-Read is used when the application or structure of data is not well understood (Ankam, 2016; Grover et al., 2015).  The agility of the process is implemented through the schema-on-read providing valuable insights on data not previously accessible (Grover et al., 2015).

            Five factors must be considered before storing data into Hadoop for processing (Grover et al., 2015).  The data storage format must be considered as there are some file formats and compression formats supported on Hadoop.  Each type of format has strengths that make it better suited to specific applications.   Although Hadoop Distributed File System (HDFS) is a building block of Hadoop ecosystem, which is used for storing data, several commonly used systems implemented on top of HDFS such as HBase for traditional data access functionality, and Hive for additional data management functionality (Grover et al., 2015).  These systems of HBase for data access functionality and Hive for data management functionality must be taken into consideration before storing data into Hadoop (Grover et al., 2015). The second factor involves the multitenancy which is a common approach for clusters to host multiple users, groups and application types. The multi-tenant clusters involve essential considerations for data storage.  The schema design factor should also be considered before storing data into Hadoop even if Hadoop is a schema-less (Grover et al., 2015).  The schema design consideration involves directory structures for data loaded into HDFS and the output of the data processing and analysis, including the schema of objects stored in systems such as HBase and Hive.  The last factor for consideration before storing data into Hadoop is represented in the metadata management.  Metadata is related to the stored data and is often regarded as necessary as the data.  The understanding of the metadata management plays a significant role as it can affect the accessibility of the data.  The security is another factor which should be considered before storing data into Hadoop system.  The security of the data decision involves authentication, fine-grained access control, and encryption. These security measures should be considered for data at rest when it gets stored as well as in motion during the processing (Grover et al., 2015).  Figure 4 summarizes these considerations before storing data into the Hadoop system. 

Figure 4.  Considerations Before Storing Data into Hadoop.

Data Storage Format Considerations

            When architecting a solution on Hadoop, the method of storing the data into Hadoop is one of the essential decisions. Primary considerations for data storage in Hadoop involve file format, compression, data storage system (Grover et al., 2015).  The standard file formats involve three types:  text data, structured text data, and binary data.  Figure 5 summarizes these three standard file formats.

Figure 5.  Standard File Formats.

The text data is widespread use of Hadoop including log file such as weblogs, and server logs (Grover et al., 2015).  These text data format can come in many forms such as CSV files, or unstructured data such as emails.  Compression of the file is recommended, and the selection of the compression is influenced by how the data will be used (Grover et al., 2015).  For instance, if the data is for archival, the most compact compression method can be used, while if the data are used in processing jobs such as MapReduce, the splittable format should be used (Grover et al., 2015).  The splittable format enables Hadoop to split files into chunks for processing, which is essential to efficient parallel processing (Grover et al., 2015).

In most cases, the use of container formats such as SequenceFiles or Avro provides benefits making it the preferred format for most file system including text (Grover et al., 2015).  It is worth noting that these container formats provide functionality to support splittable compression among other benefits (Grover et al., 2015).   The binary data involves images which can be stored in Hadoop as well.  The container format such as SequenceFile is preferred when storing binary data in Hadoop.  If the binary data splittable unit is more than 64MB, the data should be put into its file, without using the container format (Grover et al., 2015).

XML and JSON Format Challenges with Hadoop

The structured text data include formats such as XML and JSON, which can present unique challenges using Hadoop because splitting XML and JSON files for processing is not straightforward, and Hadoop does not provide a built-in InputFormat for either (Grover et al., 2015).  JSON presents more challenges to Hadoop than XML because no token is available to mark the beginning or end of a record.  When using these file format, two primary consideration must be taken.  The container format such as Avro should be used because Avro provides a compact and efficient method to store and process the data when transforming the data into Avro (Grover et al., 2015).  A library for processing XML or JSON should be designed.  XMLLoader in PiggyBank library for Pig is an example when using XML data type.  The Elephant Bird project is an example of a JSON data type file (Grover et al., 2015). 

Hadoop File Types Considerations

            Several Hadoop-based file formats created to work well with MapReduce (Grover et al., 2015).  The Hadoop-specific file formats include file-based data structures such as sequence files, serialization formats like Avro, and columnar formats such as RCFile and Parquet (Grover et al., 2015).  These files types share two essential characteristics that are important for Hadoop application: splittable compression and agnostic compression.  The ability of splittable files play a significant role during the data processing, and should not be underestimated when storing data in Hadoop because it allows large files to be split for input to MapReduce and other types of jobs, which is a fundamental part of parallel processing and a key to leveraging data locality feature of Hadoop (Grover et al., 2015).  The agnostic compression is the ability to compress using any compression codec without readers having to know the codec because the codec is stored in the header metadata of the file format (Grover et al., 2015).  Figure 6 summarizes these Hadoop-specific file formats with the typical characteristics of splittable compression and agnostic compression.

Figure 6. Three Hadoop File Types with the Two Common Characteristics.  

1.      SequenceFiles Format Consideration

SequenceFiles format is the most widely used Hadoop file-based formats.  SequenceFile format store data as binary key-value pairs (Grover et al., 2015).  It involves three formats for records stored within SequenceFiles:  uncompressed, record-compressed, and block-compressed.  Every SequenceFile uses a standard header format containing necessary metadata about the file such as the compression codec used, key and value class names, user-defined metadata, and a randomly generated syn marker.  The SequenceFiles arewell supported in Hadoop. However, it has limited support outside the Hadoop ecosystem as it is only supported in Java language.  The frequent use case for SequenceFiles is a container for smaller files.  However, storing a large number of small files in Hadoop can cause memory issue and excessive overhead in processing.  Packing smaller files into a SequenceFile can make the storage and processing of these files more efficient because Hadoop is optimized for large files (Grover et al., 2015).   Other file-based formats include the MapFiles, SetFiles, Array-Files, and BloomMapFiles.  These formats offer a high level of integration for all forms of MapReduce jobs, including those run via Pig and Hive because they were designed to work with MapReduce (Grover et al., 2015).  Figure 7 summarizes the three formats for records stored within SequenceFiles.

Figure 7.  Three Formats for Records Stored within SequenceFile.

2.      Serialization Formats Consideration

Serialization is the process of moving data structures into bytes for storage or for transferring data over the network (Grover et al., 2015).   The de-serialization is the opposite process of converting a byte stream back into a data structure (Grover et al., 2015).  The serialization process is the fundamental building block for distributed processing systems such as Hadoop because it allows data to be converted into a format that can be efficiently stored and transferred across a network connection (Grover et al., 2015).  Figure 8 summarizes the serialization formats when architecting for Hadoop.

Figure 8.  Serialization Process vs. Deserialization Process.

The serialization involves two aspects of data processing in a distributed system of interprocess communication using data storage, and remote procedure calls or RPC (Grover et al., 2015).  Hadoop utilizes Writables as the main serialization format, which is compact and fast but uses Java only.  Other serialization frameworks have been increasingly used within Hadoop ecosystems, including Thrift, Protocol Buffers and Avro (Grover et al., 2015).  Avro is a language-neutral data serialization system (Grover et al., 2015).  It was designed to address the limitation of the Writables of Hadoop which is lack of language portability.  Similar to Thrift and Protocol Buffers, Avro is described through a language-independent schema (Grover et al., 2015).   Avro, unlike Thrift and Protocol Buffers, the code generation is optional.  Table 1 provides a comparison between these serialization formats.

Table 1:  Comparison between Serialization Formats.

3.      Columnar Format Consideration

Row-oriented systems have been used to fetch data stored in the database (Grover et al., 2015).  This type of data retrieval has been used as the analysis heavily relied on fetching all fields for records that belonged to a specific time range.  This process is efficient if all columns of the record are available at the time or writing because the record can be written with a single disk seek.  The column storage has recently been used to fetch data.  The use of columnar storage has four main benefits over the row-oriented system (Grover et al., 2015).  The skips I/O and decompression on columns that are not part of the query is one of the benefits of the columnar storage.  Columnar data storage works better for queries that access a small subset of columns than the row-oriented data storage, which can be used when many columns are retrieved.  The compression on columns provides efficiency because data is more similar within the same column than it is in a block of rows.  The columnar data storage is more appropriate for data warehousing-based applications where aggregations are implemented using specific columns than an extensive collection of records (Grover et al., 2015).  Hadoop applications have been using the columnar file formats including the RCFile format, Optimized Row Columnar (ORC), and Parquet.  The RCFile format has been used as a Hive Format.  It was developed to provide fast data loading, fast query processing, and highly efficient storage space utilization.  It breaks files into row splits, and within each split uses columnar-oriented storage.  Despite its advantages of the query and compression performance compared to SequenceFiles, it has limitations, that prevent the optimal performance for query times and compression.  The newer version of the columnar formats ORC and Parquet are designed to address many of the limitations of the RCFile (Grover et al., 2015). 

Compression Consideration

Compression is another data storage consideration because it plays a crucial role in reducing the storage requirements, and in improving the data processing performance (Grover et al., 2015).  Some compression formats supported on Hadoop are not splittable (Grover et al., 2015).  MapReduce framework splits data for input to multiple tasks; the nonsplittable compression format is an obstacle to efficient processing.  Thus, the splittability is a critical consideration in selecting the compression format and file format for Hadoop.  Various compression types for Hadoop include Snappy, LZO, Gzip, bzip2.  Google developed Snappy for speed. However, it does not offer the best compression size. It is designed to be used with a container format like SequenceFile or Avro because it is not inherently splittable.  It is being distributed with Hadoop. Similar to Snappy, LZO is optimized for speed as opposed to size.  However, LZO, unlike Snappy support splittability of the compressed files, but it requires indexing.  LZO, unlike Snappy, is not distributed with Hadoop and requires a license and separate installation.  Gzip, like Snappy, provides good compression performance, but is not splittable, and it should be used with a container format. The speed read performance of the Gzip is like the Snappy.  Gzip is slower than Snappy for write processing.  Gzip is not splittable and should be used with a container format.  The use of smaller blocks with Gzip can result in better performance.   The bzip2 is another compression type for Hadoop.  It provides good compression performance, but it can be slower than another compression codec such as Snappy.  It is not an ideal codec for Hadoop storage. Bzip2, unlike Snappy and Gzip, is inherently splittable.  It inserts synchronization markers between blocks.  It can be used for active archival purposes (Grover et al., 2015).

The compression format can become splittable when used with container file formats such as Avro, SequenceFile which compress blocks of records or each record individually (Grover et al., 2015).  If the compression is implemented on the entire file without using the container file format, the compression format that inherently supports splittable must be used such as bzip2.  The compression use with Hadoop has three recommendation (Grover et al., 2015).  The first recommendation is to enable compression of MapReduce intermediate output, which improves performance by decreasing the among of intermediate data that needs to be read and written from and to disk.  The second recommendation s to pay attention to the order of the data.  When the data is close together, it provides better compression levels. The data in Hadoop file format is compressed in chunks, and the organization of those chunks determines the final compression.   The last recommendation is to consider the use of a compact file format with support for splittable compression such as Avro.  Avro and SequenceFiles support splittability with non-splittable compression formats.  A single HDFS block can contain multiple Avro or SequenceFile blocks. Each block of the Avro or SequenceFile can be compressed and decompressed individually and independently of any other blocks of Avro or SequenceFile. This technique makes the data splittable because each block can be compressed and decompressed individually.  Figure 9 shows the Avro and SequenceFile splittability support (Grover et al., 2015).  

Figure 9.  Compression Example Using Avro (Grover et al., 2015).

Design Consideration for HDFS Schema

HDFS and HBase are the commonly used storage managers in the Hadoop ecosystem.  Organizations can store the data in HDFS or HBase which internally store it on HDFS (Grover et al., 2015).  When storing data in HDFS, some design techniques must be taken into consideration.  The schema-on-read model of Hadoop does not impose any requirement when loading data into Hadoop, as data can be ingested into HDFS by one of many methods without the requirements to associate a schema or preprocess the data.  Although Hadoop has been used to load many types of data such as the unstructured data, semi-structured data, some order is still required, because Hadoop serves as a central location for the entire organization and the data stored in HDFS is intended to be shared across various departments and teams in the organization (Grover et al., 2015).  The data repository should be carefully structured and organized to provide various benefits to the organization  (Grover et al., 2015).   When there is a standard directory structure, it becomes easier to share data among teams working with the same data set.  The data gets staged in a separate location before processing it.  The standard stage technique can help not processing data that has not been appropriately staged or entirely yet.  The standard organization of data allows for some code reuse that may process the data (Grover et al., 2015).  The placement of data assumptions can help simplify the loading process of the data into Hadoop.   The HDFS data model design for projects such as data warehouse implementation is likely to use structure facts and dimension tables similar to the traditional schema  (Grover et al., 2015).  The HDFS data model design for projects of unstructured and semi-structured data is likely to focus on directory placement and metadata management (Grover et al., 2015). 

Grover et al. (2015) suggested three key considerations when designing the schema, regardless of the data model design project.  The first consideration is to develop standard practices that can be followed by all teams.  The second point is to ensure the design works well with the chosen tools.  For instance, if the version of Hive can support only table partitions on directories that are named a certain way, it will affect the schema design and the names of the table subdirectories.  The last consideration when designing a schema is to keep usage patterns in mind, because different data processing and querying patterns work better with different schema designs (Grover et al., 2015). 

HDFS Files Location Consideration

            The first step when designing an HDFS schema involves the determination of the location of the file.  Standard file location plays a significant role in finding and sharing data among various departments and teams. It also helps in the assignment of permission to access files to various groups and users.  The recommended file locations are summarized in Table 2.

Table 2.  Standard Files Locations.

HDFS Schema Design Consideration

The HDFS schema design involves advanced techniques to organize data into files (Grover et al., 2015).   A few strategies are recommended to organize the data set. These strategies for data organization involve partitioning, bucketing, and denormalizing process.  The partitioning process of the data set is a common technique used to reduce the amount of I/O required to process the data set.  HDFS does not store indexes on the data unlike the traditional data warehouse. Such a lack of indexes in HDFS plays a key role in speeding up data ingest, with a full table scan cost where every query will have to read the entire dataset even when processing a small subset of data. Breaking up the data set into smaller subsets, or partitions can help with the full table scan, allowing queries to read only the specific partitions reducing the amount of I/O and improving the query time processing significantly (Grover et al., 2015). When data is placed in the filesystem, the directory format for partition should be as shown below.  The order data sets are partitioned by date because there are a large number of orders done daily and the partitions will contain large enough files which are optimized by HDFS.  Various tools such as HCatalog, Hive, Impala, and Pig understand this directory structure leveraging the partitioning to reduce the amount of I/O requiring during the data processing (Grover et al., 2015).

• <data set name>/<partition_column_name=partition_column_value>/(Armstrong)
• e.g. medication_orders/date=20181107/[order1.csv, order2.csv]

Bucketing is another technique for breaking a large data set into manageable sub-sets (Grover et al., 2015).  The bucketing technique is similar to the hash partitions which is used in the relational database.   Various tools such as HCatalog, Hive, Impala, and Pig understand this directory structure leveraging the partitioning to reduce the amount of I/O requiring during the data processing. The partition example above was implemented using the date which resulted in large data files which can be optimized by HDFS (Grover et al., 2015).  However, if the data sets are partitioned by a the category of the physician, the result will be too many small files, which leads to small file problems, which can lead to excessive memory use for the NameNode, since metadata for each file stored in HDFS is stored in memory (Grover et al., 2015).  Many small files can also lead to many processing tasks, causing excessive overhead in processing.  The solution for too many small files is to use the bucketing process for the physician in this example, which uses the hashing function to map physician into a specified number of buckets (Grover et al., 2015).

The bucketing technique controls the size of the data subsets and optimizes the query speed (Grover et al., 2015).  The recommended average bucket size is a few multiples of the HDFS block size. The distribution of data when hashed on the bucketing column is essential because it results in consistent bucketing (Grover et al., 2015).  The use of the number of buckets as a power of two is every day.   Bucketing allows joining two data sets.  The join, in this case, is used to represent the general idea of combining two data sets to retrieve a result. The joins can be implemented through the SQL-on-Hadoop systems and also in MapReduce, or Spark, or other programming interfaces to Hadoop.  When using join in the bucketing technique, it joins corresponding buckets individually without having to join the entire datasets, which help in minimizing the time complexity for the reduce-side join of the two datasets process, which is computationally expensive (Grover et al., 2015).   The join is implemented in the map stage of a MapReduce job by loading the smaller of the buckets in memory because the buckets are small enough to easily fit into memory, which is called map-side join process.  The map-side join process improves the join performance as compared to a reduce-side join process.  A hive for data analysis recognizes the tables are bucketed and optimize the process accordingly.

Further optimization can be implemented if the data in the bucket is sorted, the merge join can be used, and the entire bucket does not get stored in memory when joining, resulting in the faster process and much less memory than a simple bucket join.  Hive supports this optimization as well.  The use of both sorting and bucketing on large tables that are frequently joined together using the join key for bucketing is recommended (Grover et al., 2015).

The schema design depends on how the data will be queried (Grover et al., 2015).  Thus, the columns to be used for joining and filtering must be identified before the portioning and bucketing of the data is implemented.   In some cases, when the identification of one partitioning key is challenging, storing the same data set multiple times can be implemented, each with the different physical organization, which is regarded to be an anti-pattern in a relational database.  However, this solution can be implemented with Hadoop, because with Hadoop is write-once, and few updates are expected.  Thus, the overhead of keeping duplicated data set in sync is reduced.  The cost of storage in Hadoop clusters is reduced as well  (Grover et al., 2015). The duplicated data set in sync provides better query speed processing in such cases (Grover et al., 2015). 

Regarding the denormalizing process, it is another technique of trading the disk space for query performance, where joining the entire data set need is minimized (Grover et al., 2015).   In the relational database model, the data is stored in the third standard form (NF3), where redundancy is minimized, and data integrity is enforced by splitting data into smaller tables, each holding a particular entity.  In this relational model, most queries require joining a large number of tables together to produce a final result as desired (Grover et al., 2015).  However, in Hadoop, joins are often the slowest operations and consume the most resources from the cluster.  Specifically, the reduce-side join requires sending the entire table over the network, which is computationally costly.  While sorting and bucketing help minimizing this computational cost, another solution is to create data sets that are pre-joined or pre-aggregated (Grover et al., 2015).  Thus, the data can be joined once and store it in this form instead of running the join operations every time there is a query for that data.  Hadoop schema consolidates many of the small dimension tables into a few larger dimensions by joining them during the ETL process  (Grover et al., 2015).  Other techniques to speed up the process include the aggregation or data type conversion.  The duplication of the data is of less concern; thus, when the processing is frequent for a large number of queries, it is recommended to doing it one and reuse as the case with a materialized view in the relational database.  In Hadoop, the new dataset is created that contains the same data in its aggregated form (Grover et al., 2015).

To summarize, the partitioning process is used to reduce the I/O overhead of processing by selectively reading and writing data in particular partitions.  The bucketing can be used to speed up queries that involve joins or sampling, by reducing the I/O as well.  The denormalization can be implemented to speed up Hadoop jobs.   In this section, a review of advanced techniques to organize data into files is discussed.  The discussion includes the use of a small number of large files versus a large number of small files.  Hadoop prefers working with a small number of large files than a large number of small files.  The discussion also addresses the reduce-side join versus map-side join techniques.   The reduce-side join is computationally costly. Hence, the map-side join technique is preferred and recommend. 

HBase Schema Design Consideration

HBase is not a relational database (Grover et al., 2015; Yang, Liu, Hsu, Lu, & Chu, 2013).  HBase is similar to a large hash table, which allows the association of values with keys and performs a fast lookup of the value based on a given key  (Grover et al., 2015). The operations of hash tables involve put, get, scan, increment and delete.  HBase provides scalability and flexibility and is useful in many applications, including fraud detection, which is a widespread application for HBase (Grover et al., 2015).

The framework of HBase involves Master Server, Region Servers, Write-Ahead Log (WAL), Memstore, HFile, API and Hadoop HDFS (Bhojwani & Shah, 2016).  Each component of the HBase framework plays a significant role in data storage and processing.  Figure 10 illustrated the HBase framework.

Figure 10.  HBase Architecture (Bhojwani & Shah, 2016).

            The following consideration must be taken when designing the schema for HBase (Grover et al., 2015).

• Row Key Consideration.
• Timestamp Consideration.
• Hops Consideration.
• Tables and Regions Consideration.
• Columns Use Consideration.
• Column Families Use Consideration.
• Time-To-Live Consideration.

The row key is one of the most critical factors for well-architected HBase schema design (Grover et al., 2015).  The row key consideration involves record retrieval, distribution, block cache, the ability to scan, size, readability, and uniqueness.  The row key is critical for retrieving records from HBase. In the relational database, the composite key can be used to combine multiple primary keys.  In HBase, multiple pieces of information can be combined in a single key.  For instance, a key of customer_id, order_id, and timestamp will be a row key for a row describing an order. In a relational database, they are three different columns in the relational database, but in HBase, they will be combined into a single unique identifier.  Another consideration for selecting the row key is the get operation because a get operation of a single record is the fasted operation in HBase.  A single get operation can retrieve the most common uses of the data improves the performance, which requires to put much information in a single record which is called denormalized design.    For instance, while in the relational database, customer information will be placed in various tables, in HBase all customer information will be stored in a single record where get operation will be used. The distribution is another consideration for HBase schema design.  The row key determines the regions of HBase cluster for a given table, which will be scattered throughout various regions (Grover et al., 2015; Yang et al., 2013).   The row keys are sorted, and each region stores a range of these sorted row keys  (Grover et al., 2015).  Each region is pinned to a region server namely a node in the cluster  (Grover et al., 2015).  The combination of device ID and timestamp or reverse timestamp is commonly used to “salt” the key in machine data  (Grover et al., 2015).  The block cache is a least recently used (LRU) cache which caches data blocks in memory  (Grover et al., 2015).  HBase reads records in chunks of 64KB from the disk by default. Each of these chunks is called HBase block  (Grover et al., 2015).  When the HBase block is read from disk, it will be put into the block cache  (Grover et al., 2015).   The choice of the row key can affect the scan operation as well.  HBase scan rates are about eight times slower than HSFS scan rates.  Thus, reducing I/O requirements has a significant performance advantage.  The size of the row key determines the performance of the workload.  The short row key is better than, the long row key because it has lower storage overhead and faster read/ writes performance.  The readability of the row key is critical. Thus, it is essential to start with human-readable row key.  The uniqueness of the row key is also critical since a row key is equivalent to a key in hash table analogy.  If the row key is based on the non-unique attribute, the application should handle such cases and only put data in HBase with a unique row key (Grover et al., 2015).

The timestamp is the second essential consideration for good HBase schema design (Grover et al., 2015).  The timestamp provides advantages of determining which records are newer in case of put operation to modify the record.  It also determines the order where records are returned when multiple versions of a single record are requested. The timestamp is also utilized to remove out-of-date records because time-to-live (TTL) operation compared with the timestamp shows the record value has either been overwritten by another put or deleted (Grover et al., 2015).

The hop term refers to the number of synchronized “get” requests to retrieve specific data from HBase (Grover et al., 2015). The less hop, the better because of the overhead.  Although multi-hop requests with HBase can be made, it is best to avoid them through better schema design, for example by leveraging de-normalization, because every hop is a round-trip to HBase which has a significant performance overhead (Grover et al., 2015).

The number of tables and regions per table in HBase can have a negative impact on the performance and distribution of the data (Grover et al., 2015).  If the number of tables and regions are not implemented correctly, it can result in an imbalance in the distribution of the load.  Important considerations include one region server per node, many regions in a region server, a give region is pinned to a particular region server, and tables are split into regions and scattered across region servers.  A table must have at least one region.  All regions in a region server receive “put” requests and share the region server’s “memstore,” which is a cache structure present on every HBase region server. The “memstore” caches the write is sent to that region server and sorts them in before it flushes them when certain memory thresholds are reached. Thus, the more regions exist in a region server; the less memstore space is available per region.  The default configuration sets the ideal flush size to 100MB. Thus, the “memstore” size can be divided by 100MB and result should be the maximum number of regions which can be put on that region server.   The vast region takes a long time to compact.  The upper limit on the size of a region is around 20GB. However, there are successful HBase clusters with upward of 120GB regions.  The regions can be assigned to HBase table using one of two techniques. The first technique is to create the table with a single default region, which auto splits as data increases.  The second technique is to create the table with a given number of regions and set the region size to a high enough value, e.g., 100GB per region to avoid auto splitting (Grover et al., 2015).  Figure 11 shows a topology of region servers, regions and tables. 

Figure 11.  The Topology of Region Servers, Regions, and Tables (Grover et al., 2015).

The columns used in HBase is different from the traditional relational database (Grover et al., 2015; Yang et al., 2013).  In HBase, unlike the traditional database, a record can have a million columns, and the next record can have a million completely different columns, which is not recommended but possible (Grover et al., 2015).   HBase stores data in a format called HFile, where each column value gets its row in HFile (Grover et al., 2015; Yang et al., 2013). The row has files like row key, timestamp, column names, and values. The file format provides various functionality, like versioning and sparse column storage (Grover et al., 2015). 

HBase, include the concept of column families (Grover et al., 2015; Yang et al., 2013).  A column family is a container for columns.  In HBase, a table can have one or more column families.  Each column family has its set of HFiles and gets compacted independently of other column families in the same table.  In many cases, no more than one column family is needed per table.  The use of more than one column family per table can be done when the operation is done, or the rate of change on a subset of the columns of a table is different from the other columns (Grover et al., 2015; Yang et al., 2013).  The last consideration for HBase schema design is the use of TTL, which is a built-in feature of HBase which ages out data based on its timestamp (Grover et al., 2015).  If TTL is not used and an aging requirement is needed, then a much more I/O intensive operation would need to be done.   The objects in HBase begin with table object, followed by regions for the table, store per column family for each region for the table, memstore, store files, and block (Yang et al., 2013).  Figure 12 shows the hierarchy of objects in HBase.

Figure 12.  The Hierarchy of Objects in HBase (Yang et al., 2013).

To summarize this section, HBase schema design requires seven key consideration starting with the row key, which should be selected carefully for record retrieval, distribution, block cache, ability to scan, size, readability, and uniqueness.  The timestamp and hops are other schema design consideration for HBase.  Tables and regions must be considered for put performance, and compacting time.  The use of columns and column families should also be considered when designing the schema for HBase. The TTL to remove data that aged is another consideration for HBase schema design. 

Metadata Consideration

The above discussion has been about the data and the techniques to store it in Hadoop.  Metadata is as essential as the data itself.  Metadata is data about the data (Grover et al., 2015)).  Hadoop ecosystem has various forms of metadata.   Metadata about logical dataset usually stored in a separate metadata repository include the information like the location of a data set such as directory in HDFS or HBase table name, the schema associated with the dataset, the partitioning and sorting properties of the data set, the format of the data set e.g. CSV, SequenceFile, etc. (Grover et al., 2015). The metadata about files on HDFS includes the permission and ownership of such files and the location of various blocks on data nodes, usually stored and managed by Hadoop NameNode (Grover et al., 2015).  Metadata about tables in HBase include information like table names, associated namespace, associated attributes, e.g. MAX_FILESIZE, READONLY, etc., and the names of column families, usually stored and managed by HBase (Grover et al., 2015).  Metadata about data ingest and transformation include information like which user-generated a given dataset, where the dataset came from, how long it took to generate it, and how many records there are, or the size of the data load (Grover et al., 2015).  Metadata about dataset statistics include information like the number of rows in a dataset, number of unique values in each column, a histogram of the distribution of the data, and maximum and minimum values (Grover et al., 2015).  Figure 13 summarizes this various metadata.

Figure 13.  Various Metadata in Hadoop.

Apache Hive was the first project in the Hadoop ecosystem to store, manage and leverage metadata (Antony et al., 2016; Grover et al., 2015).  Hives stores this metadata in a relational database called the Hive “metastore” (Antony et al., 2016; Grover et al., 2015).  Hive also provides a “metastore” service which interfaces with the Hive metastore database (Antony et al., 2016; Grover et al., 2015).  The query process in Hive goes to the metastore to get the metadata for the desired query, and metastore sends the metadata to Hive generating execution plan, followed by executing the job using the Hadoop cluster, which implements the job and Hive send the fetched result to the user (Antony et al., 2016; Grover et al., 2015).  Figure 14 shows the query process and the role of the metastore in Hive framework.

Figure 14.  Query Process and the Role of Metastore in Hive (Antony et al., 2016).

More projects have utilized the concept of metadata that was introduced by Hive and created a separate project called HCatalog to enable the usage of Hive metastore outside of Hive (Grover et al., 2015).  HCatalog is a part of Hive and allows other tools like Pig and MapReduce to integrate with Hive metastore.  It also opens the access to Hive metastore to other tools such as REST API via WebHCat server.  MapReduce, Pig, and standalone applications can talk directly to the metastore of Hive through its APIs, but HCatalog allows easy access through its WebHCat REST APIs, and it allows the cluster administrators to lock down access to the Hive metastore to address security concerns. Other ways to store metadata include the embedding of metadata in file paths and names.  Another technique to store metadata involves storing it in HDFS in a hidden file, e.g., .metadata.  Figure 15 shows the HCatalog as an accessibility veneer around the Hive metastore (Grover et al., 2015). 

Figure 15.  HCatalog acts an accessibility veneer around the Hive metastore (Grover et al., 2015).

Hive Metastore and HCatalog Limitations

There are some limitations for Hive metastore and HCatalog, including the problem with high availability (Grover et al., 2015).  The HA database cluster solutions to bring HA to the Hive metastore database.  For the metastore service of Hive, there is support concurrently to run multiple metastores on more than one node in the cluster.  However, concurrency issues related to data definition language operations (DDL) can occur, and Hive community is working on fixing these issues. 

The fixed schema for metadata is another limitation.  Hadoop provides much flexibility on the type of data that can be stored, mainly because of the Schema-on-Read concept. Hive metastore provides a fixed schema for the metadata itself. It provides a tabular abstraction for the data sets.   The data in metastore is moving the part in the infrastructure which requires to be running and secured as part of Hadoop infrastructure (Grover et al., 2015).

Conclusion

This project has discussed essential topics related to Hadoop technology.  It began with an overview of Hadoop providing a history of Hadoop and the difference between Hadoop 1.x and Hadoop 2.x.  The discussion involved the Big Data Analytics Process using Hadoop technology.  The process involves six significant steps starting with the problem identification, required data to be collected, and the data collection process. The pre-processing data and ETL process must be implemented before performing the analytics. The last step is the visualization of the data for decision making.  Before processing any data and before collecting any data for storage, some considerations must be taken for data preprocessing, modeling and schema design in Hadoop for better processing and better data retrieval, giving some tools cannot split the data while others can.  These considerations begin with data storage format, followed by Hadoop file types consideration and XML and JSON format challenges in Hadoop.  Compression must be considered when designing the schema for Hadoop. Since HDFS and HBase are commonly used in Hadoop for data storage, the discussion involved the consideration for the HDFS and HBase schema design considerations.  To summarize the design of the schema for Hadoop, HDFS, and HBase makes a difference in storing data in various nodes using the right tools for splitting the data.  Thus, organizations must pay attention to the process and the design requirements before storing data into Hadoop for better computational processing. 

References

Alguliyev, R., & Imamverdiyev, Y. (2014). Big data: big promises for information security. Paper presented at the Application of Information and Communication Technologies (AICT), 2014 IEEE 8th International Conference on.

Ankam, V. (2016). Big Data Analytics: Packt Publishing Ltd.

Antony, B., Boudnik, K., Adams, C., Lee, C., Shao, B., & Sasaki, K. (2016). Professional Hadoop: John Wiley & Sons.

Armstrong, D. (n.d.). R: Learning by Example: Lattice Graphics. Retrieved from https://quantoid.net/files/rbe/lattice.pdf.

Bhojwani, N., & Shah, A. P. V. (2016). A SURVEY ON HADOOP HBASE SYSTEM. Development, 3(1).

Dittrich, J., & Quiané-Ruiz, J.-A. (2012). Efficient big data processing in Hadoop MapReduce. Proceedings of the VLDB Endowment, 5(12), 2014-2015.

Grover, M., Malaska, T., Seidman, J., & Shapira, G. (2015). Hadoop Application Architectures: Designing Real-World Big Data Applications: ” O’Reilly Media, Inc.”.

Hu, H., Wen, Y., Chua, T.-S., & Li, X. (2014). Toward scalable systems for big data analytics: A technology tutorial. IEEE Access, 2, 652-687.

Karanth, S. (2014). Mastering Hadoop: Packt Publishing Ltd.

Lublinsky, B., Smith, K. T., & Yakubovich, A. (2013). Professional hadoop solutions: John Wiley & Sons.

sas.com. (2018). Hadoop – why it is and why it matters. Retrieved from https://www.sas.com/en_us/insights/big-data/hadoop.html.

Yang, C. T., Liu, J. C., Hsu, W. H., Lu, H. W., & Chu, W. C. C. (2013, 16-18 Dec. 2013). Implementation of Data Transform Method into NoSQL Database for Healthcare Data. Paper presented at the 2013 International Conference on Parallel and Distributed Computing, Applications and Technologies.

 

Abstract

The purpose of this project is to discuss how data can be handled before Hadoop can take action on breaking data into manageable sizes.  The discussion begins with an overview of Hadoop providing a brief history of Hadoop and the difference between Hadoop 1.x and Hadoop 2.x. The discussion involves the Big Data Analytics process using Hadoop which involves six significant steps including the pre-processing data and ETL process where the data must be converted and cleaned before processing it.  Before data processing, some consideration must be taken for data preprocessing, modeling and schema design in Hadoop for better processing and data retrieval as it will affect how data can be split among various nodes in the distributed environment because not all tools can split the data.  This consideration begins with the data storage format, followed by Hadoop file types consideration and XML and JSON format challenges in Hadoop.  The compression of the data must be considered carefully because not all compression types are “splittable.” The discussion also involves the schema design consideration for HDFS and HBase since they are used often in the Hadoop ecosystem. 

Keywords: Big Data Analytics; Hadoop; Data Modelling in Hadoop; Schema Design in Hadoop.

Introduction

In the age of Big Data, dealing with large datasets in terabytes and petabytes is a reality and requires specific technology as the traditional technology was found inappropriate for it (Dittrich & Quiané-Ruiz, 2012).  Hadoop is developed to store, and process such large datasets efficiently.  Hadoop is becoming a data processing engine for Big Data (Dittrich & Quiané-Ruiz, 2012).  One of the significant advantages of Hadoop MapReduce is allowing non-expert users to run easily analytical tasks over Big Data (Dittrich & Quiané-Ruiz, 2012). However, before the analytical process takes place, some schema design and data modeling consideration must be taken for Hadoop so that the data process can be efficient (Grover, Malaska, Seidman, & Shapira, 2015).  Hadoop requires splitting the data. Some tools can split the data while others cannot split the data natively and requires integration (Grover et al., 2015). 

This project discusses these considerations to ensure the appropriate schema design for Hadoop and its components of HDFS, HBase where the data gets stored in a distributed environment.   The discussion begins with an overview of Hadoop first, followed by the data analytics process and ends with the data modeling techniques and consideration for Hadoop which can assist in splitting the data appropriately for better data processing performance and better data retrieval.

Overview of Hadoop

            Google published and disclosed its MapReduce technique and implementation early around 2004 (Karanth, 2014).  It also introduced the Google File System (GFS) which is associated with MapReduce implementation.  The MapReduce, since then, has become the most common technique to process massive data sets in parallel and distributed settings across many companies (Karanth, 2014).  In 2008, Yahoo released Hadoop as an open-source implementation of the MapReduce framework (Karanth, 2014; sas.com, 2018). Hadoop and its file system HDFS are inspired by Google’s MapReduce and GFS (Ankam, 2016; Karanth, 2014).  

The Apache Hadoop is the parent project for all subsequence projects of Hadoop (Karanth, 2014).  It contains three essential branches 0.20.1 branch, 0.20.2 branch, and 0.21 branch.  The 0.20.2 branch is often termed MapReduce v2.0, MRv2, or Hadoop 2.0.  Two additional releases for Hadoop involves the Hadoop-0.20-append and Hadoop-0.20-Security, introducing HDFS append and security-related features into Hadoop respectively.  The timeline for Hadoop technology is outlined in Figure 1.

Figure 1.  Hadoop Timeline from 2003 until 2013 (Karanth, 2014).

Hadoop version 1.0 was the inception and evolution of Hadoop as a simple MapReduce job-processing framework (Karanth, 2014).  It exceeded its expectations with wide adoption of massive data processing.  The stable version of the 1.x release includes features such as append and security.  Hadoop version 2.0 release came out in 2013 to increase efficiency and mileage from existing Hadoop clusters in enterprises.  Hadoop is becoming a common cluster-computing and storage platform from being limited to MapReduce only, because it has been moving faster than MapReduce to stay leading in massive scale data processing with the challenge of being backward compatible (Karanth, 2014). 

            In Hadoop 1.x, the JobTracker was responsible for the resource allocation and job execution (Karanth, 2014).  MapReduce was the only supported model since the computing model was tied to the resources in the cluster. The yet another resource negotiator (YARN) was developed to separate concerns relating to resource management and application execution, which enables other application paradigms to be added into Hadoop computing cluster. The support for diverse applications result in the efficient and effective utilization of the resources and integrates well with the infrastructure of the business (Karanth, 2014).  YARN maintains backward compatibility with Hadoop version 1.x APIs  (Karanth, 2014).  Thus, the old MapReduce program can still execute in YARN with no code changes, but it has to be recompiled (Karanth, 2014).

            YARN abstracts out the resource management functions to form a platform layer called ResourceManager (RM) (Karanth, 2014).  Every cluster must have RM to keep track of cluster resource usage and activity.  RM is also responsible for allocation of the resources and resolving contentions among resource seekers in the cluster.  RM utilizes a generalized resource model and is agnostic to application-specific resource needs.  RM does not need to know the resources corresponding to a single Map or Reduce slot (Karanth, 2014). Figure 2 shows Hadoop 1.x and Hadoop 2.x with YARN layer.   

Figure 2. Hadoop 1.x vs. Hadoop 2.x (Karanth, 2014).

Hadoop 2.x involves various enhancement at the storage layer as well.   These enhancements include the high availability feature to have a hot standby of NameNode (Karanth, 2014), when the active NameNode fails, the standby can become active NameNode in a matter of minutes.  The Zookeeper or any other HA monitoring service can be utilized to track NameNode failure (Karanth, 2014).  The failover process to promote the hot standby as the active NameNode is triggered with the assistance of the Zookeeper.  The HDFS federation is another enhancement in Hadoop 2.x, which is a more generalized storage model, where the block storage has been generalized and separated from the filesystem layer (Karanth, 2014).  The HDFS snapshots is another enhancement to the Hadoop 2.x which provides a read-only image of the entire or a particular subset of a filesystem to protect against user errors, backup, and disaster recovery.   Other enhancements added in Hadoop 2.x include the Protocol Buffers (Karanth, 2014). The wire protocol for RPCs within Hadoop is based on Protocol Buffers.  Hadoop 2.x is aware of the type of storage and expose this information to the application, to optimize data fetch and placement strategies (Karanth, 2014).  HDFS append support has been another enhancement in Hadoop 2.x.

Hadoop is regarded to be the de facto open-source framework for dealing with large-scale, massively parallel, and distributed data processing (Karanth, 2014).  The framework of Hadoop includes two layers for computation and data layer (Karanth, 2014).  The computation layer is used for parallel and distributed computation processing, while the data layer is used for a highly fault-tolerant data storage layer which is associated with the computation layer.  These two layers run on commodity hardware, which is not expensive, readily available, and compatible with other similar hardware (Karanth, 2014).

Hadoop Architecture

Apache Hadoop has four projects: Hadoop Common, Hadoop Distributed File System, Yet Another Resource Negotiator (YARN), and MapReduce (Ankam, 2016).  The HDFS is used to store data, MapReduce is used to process data, and YARN is used to manage the resources such as CPU and memory of the cluster and common utilities that support Hadoop framework (Ankam, 2016; Karanth, 2014).  Apache Hadoop integrates with other tools such as Avro, Hive, Pig, HBase, Zookeeper, and Apache Spark (Ankam, 2016; Karanth, 2014).

            Hadoop three significant components for Big Data Analytics.  The HDFS is a framework for reliable distributed data storage (Ankam, 2016; Karanth, 2014).  Some considerations must be taken when storing data into HDFS (Grover et al., 2015).  The multiple frameworks for parallel processing of data include MapReduce, Crunch, Cascading, Hive, Tez, Impala, Pig, Mahout, Spark, and Giraph (Ankam, 2016; Karanth, 2014). The Hadoop architecture includes NameNodes and DataNodes.  It also includes Oozie for workflow, Pig for scripting, Mahout for machine learning, Hive for the data warehouse.  Sqoop for data exchange, and Flume for log collection.  YARN is in Hadoop 2.0 as discussed earlier for distributed computing, while HCatalog for Hadoop metadata management.  HBase is for columnar database and Zookeeper for coordination (Alguliyev & Imamverdiyev, 2014).  Figure 3 shows the Hadoop ecosystem components.

Figure 3.  Hadoop Architecture (Alguliyev & Imamverdiyev, 2014).

Big Data Analytics Process Using Hadoop

The process of Big Data Analytics involves six essential steps (Ankam, 2016). The identification of the business problem and outcomes is the first step.  Examples of business problems include sales are going down, or shopping carts are abandoned by customers, a sudden rise in the call volumes, and so forth.  Examples of the outcome include improving the buying rate by 10%, decreasing shopping cart abandonment by 50%, and reducing call volume by 50% by next quarter while keeping customers happy.  The required data must be identified where data sources can be data warehouse using online analytical processing, application database using online transactional processing, log files from servers, documents from the internet, sensor-generated data, and so forth, based on the case and the problem.  Data collection is the third step in analyzing the Big Data (Ankam, 2016).  Sqoop tool can be used to collect data from the relational database, and Flume can be used for stream data.  Apache Kafka can be used for reliable intermediate storage.  The data collection and design should be implemented using the fault tolerance strategy (Ankam, 2016).  The preprocessing data and ETL process is the fourth step in the analytical process.  The collected data comes in various formats, and the data quality can be an issue. Thus, before processing it, it needs to be converted to the required format and cleaned from inconsistent, invalid or corrupted data.  Apache Hive, Apache Pig, and Spark SQL can be used for preprocessing massive amounts of data.  The analytics implementation is the fifth steps which should be in order to answer the business questions and problems. The analytical process requires understanding the data and relationships between data points.  The types of data analytics include descriptive and diagnostic analytics to present the past and current views of the data, to answer questions such as what and why happened.  The predictive analytics is performed to answer questions such as what would happen based on a hypothesis. Apache Hive, Pig, Impala, Drill, Tez, Apache Spark, and HBase can be used for data analytics in batch processing mode.  Real-time analytics tools including Impala, Tez, Drill, and Spark SQL can be integrated into the traditional business intelligence (BI) using any of BI tools such as Tableau, QlikView, and others for interactive analytics. The last step in this process involves the visualization of the data to present the analytics output in a graphical or pictorial format to understand the analysis better for decision making.  The finished data is exported from Hadoop to a relational database using Sqoop, for integration into visualization systems or visualizing systems are directly integrated into tools such as Tableau, QlikView, Excel, and so forth.  Web-based notebooks such as Jupyter, Zeppelin, and Data bricks cloud are also used to visualize data by integrating Hadoop and Spark components (Ankam, 2016). 

Data Preprocessing, Modeling and Design Consideration in Hadoop

            Before processing any data, and before collecting any data for storage, some considerations must be taken for data modeling and design in Hadoop for better processing and better retrieval (Grover et al., 2015).  The traditional data management system is referred to as Schema-on-Write system which requires the definition of the schema of the data store before the data is loaded (Grover et al., 2015).  This traditional data management system results in long analysis cycles, data modeling, data transformation loading, testing, and so forth before the data can be accessed (Grover et al., 2015).   In addition to this long analysis cycle, if anything changes or wrong decision was made, the cycle must start from the beginning which will take longer time for processing (Grover et al., 2015).   This section addresses various types of consideration before processing the data from Hadoop for analytical purpose.

Data Pre-Processing Consideration

The dataset may have various levels of quality regarding noise, redundancy, and consistency (Hu, Wen, Chua, & Li, 2014).  Preprocessing techniques must be used to improve data quality should be in place in Big Data systems (Hu et al., 2014; Lublinsky, Smith, & Yakubovich, 2013).  The data pre-processing involves three techniques: data integration, data cleansing, and redundancy elimination.

The data integration techniques are used to combine data residing in different sources and provide users with a unified view of the data (Hu et al., 2014).  The traditional database approach has well-established data integration system including the data warehouse method, and the data federation method (Hu et al., 2014).  The data warehouse approach is also known as ETL consisting of extraction, transformation, and loading (Hu et al., 2014).  The extraction step involves the connection to the source systems and selecting and collecting the required data to be processed for analytical purposes.  The transformation step involves the application of a series of rules to the extracted data to convert it into a standard format.  The load step involves importing extracted and transformed data into a target storage infrastructure (Hu et al., 2014).  The federation approach creates a virtual database to query and aggregate data from various sources (Hu et al., 2014).  The virtual database contains information or metadata about the actual data, and its location and does not contain data itself (Hu et al., 2014).  These two data pre-processing are called store-and-pull techniques which is not appropriate for Big Data processing, with high computation and high streaming, and dynamic nature (Hu et al., 2014).  

The data cleansing process is a vital process to keep the data consistent and updated to get widely used in many fields such as banking, insurance, and retailing (Hu et al., 2014).  The cleansing process is required to determine the incomplete, inaccurate, or unreasonable data and then remove these data to improve the quality of the data (Hu et al., 2014). The data cleansing process includes five steps (Hu et al., 2014).  The first step is to define and determine the error types.  The second step is to search and identify error instances.  The third step is to correct the errors, and then document error instances and error types. The last step is to modify data entry procedures to reduce future errors.  Various types of checks must be done at the cleansing process, including the format checks, completeness checks, reasonableness checks, and limit checks (Hu et al., 2014).  The process of data cleansing is required to improve the accuracy of the analysis (Hu et al., 2014).  The data cleansing process depends on the complex relationship model, and it has extra computation and delay overhead (Hu et al., 2014).  Organizations must seek a balance between the complexity of the data-cleansing model and the resulting improvement in the accuracy analysis (Hu et al., 2014). 

The data redundancy is the third data pre-processing step where data is repeated increasing the overhead of the data transmission and causes limitawtions for storage systems, including wasted space, inconsistency of the data, corruption of the dta, and reduced reliability (Hu et al., 2014).  Various redundancy reduction methods include redundancy detection and data compression (Hu et al., 2014).  The data compression method poses an extra computation burden in the data compression and decompression processes (Hu et al., 2014).

Data Modeling and Design Consideration

Schema-on-Write system is used when the application or structure is well understood and frequently accessed through queries and reports on high-value data (Grover et al., 2015).        The term Schema-on-Read is used in the context of Hadoop data management system (Ankam, 2016; Grover et al., 2015). This term refers to the raw data, that is not processed and can be loaded to Hadoop using the required structure at processing time based on the requirement of the processing application (Ankam, 2016; Grover et al., 2015).  The Schema-on-Read is used when the application or structure of data is not well understood (Ankam, 2016; Grover et al., 2015).  The agility of the process is implemented through the schema-on-read providing valuable insights on data not previously accessible (Grover et al., 2015).

            Five factors must be considered before storing data into Hadoop for processing (Grover et al., 2015).  The data storage format must be considered as there are some file formats and compression formats supported on Hadoop.  Each type of format has strengths that make it better suited to specific applications.   Although Hadoop Distributed File System (HDFS) is a building block of Hadoop ecosystem, which is used for storing data, several commonly used systems implemented on top of HDFS such as HBase for traditional data access functionality, and Hive for additional data management functionality (Grover et al., 2015).  These systems of HBase for data access functionality and Hive for data management functionality must be taken into consideration before storing data into Hadoop (Grover et al., 2015). The second factor involves the multitenancy which is a common approach for clusters to host multiple users, groups and application types. The multi-tenant clusters involve essential considerations for data storage.  The schema design factor should also be considered before storing data into Hadoop even if Hadoop is a schema-less (Grover et al., 2015).  The schema design consideration involves directory structures for data loaded into HDFS and the output of the data processing and analysis, including the schema of objects stored in systems such as HBase and Hive.  The last factor for consideration before storing data into Hadoop is represented in the metadata management.  Metadata is related to the stored data and is often regarded as necessary as the data.  The understanding of the metadata management plays a significant role as it can affect the accessibility of the data.  The security is another factor which should be considered before storing data into Hadoop system.  The security of the data decision involves authentication, fine-grained access control, and encryption. These security measures should be considered for data at rest when it gets stored as well as in motion during the processing (Grover et al., 2015).  Figure 4 summarizes these considerations before storing data into the Hadoop system. 

Figure 4.  Considerations Before Storing Data into Hadoop.

Data Storage Format Considerations

            When architecting a solution on Hadoop, the method of storing the data into Hadoop is one of the essential decisions. Primary considerations for data storage in Hadoop involve file format, compression, data storage system (Grover et al., 2015).  The standard file formats involve three types:  text data, structured text data, and binary data.  Figure 5 summarizes these three standard file formats.

Figure 5.  Standard File Formats.

The text data is widespread use of Hadoop including log file such as weblogs, and server logs (Grover et al., 2015).  These text data format can come in many forms such as CSV files, or unstructured data such as emails.  Compression of the file is recommended, and the selection of the compression is influenced by how the data will be used (Grover et al., 2015).  For instance, if the data is for archival, the most compact compression method can be used, while if the data are used in processing jobs such as MapReduce, the splittable format should be used (Grover et al., 2015).  The splittable format enables Hadoop to split files into chunks for processing, which is essential to efficient parallel processing (Grover et al., 2015).

In most cases, the use of container formats such as SequenceFiles or Avro provides benefits making it the preferred format for most file system including text (Grover et al., 2015).  It is worth noting that these container formats provide functionality to support splittable compression among other benefits (Grover et al., 2015).   The binary data involves images which can be stored in Hadoop as well.  The container format such as SequenceFile is preferred when storing binary data in Hadoop.  If the binary data splittable unit is more than 64MB, the data should be put into its file, without using the container format (Grover et al., 2015).

XML and JSON Format Challenges with Hadoop

The structured text data include formats such as XML and JSON, which can present unique challenges using Hadoop because splitting XML and JSON files for processing is not straightforward, and Hadoop does not provide a built-in InputFormat for either (Grover et al., 2015).  JSON presents more challenges to Hadoop than XML because no token is available to mark the beginning or end of a record.  When using these file format, two primary consideration must be taken.  The container format such as Avro should be used because Avro provides a compact and efficient method to store and process the data when transforming the data into Avro (Grover et al., 2015).  A library for processing XML or JSON should be designed.  XMLLoader in PiggyBank library for Pig is an example when using XML data type.  The Elephant Bird project is an example of a JSON data type file (Grover et al., 2015). 

Hadoop File Types Considerations

            Several Hadoop-based file formats created to work well with MapReduce (Grover et al., 2015).  The Hadoop-specific file formats include file-based data structures such as sequence files, serialization formats like Avro, and columnar formats such as RCFile and Parquet (Grover et al., 2015).  These files types share two essential characteristics that are important for Hadoop application: splittable compression and agnostic compression.  The ability of splittable files play a significant role during the data processing, and should not be underestimated when storing data in Hadoop because it allows large files to be split for input to MapReduce and other types of jobs, which is a fundamental part of parallel processing and a key to leveraging data locality feature of Hadoop (Grover et al., 2015).  The agnostic compression is the ability to compress using any compression codec without readers having to know the codec because the codec is stored in the header metadata of the file format (Grover et al., 2015).  Figure 6 summarizes these Hadoop-specific file formats with the typical characteristics of splittable compression and agnostic compression.

Figure 6. Three Hadoop File Types with the Two Common Characteristics.  

1.      SequenceFiles Format Consideration

SequenceFiles format is the most widely used Hadoop file-based formats.  SequenceFile format store data as binary key-value pairs (Grover et al., 2015).  It involves three formats for records stored within SequenceFiles:  uncompressed, record-compressed, and block-compressed.  Every SequenceFile uses a standard header format containing necessary metadata about the file such as the compression codec used, key and value class names, user-defined metadata, and a randomly generated syn marker.  The SequenceFiles arewell supported in Hadoop. However, it has limited support outside the Hadoop ecosystem as it is only supported in Java language.  The frequent use case for SequenceFiles is a container for smaller files.  However, storing a large number of small files in Hadoop can cause memory issue and excessive overhead in processing.  Packing smaller files into a SequenceFile can make the storage and processing of these files more efficient because Hadoop is optimized for large files (Grover et al., 2015).   Other file-based formats include the MapFiles, SetFiles, Array-Files, and BloomMapFiles.  These formats offer a high level of integration for all forms of MapReduce jobs, including those run via Pig and Hive because they were designed to work with MapReduce (Grover et al., 2015).  Figure 7 summarizes the three formats for records stored within SequenceFiles.

Figure 7.  Three Formats for Records Stored within SequenceFile.

2.      Serialization Formats Consideration

Serialization is the process of moving data structures into bytes for storage or for transferring data over the network (Grover et al., 2015).   The de-serialization is the opposite process of converting a byte stream back into a data structure (Grover et al., 2015).  The serialization process is the fundamental building block for distributed processing systems such as Hadoop because it allows data to be converted into a format that can be efficiently stored and transferred across a network connection (Grover et al., 2015).  Figure 8 summarizes the serialization formats when architecting for Hadoop.

Figure 8.  Serialization Process vs. Deserialization Process.

The serialization involves two aspects of data processing in a distributed system of interprocess communication using data storage, and remote procedure calls or RPC (Grover et al., 2015).  Hadoop utilizes Writables as the main serialization format, which is compact and fast but uses Java only.  Other serialization frameworks have been increasingly used within Hadoop ecosystems, including Thrift, Protocol Buffers and Avro (Grover et al., 2015).  Avro is a language-neutral data serialization system (Grover et al., 2015).  It was designed to address the limitation of the Writables of Hadoop which is lack of language portability.  Similar to Thrift and Protocol Buffers, Avro is described through a language-independent schema (Grover et al., 2015).   Avro, unlike Thrift and Protocol Buffers, the code generation is optional.  Table 1 provides a comparison between these serialization formats.

Table 1:  Comparison between Serialization Formats.

3.      Columnar Format Consideration

Row-oriented systems have been used to fetch data stored in the database (Grover et al., 2015).  This type of data retrieval has been used as the analysis heavily relied on fetching all fields for records that belonged to a specific time range.  This process is efficient if all columns of the record are available at the time or writing because the record can be written with a single disk seek.  The column storage has recently been used to fetch data.  The use of columnar storage has four main benefits over the row-oriented system (Grover et al., 2015).  The skips I/O and decompression on columns that are not part of the query is one of the benefits of the columnar storage.  Columnar data storage works better for queries that access a small subset of columns than the row-oriented data storage, which can be used when many columns are retrieved.  The compression on columns provides efficiency because data is more similar within the same column than it is in a block of rows.  The columnar data storage is more appropriate for data warehousing-based applications where aggregations are implemented using specific columns than an extensive collection of records (Grover et al., 2015).  Hadoop applications have been using the columnar file formats including the RCFile format, Optimized Row Columnar (ORC), and Parquet.  The RCFile format has been used as a Hive Format.  It was developed to provide fast data loading, fast query processing, and highly efficient storage space utilization.  It breaks files into row splits, and within each split uses columnar-oriented storage.  Despite its advantages of the query and compression performance compared to SequenceFiles, it has limitations, that prevent the optimal performance for query times and compression.  The newer version of the columnar formats ORC and Parquet are designed to address many of the limitations of the RCFile (Grover et al., 2015). 

Compression Consideration

Compression is another data storage consideration because it plays a crucial role in reducing the storage requirements, and in improving the data processing performance (Grover et al., 2015).  Some compression formats supported on Hadoop are not splittable (Grover et al., 2015).  MapReduce framework splits data for input to multiple tasks; the nonsplittable compression format is an obstacle to efficient processing.  Thus, the splittability is a critical consideration in selecting the compression format and file format for Hadoop.  Various compression types for Hadoop include Snappy, LZO, Gzip, bzip2.  Google developed Snappy for speed. However, it does not offer the best compression size. It is designed to be used with a container format like SequenceFile or Avro because it is not inherently splittable.  It is being distributed with Hadoop. Similar to Snappy, LZO is optimized for speed as opposed to size.  However, LZO, unlike Snappy support splittability of the compressed files, but it requires indexing.  LZO, unlike Snappy, is not distributed with Hadoop and requires a license and separate installation.  Gzip, like Snappy, provides good compression performance, but is not splittable, and it should be used with a container format. The speed read performance of the Gzip is like the Snappy.  Gzip is slower than Snappy for write processing.  Gzip is not splittable and should be used with a container format.  The use of smaller blocks with Gzip can result in better performance.   The bzip2 is another compression type for Hadoop.  It provides good compression performance, but it can be slower than another compression codec such as Snappy.  It is not an ideal codec for Hadoop storage. Bzip2, unlike Snappy and Gzip, is inherently splittable.  It inserts synchronization markers between blocks.  It can be used for active archival purposes (Grover et al., 2015).

The compression format can become splittable when used with container file formats such as Avro, SequenceFile which compress blocks of records or each record individually (Grover et al., 2015).  If the compression is implemented on the entire file without using the container file format, the compression format that inherently supports splittable must be used such as bzip2.  The compression use with Hadoop has three recommendation (Grover et al., 2015).  The first recommendation is to enable compression of MapReduce intermediate output, which improves performance by decreasing the among of intermediate data that needs to be read and written from and to disk.  The second recommendation s to pay attention to the order of the data.  When the data is close together, it provides better compression levels. The data in Hadoop file format is compressed in chunks, and the organization of those chunks determines the final compression.   The last recommendation is to consider the use of a compact file format with support for splittable compression such as Avro.  Avro and SequenceFiles support splittability with non-splittable compression formats.  A single HDFS block can contain multiple Avro or SequenceFile blocks. Each block of the Avro or SequenceFile can be compressed and decompressed individually and independently of any other blocks of Avro or SequenceFile. This technique makes the data splittable because each block can be compressed and decompressed individually.  Figure 9 shows the Avro and SequenceFile splittability support (Grover et al., 2015).  

Figure 9.  Compression Example Using Avro (Grover et al., 2015).

Design Consideration for HDFS Schema

HDFS and HBase are the commonly used storage managers in the Hadoop ecosystem.  Organizations can store the data in HDFS or HBase which internally store it on HDFS (Grover et al., 2015).  When storing data in HDFS, some design techniques must be taken into consideration.  The schema-on-read model of Hadoop does not impose any requirement when loading data into Hadoop, as data can be ingested into HDFS by one of many methods without the requirements to associate a schema or preprocess the data.  Although Hadoop has been used to load many types of data such as the unstructured data, semi-structured data, some order is still required, because Hadoop serves as a central location for the entire organization and the data stored in HDFS is intended to be shared across various departments and teams in the organization (Grover et al., 2015).  The data repository should be carefully structured and organized to provide various benefits to the organization  (Grover et al., 2015).   When there is a standard directory structure, it becomes easier to share data among teams working with the same data set.  The data gets staged in a separate location before processing it.  The standard stage technique can help not processing data that has not been appropriately staged or entirely yet.  The standard organization of data allows for some code reuse that may process the data (Grover et al., 2015).  The placement of data assumptions can help simplify the loading process of the data into Hadoop.   The HDFS data model design for projects such as data warehouse implementation is likely to use structure facts and dimension tables similar to the traditional schema  (Grover et al., 2015).  The HDFS data model design for projects of unstructured and semi-structured data is likely to focus on directory placement and metadata management (Grover et al., 2015). 

Grover et al. (2015) suggested three key considerations when designing the schema, regardless of the data model design project.  The first consideration is to develop standard practices that can be followed by all teams.  The second point is to ensure the design works well with the chosen tools.  For instance, if the version of Hive can support only table partitions on directories that are named a certain way, it will affect the schema design and the names of the table subdirectories.  The last consideration when designing a schema is to keep usage patterns in mind, because different data processing and querying patterns work better with different schema designs (Grover et al., 2015). 

HDFS Files Location Consideration

            The first step when designing an HDFS schema involves the determination of the location of the file.  Standard file location plays a significant role in finding and sharing data among various departments and teams. It also helps in the assignment of permission to access files to various groups and users.  The recommended file locations are summarized in Table 2.

Table 2.  Standard Files Locations.

HDFS Schema Design Consideration

The HDFS schema design involves advanced techniques to organize data into files (Grover et al., 2015).   A few strategies are recommended to organize the data set. These strategies for data organization involve partitioning, bucketing, and denormalizing process.  The partitioning process of the data set is a common technique used to reduce the amount of I/O required to process the data set.  HDFS does not store indexes on the data unlike the traditional data warehouse. Such a lack of indexes in HDFS plays a key role in speeding up data ingest, with a full table scan cost where every query will have to read the entire dataset even when processing a small subset of data. Breaking up the data set into smaller subsets, or partitions can help with the full table scan, allowing queries to read only the specific partitions reducing the amount of I/O and improving the query time processing significantly (Grover et al., 2015). When data is placed in the filesystem, the directory format for partition should be as shown below.  The order data sets are partitioned by date because there are a large number of orders done daily and the partitions will contain large enough files which are optimized by HDFS.  Various tools such as HCatalog, Hive, Impala, and Pig understand this directory structure leveraging the partitioning to reduce the amount of I/O requiring during the data processing (Grover et al., 2015).

• <data set name>/<partition_column_name=partition_column_value>/(Armstrong)
• e.g. medication_orders/date=20181107/[order1.csv, order2.csv]

Bucketing is another technique for breaking a large data set into manageable sub-sets (Grover et al., 2015).  The bucketing technique is similar to the hash partitions which is used in the relational database.   Various tools such as HCatalog, Hive, Impala, and Pig understand this directory structure leveraging the partitioning to reduce the amount of I/O requiring during the data processing. The partition example above was implemented using the date which resulted in large data files which can be optimized by HDFS (Grover et al., 2015).  However, if the data sets are partitioned by a the category of the physician, the result will be too many small files, which leads to small file problems, which can lead to excessive memory use for the NameNode, since metadata for each file stored in HDFS is stored in memory (Grover et al., 2015).  Many small files can also lead to many processing tasks, causing excessive overhead in processing.  The solution for too many small files is to use the bucketing process for the physician in this example, which uses the hashing function to map physician into a specified number of buckets (Grover et al., 2015).

The bucketing technique controls the size of the data subsets and optimizes the query speed (Grover et al., 2015).  The recommended average bucket size is a few multiples of the HDFS block size. The distribution of data when hashed on the bucketing column is essential because it results in consistent bucketing (Grover et al., 2015).  The use of the number of buckets as a power of two is every day.   Bucketing allows joining two data sets.  The join, in this case, is used to represent the general idea of combining two data sets to retrieve a result. The joins can be implemented through the SQL-on-Hadoop systems and also in MapReduce, or Spark, or other programming interfaces to Hadoop.  When using join in the bucketing technique, it joins corresponding buckets individually without having to join the entire datasets, which help in minimizing the time complexity for the reduce-side join of the two datasets process, which is computationally expensive (Grover et al., 2015).   The join is implemented in the map stage of a MapReduce job by loading the smaller of the buckets in memory because the buckets are small enough to easily fit into memory, which is called map-side join process.  The map-side join process improves the join performance as compared to a reduce-side join process.  A hive for data analysis recognizes the tables are bucketed and optimize the process accordingly.

Further optimization can be implemented if the data in the bucket is sorted, the merge join can be used, and the entire bucket does not get stored in memory when joining, resulting in the faster process and much less memory than a simple bucket join.  Hive supports this optimization as well.  The use of both sorting and bucketing on large tables that are frequently joined together using the join key for bucketing is recommended (Grover et al., 2015).

The schema design depends on how the data will be queried (Grover et al., 2015).  Thus, the columns to be used for joining and filtering must be identified before the portioning and bucketing of the data is implemented.   In some cases, when the identification of one partitioning key is challenging, storing the same data set multiple times can be implemented, each with the different physical organization, which is regarded to be an anti-pattern in a relational database.  However, this solution can be implemented with Hadoop, because with Hadoop is write-once, and few updates are expected.  Thus, the overhead of keeping duplicated data set in sync is reduced.  The cost of storage in Hadoop clusters is reduced as well  (Grover et al., 2015). The duplicated data set in sync provides better query speed processing in such cases (Grover et al., 2015). 

Regarding the denormalizing process, it is another technique of trading the disk space for query performance, where joining the entire data set need is minimized (Grover et al., 2015).   In the relational database model, the data is stored in the third standard form (NF3), where redundancy is minimized, and data integrity is enforced by splitting data into smaller tables, each holding a particular entity.  In this relational model, most queries require joining a large number of tables together to produce a final result as desired (Grover et al., 2015).  However, in Hadoop, joins are often the slowest operations and consume the most resources from the cluster.  Specifically, the reduce-side join requires sending the entire table over the network, which is computationally costly.  While sorting and bucketing help minimizing this computational cost, another solution is to create data sets that are pre-joined or pre-aggregated (Grover et al., 2015).  Thus, the data can be joined once and store it in this form instead of running the join operations every time there is a query for that data.  Hadoop schema consolidates many of the small dimension tables into a few larger dimensions by joining them during the ETL process  (Grover et al., 2015).  Other techniques to speed up the process include the aggregation or data type conversion.  The duplication of the data is of less concern; thus, when the processing is frequent for a large number of queries, it is recommended to doing it one and reuse as the case with a materialized view in the relational database.  In Hadoop, the new dataset is created that contains the same data in its aggregated form (Grover et al., 2015).

To summarize, the partitioning process is used to reduce the I/O overhead of processing by selectively reading and writing data in particular partitions.  The bucketing can be used to speed up queries that involve joins or sampling, by reducing the I/O as well.  The denormalization can be implemented to speed up Hadoop jobs.   In this section, a review of advanced techniques to organize data into files is discussed.  The discussion includes the use of a small number of large files versus a large number of small files.  Hadoop prefers working with a small number of large files than a large number of small files.  The discussion also addresses the reduce-side join versus map-side join techniques.   The reduce-side join is computationally costly. Hence, the map-side join technique is preferred and recommend. 

HBase Schema Design Consideration

HBase is not a relational database (Grover et al., 2015; Yang, Liu, Hsu, Lu, & Chu, 2013).  HBase is similar to a large hash table, which allows the association of values with keys and performs a fast lookup of the value based on a given key  (Grover et al., 2015). The operations of hash tables involve put, get, scan, increment and delete.  HBase provides scalability and flexibility and is useful in many applications, including fraud detection, which is a widespread application for HBase (Grover et al., 2015).

The framework of HBase involves Master Server, Region Servers, Write-Ahead Log (WAL), Memstore, HFile, API and Hadoop HDFS (Bhojwani & Shah, 2016).  Each component of the HBase framework plays a significant role in data storage and processing.  Figure 10 illustrated the HBase framework.

Figure 10.  HBase Architecture (Bhojwani & Shah, 2016).

            The following consideration must be taken when designing the schema for HBase (Grover et al., 2015).

• Row Key Consideration.
• Timestamp Consideration.
• Hops Consideration.
• Tables and Regions Consideration.
• Columns Use Consideration.
• Column Families Use Consideration.
• Time-To-Live Consideration.

The row key is one of the most critical factors for well-architected HBase schema design (Grover et al., 2015).  The row key consideration involves record retrieval, distribution, block cache, the ability to scan, size, readability, and uniqueness.  The row key is critical for retrieving records from HBase. In the relational database, the composite key can be used to combine multiple primary keys.  In HBase, multiple pieces of information can be combined in a single key.  For instance, a key of customer_id, order_id, and timestamp will be a row key for a row describing an order. In a relational database, they are three different columns in the relational database, but in HBase, they will be combined into a single unique identifier.  Another consideration for selecting the row key is the get operation because a get operation of a single record is the fasted operation in HBase.  A single get operation can retrieve the most common uses of the data improves the performance, which requires to put much information in a single record which is called denormalized design.    For instance, while in the relational database, customer information will be placed in various tables, in HBase all customer information will be stored in a single record where get operation will be used. The distribution is another consideration for HBase schema design.  The row key determines the regions of HBase cluster for a given table, which will be scattered throughout various regions (Grover et al., 2015; Yang et al., 2013).   The row keys are sorted, and each region stores a range of these sorted row keys  (Grover et al., 2015).  Each region is pinned to a region server namely a node in the cluster  (Grover et al., 2015).  The combination of device ID and timestamp or reverse timestamp is commonly used to “salt” the key in machine data  (Grover et al., 2015).  The block cache is a least recently used (LRU) cache which caches data blocks in memory  (Grover et al., 2015).  HBase reads records in chunks of 64KB from the disk by default. Each of these chunks is called HBase block  (Grover et al., 2015).  When the HBase block is read from disk, it will be put into the block cache  (Grover et al., 2015).   The choice of the row key can affect the scan operation as well.  HBase scan rates are about eight times slower than HSFS scan rates.  Thus, reducing I/O requirements has a significant performance advantage.  The size of the row key determines the performance of the workload.  The short row key is better than, the long row key because it has lower storage overhead and faster read/ writes performance.  The readability of the row key is critical. Thus, it is essential to start with human-readable row key.  The uniqueness of the row key is also critical since a row key is equivalent to a key in hash table analogy.  If the row key is based on the non-unique attribute, the application should handle such cases and only put data in HBase with a unique row key (Grover et al., 2015).

The timestamp is the second essential consideration for good HBase schema design (Grover et al., 2015).  The timestamp provides advantages of determining which records are newer in case of put operation to modify the record.  It also determines the order where records are returned when multiple versions of a single record are requested. The timestamp is also utilized to remove out-of-date records because time-to-live (TTL) operation compared with the timestamp shows the record value has either been overwritten by another put or deleted (Grover et al., 2015).

The hop term refers to the number of synchronized “get” requests to retrieve specific data from HBase (Grover et al., 2015). The less hop, the better because of the overhead.  Although multi-hop requests with HBase can be made, it is best to avoid them through better schema design, for example by leveraging de-normalization, because every hop is a round-trip to HBase which has a significant performance overhead (Grover et al., 2015).

The number of tables and regions per table in HBase can have a negative impact on the performance and distribution of the data (Grover et al., 2015).  If the number of tables and regions are not implemented correctly, it can result in an imbalance in the distribution of the load.  Important considerations include one region server per node, many regions in a region server, a give region is pinned to a particular region server, and tables are split into regions and scattered across region servers.  A table must have at least one region.  All regions in a region server receive “put” requests and share the region server’s “memstore,” which is a cache structure present on every HBase region server. The “memstore” caches the write is sent to that region server and sorts them in before it flushes them when certain memory thresholds are reached. Thus, the more regions exist in a region server; the less memstore space is available per region.  The default configuration sets the ideal flush size to 100MB. Thus, the “memstore” size can be divided by 100MB and result should be the maximum number of regions which can be put on that region server.   The vast region takes a long time to compact.  The upper limit on the size of a region is around 20GB. However, there are successful HBase clusters with upward of 120GB regions.  The regions can be assigned to HBase table using one of two techniques. The first technique is to create the table with a single default region, which auto splits as data increases.  The second technique is to create the table with a given number of regions and set the region size to a high enough value, e.g., 100GB per region to avoid auto splitting (Grover et al., 2015).  Figure 11 shows a topology of region servers, regions and tables. 

Figure 11.  The Topology of Region Servers, Regions, and Tables (Grover et al., 2015).

The columns used in HBase is different from the traditional relational database (Grover et al., 2015; Yang et al., 2013).  In HBase, unlike the traditional database, a record can have a million columns, and the next record can have a million completely different columns, which is not recommended but possible (Grover et al., 2015).   HBase stores data in a format called HFile, where each column value gets its row in HFile (Grover et al., 2015; Yang et al., 2013). The row has files like row key, timestamp, column names, and values. The file format provides various functionality, like versioning and sparse column storage (Grover et al., 2015). 

HBase, include the concept of column families (Grover et al., 2015; Yang et al., 2013).  A column family is a container for columns.  In HBase, a table can have one or more column families.  Each column family has its set of HFiles and gets compacted independently of other column families in the same table.  In many cases, no more than one column family is needed per table.  The use of more than one column family per table can be done when the operation is done, or the rate of change on a subset of the columns of a table is different from the other columns (Grover et al., 2015; Yang et al., 2013).  The last consideration for HBase schema design is the use of TTL, which is a built-in feature of HBase which ages out data based on its timestamp (Grover et al., 2015).  If TTL is not used and an aging requirement is needed, then a much more I/O intensive operation would need to be done.   The objects in HBase begin with table object, followed by regions for the table, store per column family for each region for the table, memstore, store files, and block (Yang et al., 2013).  Figure 12 shows the hierarchy of objects in HBase.

Figure 12.  The Hierarchy of Objects in HBase (Yang et al., 2013).

To summarize this section, HBase schema design requires seven key consideration starting with the row key, which should be selected carefully for record retrieval, distribution, block cache, ability to scan, size, readability, and uniqueness.  The timestamp and hops are other schema design consideration for HBase.  Tables and regions must be considered for put performance, and compacting time.  The use of columns and column families should also be considered when designing the schema for HBase. The TTL to remove data that aged is another consideration for HBase schema design. 

Metadata Consideration

The above discussion has been about the data and the techniques to store it in Hadoop.  Metadata is as essential as the data itself.  Metadata is data about the data (Grover et al., 2015)).  Hadoop ecosystem has various forms of metadata.   Metadata about logical dataset usually stored in a separate metadata repository include the information like the location of a data set such as directory in HDFS or HBase table name, the schema associated with the dataset, the partitioning and sorting properties of the data set, the format of the data set e.g. CSV, SequenceFile, etc. (Grover et al., 2015). The metadata about files on HDFS includes the permission and ownership of such files and the location of various blocks on data nodes, usually stored and managed by Hadoop NameNode (Grover et al., 2015).  Metadata about tables in HBase include information like table names, associated namespace, associated attributes, e.g. MAX_FILESIZE, READONLY, etc., and the names of column families, usually stored and managed by HBase (Grover et al., 2015).  Metadata about data ingest and transformation include information like which user-generated a given dataset, where the dataset came from, how long it took to generate it, and how many records there are, or the size of the data load (Grover et al., 2015).  Metadata about dataset statistics include information like the number of rows in a dataset, number of unique values in each column, a histogram of the distribution of the data, and maximum and minimum values (Grover et al., 2015).  Figure 13 summarizes this various metadata.

Figure 13.  Various Metadata in Hadoop.

Apache Hive was the first project in the Hadoop ecosystem to store, manage and leverage metadata (Antony et al., 2016; Grover et al., 2015).  Hives stores this metadata in a relational database called the Hive “metastore” (Antony et al., 2016; Grover et al., 2015).  Hive also provides a “metastore” service which interfaces with the Hive metastore database (Antony et al., 2016; Grover et al., 2015).  The query process in Hive goes to the metastore to get the metadata for the desired query, and metastore sends the metadata to Hive generating execution plan, followed by executing the job using the Hadoop cluster, which implements the job and Hive send the fetched result to the user (Antony et al., 2016; Grover et al., 2015).  Figure 14 shows the query process and the role of the metastore in Hive framework.

Figure 14.  Query Process and the Role of Metastore in Hive (Antony et al., 2016).

More projects have utilized the concept of metadata that was introduced by Hive and created a separate project called HCatalog to enable the usage of Hive metastore outside of Hive (Grover et al., 2015).  HCatalog is a part of Hive and allows other tools like Pig and MapReduce to integrate with Hive metastore.  It also opens the access to Hive metastore to other tools such as REST API via WebHCat server.  MapReduce, Pig, and standalone applications can talk directly to the metastore of Hive through its APIs, but HCatalog allows easy access through its WebHCat REST APIs, and it allows the cluster administrators to lock down access to the Hive metastore to address security concerns. Other ways to store metadata include the embedding of metadata in file paths and names.  Another technique to store metadata involves storing it in HDFS in a hidden file, e.g., .metadata.  Figure 15 shows the HCatalog as an accessibility veneer around the Hive metastore (Grover et al., 2015). 

Figure 15.  HCatalog acts an accessibility veneer around the Hive metastore (Grover et al., 2015).

Hive Metastore and HCatalog Limitations

There are some limitations for Hive metastore and HCatalog, including the problem with high availability (Grover et al., 2015).  The HA database cluster solutions to bring HA to the Hive metastore database.  For the metastore service of Hive, there is support concurrently to run multiple metastores on more than one node in the cluster.  However, concurrency issues related to data definition language operations (DDL) can occur, and Hive community is working on fixing these issues. 

The fixed schema for metadata is another limitation.  Hadoop provides much flexibility on the type of data that can be stored, mainly because of the Schema-on-Read concept. Hive metastore provides a fixed schema for the metadata itself. It provides a tabular abstraction for the data sets.   The data in metastore is moving the part in the infrastructure which requires to be running and secured as part of Hadoop infrastructure (Grover et al., 2015).

Conclusion

This project has discussed essential topics related to Hadoop technology.  It began with an overview of Hadoop providing a history of Hadoop and the difference between Hadoop 1.x and Hadoop 2.x.  The discussion involved the Big Data Analytics Process using Hadoop technology.  The process involves six significant steps starting with the problem identification, required data to be collected, and the data collection process. The pre-processing data and ETL process must be implemented before performing the analytics. The last step is the visualization of the data for decision making.  Before processing any data and before collecting any data for storage, some considerations must be taken for data preprocessing, modeling and schema design in Hadoop for better processing and better data retrieval, giving some tools cannot split the data while others can.  These considerations begin with data storage format, followed by Hadoop file types consideration and XML and JSON format challenges in Hadoop.  Compression must be considered when designing the schema for Hadoop. Since HDFS and HBase are commonly used in Hadoop for data storage, the discussion involved the consideration for the HDFS and HBase schema design considerations.  To summarize the design of the schema for Hadoop, HDFS, and HBase makes a difference in storing data in various nodes using the right tools for splitting the data.  Thus, organizations must pay attention to the process and the design requirements before storing data into Hadoop for better computational processing. 

References

Alguliyev, R., & Imamverdiyev, Y. (2014). Big data: big promises for information security. Paper presented at the Application of Information and Communication Technologies (AICT), 2014 IEEE 8th International Conference on.

Ankam, V. (2016). Big Data Analytics: Packt Publishing Ltd.

Antony, B., Boudnik, K., Adams, C., Lee, C., Shao, B., & Sasaki, K. (2016). Professional Hadoop: John Wiley & Sons.

Armstrong, D. (n.d.). R: Learning by Example: Lattice Graphics. Retrieved from https://quantoid.net/files/rbe/lattice.pdf.

Bhojwani, N., & Shah, A. P. V. (2016). A SURVEY ON HADOOP HBASE SYSTEM. Development, 3(1).

Dittrich, J., & Quiané-Ruiz, J.-A. (2012). Efficient big data processing in Hadoop MapReduce. Proceedings of the VLDB Endowment, 5(12), 2014-2015.

Grover, M., Malaska, T., Seidman, J., & Shapira, G. (2015). Hadoop Application Architectures: Designing Real-World Big Data Applications: ” O’Reilly Media, Inc.”.

Hu, H., Wen, Y., Chua, T.-S., & Li, X. (2014). Toward scalable systems for big data analytics: A technology tutorial. IEEE Access, 2, 652-687.

Karanth, S. (2014). Mastering Hadoop: Packt Publishing Ltd.

Lublinsky, B., Smith, K. T., & Yakubovich, A. (2013). Professional hadoop solutions: John Wiley & Sons.

sas.com. (2018). Hadoop – why it is and why it matters. Retrieved from https://www.sas.com/en_us/insights/big-data/hadoop.html.

Yang, C. T., Liu, J. C., Hsu, W. H., Lu, H. W., & Chu, W. C. C. (2013, 16-18 Dec. 2013). Implementation of Data Transform Method into NoSQL Database for Healthcare Data. Paper presented at the 2013 International Conference on Parallel and Distributed Computing, Applications and Technologies.

 

 

Dr. Aly, O.
Computer Science

The purpose of this discussion is to identify and describe a tool in the market for data analytics, how the tool is used and where it can be used.  The discussion begins with an overview of the Big Data Analytics tools, followed by the top five tools for 2018, among which RapidMiner is selected as the BDA tool for this discussion.  The discussion of the RapidMiner as one of the top five BDA tools include the features, technical specification, use, advantages, and limitation.  The application of RapidMiner in various industries such as medical and education is also addressed in this discussion. 

Overview of Big Data Analytics Tools

Organizations must be able to quickly and effectively analyze a large amount of data and extract value from such data for sound business decisions.  The benefits of Big Data Analytics are driving organizations and businesses to implement the Big Data Analytics techniques to be able to compete in the market.  A survey conducted by CIO Insight has shown that 65% of the executives and senior decisions makers have indicated that organizations will risk becoming uncompetitive or irrelevant if Big Data is not embraced (McCafferly, 2015).  The same survey also has shown that 56% have anticipated a higher investment for big data, and 15% have indicated that such increasing trend in the budget allocation will be significant (McCafferly, 2015). Such budget allocation can be used for skilled professionals, BD data storage, BDA tools, and so forth.

Regarding the BDA tools, various BDA tools exist in the market for different business purposes based on the business model of the organization.  Organizations must select the right tool that will serve their business model.  Various studies have discussed various tools for BDA implementation.  (Chen & Zhang, 2014) have examined various types of BD tools. Some tools are based on batch processing such as Apache Hadoop, Dryad, Apache Mahout, and Tableau, while other tools are based on stream processing such as Storm, S4, Splunk, Apache Kafka, and SAP Hana as summarized in Table 1 and Table 2.  Each tool provides certain features for BDA implementation and offers various advantages to those BDA-adapted organizations.

Table 1.  Big Data Tools Based on Batch Processing (Chen & Zhang, 2014).

Table 2.  Big Data Tools Based on Stream Processing (Chen & Zhang, 2014).

Other studies such as (Rangra & Bansal, 2014) have provided a comparative study of data mining tools such as Weka, Keel, R-Programming, Knime, RapidMiner, and Orange, their technical specification, general features, specialization, advantages, and limitations. (Choi, 2017) have discussed the BDA tools by categories.  These BDA tools are categorized by open source data tools, data visualization tools, sentiment tools, and data extraction tools. Figure 1 provides a summary of some of the examples of BDA tools including the databases sources to download big datasets for analysis.

Figure 1.  A Summary of Big Data Analytics Tools.

(Al-Khoder & Harmouch, 2014) have evaluated four of the most popular open source and free data mining tools including R, RapidMiner, Weka, and Knime.  R foundation has developed R-Programming, while Rapid-I company have developed RapidMinder.  Weka is developed by University of Waikato, and Knime is developed by Knime.com AG. Figure 2 provides a summary of these four BDA most popular open source and free data mining tools, with the logo, description, launch date, current version at the time of writing the study, and development team.

Figure 2.  Open Source and Free Data Mining Tools Analyzed by (Al-Khoder & Harmouch, 2014).

The top five of BDA tools for 2018 include Tableau Public, Rapid Miner, Hadoop, R-Programming, IBM Big Data (Seli, 2017). The present discussion focuses on one of these two five BDA tools for 2018.  Figure 3 summarizes these top five BDA tools for 2018.

Figure 3.  Top Five BDA Tools for 2018.

RapidMiner Big Data Analytic Tool

RapidMiner Big Data Analytic tool is selected for the present discussion since it was among the top five BDA tools for 2018.  RapidMiner is an open source platform for BDA, based on Java programming language. RapidMiner provides machine learning procedures and data mining.  It also provides data visualization, processing, statistical modeling, deployment, evaluation and predictive analytics (Hofmann & Klinkenberg, 2013; Rangra & Bansal, 2014; Seli, 2017).  RapidMiner is known for its commercial and business applications, as it provides an integrated environment and platform for machine learning, data mining, predictive analysis, and business analytics  (Hofmann & Klinkenberg, 2013; Seli, 2017).  It is also used for research, education, training, rapid prototyping, and application development (Rangra & Bansal, 2014).  It is specialized in predictive analysis and statistical computing. It supports all steps of the data mining process (Hofmann & Klinkenberg, 2013; Rangra & Bansal, 2014). RapidMiner uses the client/server model, where the server can be software, or a service or on cloud infrastructures (Rangra & Bansal, 2014).

RapidMiner was released on 2006.  The latest version of RapidMiner server is 7.2 with a free version of server and Radoop and can be downloaded from RapidMiner site (rapidminer, 2018).  It can be installed on any operating system (Rangra & Bansal, 2014).  The advantages of the RapidMiner include an integrated environment for all steps that are required for data mining process, easy to use graphical user interface (GUI) for the design of data mining process, the visualization of the result and data, the validation and optimization of these processes.  RapidMiner can be integrated into more complex systems (Hofmann & Klinkenberg, 2013).  RapidMiner also stores the data mining processes in a machine-readable XML format, which can be executed with a click of a button, providing a visualized graphics of the data mining processes (Hofmann & Klinkenberg, 2013). It contains over a hundred learning schemes for regression classification and clustering analysis (Rangra & Bansal, 2014).  RapidMiner has a few limitations including the size constraints of the number of rows and more hardware resources than other tools such as SAS for the same task and data (Seli, 2017).  RapidMiner also requires prominent knowledge of the database handling (Rangra & Bansal, 2014).  

RapidMiner Use and Application

Data Mining requires six essential steps to extract value from a large dataset (Chisholm, 2013). The process of Data mining framework begins with business understanding, followed by the data understanding and data preparation.  The modeling, evaluation and deployment phases develop the models for predictions, testing, and deploying them in real-time. Figure 4 illustrates these six steps of the data mining.

Figure 4.  Data Mining Six Phases Process Framework (Chisholm, 2013).

Before working with RapidMiner, the user must know the common terms used by RapidMiner.  Some of these standard terms are a process, operator, macro, repository, attribute, role, label, and ID (Chisholm, 2013).  The data mining process in RapidMiner begins with loading the data into RapidMiner.  Loading the data into RapidMiner using import technique for either data in files, or databases.  The process of splitting the large file into pieces can be implemented in RapidMiner.  In some cases, the dataset can be split into chunks using RapidMiner process which reads each line in the file such as CSV file to be split into chunks. If the dataset is based on a database, a Java Data Connectivity (JDBC) driver must be used. RapidMiner support MySQL, PostgreSQL, SQL Server, Oracle and Access (Chisholm, 2013).  After loading the data into RapidMinder and generating data for testing, a predictive model can be created based on the loaded dataset, followed by the process execution and reviewing the result visually. RapidMiner provides various techniques to visualize the data.  It uses scatter plots, scatter 3D color, parallel and deviations, quartile color, plotting series, and survey plotter. Figure 5 illustrates scatter 3D color visualization of the data in RapidMiner (Chisholm, 2013).

Figure 5.  Scatter 3D Color Visualization of the Data in RapidMiner (Chisholm, 2013).

RapidMiner supports statistical analysis such as K-Nearest Neighbor Classifications, Naïve Bayes Classification, which can be used for credit approval and in education (Hofmann & Klinkenberg, 2013). RapidMiner application is also witnessed in other industries such as marketing, cross-selling and recommender system (Hofmann & Klinkenberg, 2013).  Other useful use cases of the RapidMiner application include the clustering in medical and education domains (Hofmann & Klinkenberg, 2013).  RapidMinder can also be used for text mining scenarios such as spam detection, language detection, and customer feedback analysis.  Other applications of RapidMiner include anomaly detection and instance selection.

Conclusion

This discussion has identified the different tools for Big Data Analytics (BDA). Over thirty analytic tools which can be used to overcome some of the BDA. Some are open source tools such as Knime, R-Programming, RapidMiner which can be downloaded for free, while others are described as visualization tools such as Tableau Public, Google Fusion to provide compelling visual images of the data in various scenarios.  Other tools are more semantic such as OpenText and Opinion Crawl.  Data extraction tools for BDA include Octoparse and Content Grabber. The users can download large datasets for BDA from various databases such as data.gov. 

The discussion has also addressed the top five BDA tools for 2018, such as Tableau Public, RapidMiner, Hadoop, R-Programming and IBM Big Data. RapidMiner was selected as BDA tools for this discussion.  The focus of the discussion on RapidMiner included the technical specification, use, advantages, and limitation.  The data mining process and steps when using RapidMiner have also been discussed.  The analytic process begins with the data upload to RapidMiner, during which the data can be split using the RapidMiner capabilities.  After the load and the cleaning of the data, the data model is developed and tested, followed by the visualization.  The visualization capabilities of RapidMiner include statistical analysis such as K-Nearest Neighbor and Naïve Bay Classification.  RapidMiner use cases have been addressed as well to include the medical and education domains, text mining scenarios such as spam detection.  Organizations must select the appropriate BDA tools based on the business model.

References

Al-Khoder, A., & Harmouch, H. (2014). Evaluating four of the most popular open source and free data mining tools.

Chen, C. P., & Zhang, C.-Y. (2014). Data-intensive applications, challenges, techniques and technologies: A survey on Big Data. Information Sciences, 275, 314-347.

Chisholm, A. (2013). Exploring data with RapidMiner: Packt Publishing Ltd.

Choi, N. (2017). Top 30 Big Data Tools for Data Analysis. Retrieved from https://bigdata-madesimple.com/top-30-big-data-tools-data-analysis/.

Hofmann, M., & Klinkenberg, R. (2013). RapidMiner: Data mining use cases and business analytics applications: CRC Press.

McCafferly, D. (2015). How To Overcome Big Data Barriers. Retrieved from https://www.cioinsight.com/it-strategy/big-data/slideshows/how-to-overcome-big-data-barriers.html.

Rangra, K., & Bansal, K. (2014). Comparative study of data mining tools. International journal of advanced research in computer science and software engineering, 4(6).

rapidminer. (2018). Introducing RapidMiner 7.2, Free Versions of Server & Radoop, and New Pricing. Retrieved from https://rapidminer.com/blog/introducing-new-rapidminer-pricing-free-versions-server-radoop/.

Seli, A. (2017). Top 5 Big Data Analytics Tools for 2018. Retrieved from http://heartofcodes.com/big-data-analytics-tools-for-2018/.

Dr. Aly, O.
Computer Science

The purpose of this discussion is to identify a real-life case study where Hadoop was used.  The discussion also addresses the view of the researcher whether Hadoop was used in the amplest manner.  The benefits of Hadoop to the identified industry of the use case are also discussed.  

Hadoop Real Life Case Study and Applications in the Healthcare Industry

Various research studies and reports have discussed Spark solution for real-time data processing in particular industries such as Healthcare, while others have discussed Hadoop solution for healthcare data analytics. For instance, (Shruika & Kudale, 2018) have discussed the use of Big Data in Healthcare with Spark, while (Beall, 2016) have indicated that United Healthcare is processing data using Hadoop framework for clinical advancements, financial analysis, and fraud and waste monitoring.  United Healthcare has utilized Hadoop to obtain a 360-degree view of each of its 85 million members (Beall, 2016). 

The emphasis of this discussion is on Hadoop in the Healthcare industry.  The data growth in the Healthcare industry is increasing exponentially (Dezyre, 2016).  McKinsey have anticipated the potential annual value for healthcare in the US is $300 billion, and 7% annual productivity growth using BDA (Manyika et al., 2011).  (Dezyre, 2016) have reported that the healthcare informatics poses challenges such as data knowledge representation, database design, data querying, and clinical decision support which contribute to the development of BDA.  

Big Data in healthcare include data such as patient-related data from electronic health records (EHRs), computerized physician order entry systems (CPOE), clinical decision support systems, medical devices and sensor, lab results and images such as Xrays, and so forth (Alexandru, Alexandru, Coardos, & Tudora, 2016; Wang, Kung, & Byrd, 2018).  Big Data framework for healthcare includes data layer, data aggregation layer, the analytical layer, the information exploration layer (Alexandru et al., 2016). Hadoop resides in the analytical layer of the Big Data framework (Alexandru et al., 2016).  

The data analysis involves Hadoop and MapReduce processing large dataset in batch form economically, analyzing both data types of structured and unstructured in a massively parallel processing environment (Alexandru et al., 2016).  (Alexandru et al., 2016) have indicated that stream computing can also be implemented using real-time or near real-time analysis to identify and respond to any health care fraud quickly.  The third type of analytics at the analytic layer also involves in-database analytics using data warehouse for data mining allowing high-speed parallel processing which can be used for prediction scenarios (Alexandru et al., 2016).  The in-database analytics can be used for preventive health care and pharmaceutical management.  Using Big Data framework including Hadoop ecosystem provides additional health care benefits such as scalability, security, confidentially and optimization features (Alexandru et al., 2016).

Hadoop technology was found to be the only technology that enables healthcare to store data in its native forms (Dezyre, 2016).   There are five successful use cases and applications of Hadoop in the healthcare industry (Dezyre, 2016).   The first application of Hadoop technology in healthcare is the cancer treatments and genomics.  Hadoop help develops better treatments for diseases such as cancel by accelerating the design and testing of effective treatments tailored to patients, expanding genetically based clinical cancer trials, and establishing a national “cancer knowledge network” to guide treatment decisions (Dezyre, 2016).  Hadoop can also be used to monitor the patient vitals.  The Children’s Healthcare of Atlanta is an example of using the Hadoop ecosystem to treat over 6,200 children in their ICU units.  Through Hadoop, the hospital was able to store and analyze the vital signs, and if there is any pattern change, an alert is generated and sent to the physicians (Dezyre, 2016).  The third application of Hadoop in Healthcare industry involves the hospital network.  The Cleveland Clinic spinoff company, known as “Explorys” is taking advantages of Hadoop by developing the most extensive database in the healthcare industry. As a result, Explorys was able to provide clinical support, reduce the cost of care measurement and manage the population of at-risk patients (Dezyre, 2016).  The fourth application of Hadoop in Healthcare industry involves healthcare intelligence, where healthcare insurance businesses are interested in finding the age of individuals in specific regions, who below a certain age are not a victim of certain diseases.  Through Hadoop technology, the healthcare insurance companies can compute the cost of insurance policy.  Pig, Hive, and MapReduce of Hadoop ecosystem are used in this scenario to process such a large dataset (Dezyre, 2016).  The last application of Hadoop in the healthcare industry involves fraud prevention and detection.  

Conclusion

In conclusion, the healthcare industry has taken advantages of Hadoop technology in various areas not only for better treatment and better medication but also for reducing the cost and increasing productivity and efficiency.  It has also used Hadoop for fraud protection.  These are not only the benefits which Hadoop offers the healthcare industry.  Hadoop also offers storage capabilities, scalability, and analytics capabilities of various types of datasets using parallel processing and distributed file system.  From the viewpoint of the researcher, utilizing Spark on top of Hadoop will empower the healthcare industry not only at the batching processing level but also at the real-time data processing. (Basu, 2014) have reported that the healthcare industry can take advantages of Spark and Shark with Apache Hadoop for real-time healthcare analytics.  Although Hadoop alone offers excellent benefits to the healthcare industry, its integration with other analytic tools such as Spark can make a huge difference at the patient care level as well as at the industry return on investment level.

References

Alexandru, A., Alexandru, C., Coardos, D., & Tudora, E. (2016). Healthcare, Big Data, and Cloud Computing. Management, 1, 2.

Basu, A. (2014). Real-Time Healthcare Analytics on Apache Hadoop* using Spark* and Shark. Retrieved from https://www.intel.com/content/dam/www/public/us/en/documents/white-papers/big-data-real-time-healthcare-analytics-whitepaper.pdf.

Beall, A. (2016). Big data in healthcare: How three organizations are using big data to improve patient care and more. Retrieved from https://www.sas.com/en_us/insights/articles/big-data/big-data-in-healthcare.html.

Dezyre. (2016). 5 Healthcare Applications of Hadoop and Big Data Retrieved from https://www.dezyre.com/article/5-healthcare-applications-of-hadoop-and-big-data/85.

Manyika, J., Chui, M., Brown, B., Bughin, J., Dobbs, R., Roxburgh, C., & Byers, A. H. (2011). Big data: The next frontier for innovation, competition, and productivity.

Shruika, D., & Kudale, R. A. (2018). Use of Big Data in Healthcare with Spark. International Journal of Science and Research (IJSR).

Wang, Y., Kung, L., & Byrd, T. A. (2018). Big data analytics: Understanding its capabilities and potential benefits for healthcare organizations. Technological Forecasting and Social Change, 126, 3-13.

Dr. Aly, O.
Computer Science

The standard MapReduce framework faced the challenge of the iterative computation which is required in various operations such as data mining, PageRank, network traffic analysis, graph analysis, social network analysis, and so forth (Bu, Howe, Balazinska, & Ernst, 2010; Sakr & Gaber, 2014).   These analyses techniques require the data to be processed iteratively until the computation satisfies a convergence or stropping condition (Bu et al., 2010; Sakr & Gaber, 2014).   Due to this limitation, and to this critical requirement, this iterative process is implemented and executed manually using a driver program when using the standard MapReduce framework (Bu et al., 2010; Sakr & Gaber, 2014).   However, the manual implementation and execution of such iterative computation have two major problems (Bu et al., 2010; Sakr & Gaber, 2014).  The first problem is reflected in loading unchanged data from iteration to iteration wasting input/output (I/O), network bandwidth, and CPU resources (Bu et al., 2010; Sakr & Gaber, 2014). The second problem is reflected in the overhead of the termination condition when the output of the application did not change for two consecutive iterations and reached a fixed point (Bu et al., 2010; Sakr & Gaber, 2014).  This termination condition may require an extra MapReduce job on each iteration which causes overhead for scheduling extra tasks, reading extra data from disk, and moving data across the network (Bu et al., 2010; Sakr & Gaber, 2014). 

Researchers exerted efforts to solve the iterative computation.  HaLoop is proposed by (Bu et al., 2010), and Twister by (Ekanayake et al., 2010), Pregel by (Malewicz et al., 2010).   One solution to the iterative computation limitation, as the case in HaLoop by (Bu et al., 2010) and Twister by  (Ekanayake et al., 2010) are to identify and keep invariant data during the iterations, where reading unnecessary data repeatedly is avoided.  The HaLoop by (Bu et al., 2010) implemented two caching functionalities (Bu et al., 2010; Sakr & Gaber, 2014).  The first caching technique is implemented on the invariant data in the first iteration and reusing them in a later iteration. The second caching technique is implemented on the outputs of reducer making the check for the fixpoint more efficient without adding any extra MapReduce job (Bu et al., 2010; Sakr & Gaber, 2014).

The solution of Pregel by (Malewicz et al., 2010) is more focused on the graph and was inspired by the Bulk Synchronous Parallel model (Malewicz et al., 2010).  This solution provides the synchronous computation and communication (Malewicz et al., 2010) and uses explicit messaging approach to acquire remote information and does not replicate remote values locally (Malewicz et al., 2010).  Mahoot is another solution that was introduced to solve the iterative computing by grouping a series of chained jobs to obtain the results (Polato, Ré, Goldman, & Kon, 2014).   In Mahoot solution, the result of each job is pushed into the next job until the final results are obtained (Polato et al., 2014).  The iHadoop proposed by (Elnikety, Elsayed, & Ramadan, 2011) schedules iterations asynchronously and connects the output of one iteration to the next allowing both to process their data concurrently (Elnikety et al., 2011).   The task scheduler of the iHadoop utilizes the inter-iteration data locality by scheduling tasks that exhibit a producer/consumer relation on the same physical machine allowing a fast transfer of the local data (Elnikety et al., 2011). 

Apache Hadoop and Apache Spark are the most popular technology for the iterative computation using in-memory data processing engine (Liang, Li, Wang, & Hu, 2011).  Hadoop defines the iterative computation as a series of MapReduce jobs where each job reads the data from Hadoop Distributed File System (HDFS) independently, processes the data, and writes the data back to HDFS (Liang et al., 2011).   Dacoop was proposed by Liang as an extension to Hadoop to handle the data-iterative applications, by using cache technique for repeatedly data processing and introducing shared memory-based data cache mechanism (Liang et al., 2011).  The iMapReduce is another solution proposed by (Zhang, Gao, Gao, & Wang, 2012) to provide support of iterative processing implementing the persistent tasks of the map and reduce during the whole iterative process and how the persistent tasks are terminated (Zhang et al., 2012).   The iMapReduce avoid three major overheads.  The first overhead is the job startup overhead which is avoided by building an internal loop from reduce to map within a job. The second overhead is the communication overhead which is avoided by separating the iterated state data from the static structure data.  The third overhead is the synchronization overhead which is avoided by allowing asynchronous map task execution (Zhang et al., 2012).

References

Bu, Y., Howe, B., Balazinska, M., & Ernst, M. D. (2010). HaLoop: Efficient iterative data processing on large clusters. Proceedings of the VLDB Endowment, 3(1-2), 285-296.

Ekanayake, J., Li, H., Zhang, B., Gunarathne, T., Bae, S.-H., Qiu, J., & Fox, G. (2010). Twister: a runtime for iterative MapReduce. Paper presented at the Proceedings of the 19th ACM international symposium on high performance distributed computing.

Elnikety, E., Elsayed, T., & Ramadan, H. E. (2011). iHadoop: asynchronous iterations for MapReduce. Paper presented at the Cloud Computing Technology and Science (CloudCom), 2011 IEEE Third International Conference on.

Liang, Y., Li, G., Wang, L., & Hu, Y. (2011). Dacoop: Accelerating data-iterative applications on Map/Reduce cluster. Paper presented at the Parallel and Distributed Computing, Applications, and Technologies (PDCAT), 2011 12th International Conference on.

Malewicz, G., Austern, M. H., Bik, A. J., Dehnert, J. C., Horn, I., Leiser, N., & Czajkowski, G. (2010). Pregel: a system for large-scale graph processing. Paper presented at the Proceedings of the 2010 ACM SIGMOD International Conference on Management of data.

Polato, I., Ré, R., Goldman, A., & Kon, F. (2014). A comprehensive view of Hadoop research—A systematic literature review. Journal of Network and Computer Applications, 46, 1-25.

Sakr, S., & Gaber, M. (2014). Large Scale and big data: Processing and Management: CRC Press.

Zhang, Y., Gao, Q., Gao, L., & Wang, C. (2012). imapreduce: A distributed computing framework for iterative computation. Journal of Grid Computing, 10(1), 47-68.

Dr. Aly, O.
Computer Science

Basic Concepts of the MapReduce Framework

In 2004, Google introduced MapReduce framework as a Parallel Processing framework which deals with large set of data (Bakshi, 2012; Fadzil, Khalid, & Manaf, 2012; White, 2012).  The MapReduce framework has gained much popularity because it has features for hiding sophisticated operations of the parallel processing (Fadzil et al., 2012).  Various MapReduce frameworks such as Hadoop were introduced because of the enthusiasm towards MapReduce (Fadzil et al., 2012). 

The capability of the MapReduce framework was realized by different research areas such as data warehousing, data mining, and the bioinformatics (Fadzil et al., 2012).  MapReduce framework consists of two main layers; the Distributed File System (DFS) layer to store data and the MapReduce layer for data processing (Lee, Lee, Choi, Chung, & Moon, 2012; Mishra, Dehuri, & Kim, 2016; Sakr & Gaber, 2014).  DFS is a major feature of the MapReduce framework (Fadzil et al., 2012).   

MapReduce framework is using large clusters of low-cost commodity hardware to lower the cost (Bakshi, 2012; H. Hu, Wen, Chua, & Li, 2014; Inukollu, Arsi, & Ravuri, 2014; Khan et al., 2014; Krishnan, 2013; Mishra et al., 2016; Sakr & Gaber, 2014; White, 2012).  MapReduce framework is using “Redundant Arrays of Independent (and inexpensive) Nodes (RAIN),” whose components are loosely coupled and when any node goes down there is no negative impact on the MapReduce job (Sakr & Gaber, 2014; Yang, Dasdan, Hsiao, & Parker, 2007).  MapReduce framework involves the “Fault-Tolerance” by applying the replication technique and allows replacing any crashed nodes with another node without affecting the current running job (P. Hu & Dai, 2014-7; Sakr & Gaber, 2014).  MapReduce framework involves the automatic support for the parallelization of execution which makes the MapReduce highly parallel and yet abstracted (P. Hu & Dai, 2014-7; Sakr & Gaber, 2014).  

The Top Three Features of Hadoop

The Hadoop Distributed File System (HDFS) is one of the major components of the Hadoop framework for storing large files (Bao, Ren, Zhang, Zhang, & Luo, 2012; CSA, 2013; De Mauro, Greco, & Grimaldi, 2015) and allowing access to data scattered over multiple nodes in without any exposure to the complexity of the environment (Bao et al., 2012; De Mauro et al., 2015).  The MapReduce programming model is another major component of the Hadoop framework (Bao et al., 2012; CSA, 2013; De Mauro et al., 2015) which is designed to implement the distributed and parallel algorithms efficiently (De Mauro et al., 2015).  HBase is the third component of Hadoop framework (Bao et al., 2012).  HBase is developed on the HDFS and is a NoSQL (Not only SQL) type database (Bao et al., 2012). 

The key features of Hadoop include the scalability and flexibility, cost efficiency and fault tolerance (H. Hu et al., 2014; Khan et al., 2014; Mishra et al., 2016; Polato, Ré, Goldman, & Kon, 2014; Sakr & Gaber, 2014).  Hadoop allows the nodes in the cluster to scale up and down based on the computation requirements and with no change in the data formats (H. Hu et al., 2014; Polato et al., 2014).  Hadoop also provides massively parallel computation to commodity hardware decreasing the cost per terabyte of storage which makes the massively parallel computation affordable when the volume of the data gets increased (H. Hu et al., 2014).  The Hadoop technology offers the flexibility feature as it is not tight with a schema which allows the utilization of any data either structured, non-structures, and semi-structured, and the aggregation of the data from multiple sources (H. Hu et al., 2014; Polato et al., 2014).  Hadoop also allows nodes to crash without affecting the data processing.  It provides fault tolerance environment where data and computation can be recovered without any negative impact on the processing of the data (H. Hu et al., 2014; Polato et al., 2014; White, 2012). 

Pros and Cons of MapReduce Framework

MapReduce was introduced to solve the problem of parallel processing of a large set of data in a distributed environment which required manual management of the hardware resources (Fadzil et al., 2012; Sakr & Gaber, 2014).  The complexity of the parallelization is solved by using two techniques:  Map/Reduce technique, and Distributed File System (DFS) technique (Fadzil et al., 2012; Sakr & Gaber, 2014).  The parallel framework must be reliable to ensure good resource management in the distributed environment using off-the-shelf hardware to solve the scalability issue to support any future requirement for processing (Fadzil et al., 2012).   The earlier frameworks such as the Message Passing Interface (MPI) framework was having a reliability issue and had a fault-tolerance issue when processing a large set of data (Fadzil et al., 2012).  MapReduce framework covers the two categories of the scalability; the structural scalability, and the load scalability (Fadzil et al., 2012).  It addresses the structural scalability by using the DFS which allows forming large virtual storage for the framework by adding off-the-shelf hardware.  MapReduce framework addresses the load scalability by increasing the number of the nodes to improve the performance (Fadzil et al., 2012). 

However, the earlier version of MapReduce framework faced challenges. Among these challenges are the join operation and the lack of support for aggregate functions to join multiple datasets in one task (Sakr & Gaber, 2014).  Another limitation of the standard MapReduce framework is found in the iterative processing which is required for analysis techniques such as PageRank algorithm, recursive relational queries, and social network analysis (Sakr & Gaber, 2014).  The standard MapReduce does not share the execution of work to reduce the overall amount of work  (Sakr & Gaber, 2014).  Another limitation was found in the lack of support of data index and column storage but support only for a sequential method when scanning the input data. Such a lack of data index affected the query performance (Sakr & Gaber, 2014).  Moreover, many argued that MapReduce is not regarded to be the optimal solution for structured data.   It is known as shared-nothing architecture, which supports scalability (Bakshi, 2012; Jinquan, Jie, Shengsheng, Yan, & Yuanhao, 2012; Sakr & Gaber, 2014; White, 2012), and the processing of large unstructured data sets (Bakshi, 2012).  MapReduce has the limitation of performance and efficiency (Lee et al., 2012).

References

Bakshi, K. (2012, 3-10 March 2012). Considerations for big data: Architecture and approach. Paper presented at the Aerospace Conference, 2012 IEEE.

Bao, Y., Ren, L., Zhang, L., Zhang, X., & Luo, Y. (2012). Massive sensor data management framework in cloud manufacturing based on Hadoop. Paper presented at the Industrial Informatics (INDIN), 2012 10th IEEE International Conference on.

CSA, C. S. A. (2013). Big Data Analytics for Security Intelligence. Big Data Working Group.

De Mauro, A., Greco, M., & Grimaldi, M. (2015). What is big data? A consensual definition and a review of key research topics. Paper presented at the AIP Conference Proceedings.

Fadzil, A. F. A., Khalid, N. E. A., & Manaf, M. (2012). Performance of scalable off-the-shelf hardware for data-intensive parallel processing using MapReduce. Paper presented at the Computing and Convergence Technology (ICCCT), 2012 7th International Conference on.

Hu, H., Wen, Y., Chua, T.-S., & Li, X. (2014). Toward scalable systems for big data analytics: A technology tutorial. IEEE Access, 2, 652-687.

Hu, P., & Dai, W. (2014-7). Enhancing fault tolerance based on Hadoop cluster. International Journal of Database Theory and Application, 7(1), 37-48.

Inukollu, V. N., Arsi, S., & Ravuri, S. R. (2014). Security issues associated with big data in cloud computing. International Journal of Network Security & Its Applications, 6(3), 45.

Jinquan, D., Jie, H., Shengsheng, H., Yan, L., & Yuanhao, S. (2012). The Hadoop Stack: New Paradigm for Big Data Storage and Processing. Intel Technology Journal, 16(4), 92-110.

Khan, N., Yaqoob, I., Hashem, I. A. T., Inayat, Z., Mahmoud Ali, W. K., Alam, M., . . . Gani, A. (2014). Big Data: Survey, Technologies, Opportunities, and Challenges. The Scientific World Journal, 2014.

Krishnan, K. (2013). Data warehousing in the age of big data: Newnes.

Lee, K.-H., Lee, Y.-J., Choi, H., Chung, Y. D., & Moon, B. (2012). Parallel data processing with MapReduce: a survey. ACM SIGMOD Record, 40(4), 11-20.

Mishra, B. S. P., Dehuri, S., & Kim, E. (2016). Techniques and Environments for Big Data Analysis: Parallel, Cloud, and Grid Computing (Vol. 17): Springer.

Polato, I., Ré, R., Goldman, A., & Kon, F. (2014). A comprehensive view of Hadoop research—A systematic literature review. Journal of Network and Computer Applications, 46, 1-25.

Sakr, S., & Gaber, M. (2014). Large Scale and big data: Processing and Management: CRC Press.

White, T. (2012). Hadoop: The definitive guide: ” O’Reilly Media, Inc.”.

Yang, H.-c., Dasdan, A., Hsiao, R.-L., & Parker, D. S. (2007). Map-reduce-merge: simplified relational data processing on large clusters. Paper presented at the Proceedings of the 2007 ACM SIGMOD international conference on Management of data.