Kibana―Knowing the ELK Basics

Figure 1.  ELK Stack

The most popular usage of Elasticsearch today is Log Management. It is developed alongside a data-collection and log-parsing engine called Logstash, and an analytics and visualization platform called Kibana. The three products are designed for use as an integrated solution, referred to as the "Elastic Stack" (formerly the "ELK stack"):

Logstash is the workhorse that collects the log files from Application Servers, parses them, formats them and sends them to Elastic Search. Elastic Search stores and indexes the data, which is presented by Kibana. The end users access Kibana Web Interface to view the data. 

One of the great things about Elasticsearch is its extensive JSON based REST API which allows you to integrate, manage and query the indexed data in countless different ways.

Elastic Search 

Elastic Search is a robust search and analytics tool that stores data in a document oriented data store.  It has the following features:

• Is built on top of high performance open source search engine Apache Lucene (in Java)
• Started off as scalable Lucene
• Horizontally scalable search engine
• Often a faster solution than Hadoop/Spark/Flink/etc.
• Document oriented storage
• The document oriented storage differs sharply from traditional table oriented RDBMS (Such as Oracle, MS SQL Server).
• With document oriented data storage, data is stored as structured JSON documents.
• Every field is indexed by default. 
• This is why the search speed is incredible.
• Each shard is an inverted index of documents.
• Consumes data from Logstash 
• Creates Indexes for log files, typically a date-based index
• Its architecture favors distribution
• You can scale your Elastic Search infrastructure massively and seamlessly.

You can get the latest version of Elasticsearch from elastic.co/downloads/elasticsearch.  Once you have an instance of ElasticSearch up and running you can talk to it using it's REST API residing at localhost port 9200.  For example, you can run the following curl command:

curl 'http://localhost:9200/?pretty'

to get a response like this:^[2]

{
  "name" : "Tom Foster",
  "cluster_name" : "elasticsearch",
  "version" : {
    "number" : "2.1.0",
    "build_hash" : "72cd1f1a3eee09505e036106146dc1949dc5dc87",
    "build_timestamp" : "2015-11-18T22:40:03Z",
    "build_snapshot" : false,
    "lucene_version" : "5.3.1"
  },
  "tagline" : "You Know, for Search"
}

Video 2.  Logstash Overview

Logstash

Logstash, another open source tool does the heavy lifting of consuming the logs from various systems and sends them to Elastic Search.  It has the following features:

• Is a tool for collecting & monitoring logs from remote machines
• Data pushed and stored in Elastic Search
• Is a data pipeline for Elasticsearch
• Parses, transforms, and filters data as it passes through
• Can derive structure from unstructured data
• Can anonymize personal data or exclude it entirely
• Can do geo-location lookups
• Guarantees at-least-once delivery
• Absorbs throughput from load spikes
• Can scale across many nodes

Kibana

Kibana is an open source data  exploration & visualization platform that is the presentation layer in the ELK stack. It consumes data from Elastic Search Indexes. A user accesses Kibana interface via a web browser.

• Used for 
• Log and time series analytics, application monitoring & operational intelligence
• Make queries in Elastic Search
• Enables the searching & interaction with data in Elastic Search
• Allows performing advanced analytics & creation of reports
• Provide Real-time Dashboard
• Enables creation & sharing of dynamic dashboards that get updated in realtime

The default settings configure Kibana to run on localhost:5601. To change the host or port number, or connect to Elasticsearch running on a different machine, you'll need to update your kibana.yml file.  For more information, read:

• Configuring Kibana
• Getting Kibana Up and Running

References

1. Basic Concepts (Apache  Lucene)
2. Installing and Running Elasticsearch
3. 15 Minutes Introduction to ELK (Elastic Search,LogStash,Kibana)
4. Configuring Kibana
5. Getting Kibana Up and Running
6. Filter Plugins (Logstach)
7. Elasticsearch Architecture (youtube)

HiBench Suite―How to Build and Run the Big Data Benchmarks

As known from a previous article:

Three Benchmarks for SQL Coverage in HiBench Suite ― a Bigdata Micro Benchmark Suite

HiBench Suite is a big data benchmark suite that helps evaluate different big data frameworks in terms of speed, throughput and system resource utilization.

When your big data platform (e.g.,e HDP) evolves, it comes times that you need to upgrade your benchmark suite accordingly.

In this article, we will cover how to pick up the latest HiBench Suite (i.e., version 6.1) to work with Spark 2.1.

HiBench Suite

To download the master branch of HiBench Suite (click the diagram to enlarge), you can visit its home page here . On 06/18/2017, its latest version is 6.1.

To download, we have selected "Download ZIP" and saved it to our Linux system.

Maven

From the home page, you can select "docs" link to view all available document links:

• build-hibench.md
• run-sparkbench.md
• etc

From the build-hibench.md link, it tells you how to build HiBench Suite using Maven. For example, if you want to build all workloads in HiBench, you use the below command:

mvn -Dspark=2.1 -Dscala=2.11 clean package

This could be time consuming because the hadoopbench (one of the workload) relies on 3rd party tools like Mahout and Nutch. The build process automatically downloads these tools for you. If you won't run these workloads, you can only build a specific framework (e.g., sparkbench) to speed up the build process.

To get familiar with Maven, you can start with this pdf file. In it, you will learn how to download Maven and how to setup system to run it. Here we will just discuss some issues that we have run into while building all workloads using Maven.

Maven Installation Issues and Solutions

Proxy Server

Since our Linux system sits behind the firewall, we need to set up the following environment variables:

export http_proxy=http://your.proxy.com:80/
export https_proxy=http://your.proxy.com:80/

Environment Setup

As instructed in pdf file, we have setup below additional environment variables:

export JAVA_HOME=~/JVMs/8u40_fcs

export PATH=/scratch/username/maven/apache-maven-3.5.0/bin:$PATH

export PATH=$JAVA_HOME/bin:$PATH

Maven Configuration & Debugging

POM stands for Project Object Model. which

• Is the Fundamental Unit of Work in Maven
• Is an XML file
• Always resides in the base directory of the project as pom.xml.

The POM contains information about the project and various configuration detail used by Maven to build the project(s).

In the default ~/.m2/settings, we have set the following entries for POM:

<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"

          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

          xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd">

  <localRepository>/scratch/username/.m2/repository</localRepository>

  <server>

    <id>central</id>

    <configuration>

      <httpConfiguration>

        <all>

          <connectionTimeout>120000</connectionTimeout>

          <readTimeout>120000</readTimeout>

        </all>

      </httpConfiguration>

    </configuration>

  </server>

First we have set the localRepository to a new location because an issue described here.^[7,8] Secondly, we have set longer timeout for both connection and read.

If you have run into issues with a plugin, you can use "help:describe"

mvn  help:describe -Dplugin=com.googlecode.maven-download-plugin:maven-download-plugin

to display a list of its attributes and goals for debugging.

How to Run Sparkbench

To learn how to run a specific benchmark named sparkbench, you can click on the document link below:

run-sparkbench.md

Without much ado, we will focus on the configuration and tuning part of the task. For other details, please refer to the document.

New Configuration Files

In the new HiBench, there are two levels of configuration:

(Global level)

${hibench.home}/conf/hadoop.conf 
${hibench.home}/hibench.conf 
${hibench.home}/conf/spark.conf

(Workload level)

${hibench.home}/conf/workloads/micro/terasort.conf  

It has also introduced a new hierarchy (i.e. category like micro, websearch, sql, etc) to organize workload runtime scripts:

${hibench.home}/<benchmark>/<framework>
  where <benchmark> could be:
    micro/terasort
    websearch/pagerank
    sql/aggregation
    sql/join
    sql/scan
  where <framework> could be:
    spark
    hadoop
    prepare

Similarly for the workload-specific configuration file, they are stored under the new category level:

${hibench.home}/conf/workloads/${benchmark.conf}
  where <benchmark.conf> could be:
    micro/terasort.conf
    websearch/pagerank.conf
    sql/aggregation.conf
    sql/join.conf
    sql/scan.conf

References

1. HORTONW0RKS DATA PLATFORM (HDP®)
2. Readme (HiBench 6.1)
3. HiBench Download
4. How to build HiBench (HiBench 6.1)
5. How to run sparkbench (HiBench 6.1)
6. How-to documents (HiBench 6.1)
7. Idiosyncrasies of ${HOME} that is an NFS Share (Xml and More)
8. Apache Maven Build Tool (pdf)
9. How do I set the location of my local Maven repository?
10. Guide to Configuring Plug-ins (Apache Maven Project)
11. Available Plugins (Apache Maven Project)
12. MojoExecutionException
13. Installing Maven Plugins (SourceForge.net)
14. Download Plugin For Maven » 1.2.0
15. Group: com.googlecode.maven-download-plugin

Spark SQL一Knowing the Basics

There are two ways to interact with SparkSQL:^[1]

• As a distributed SQL query engine (i.e. via Spark-shell)
• When spark-shell is run,
  • it automatically creates a SparkContext and a HiveContext
• Variants of Spark shell
• spark-shell for Scala
• pyspark for Python
• As a library
• Which operates through one of the following two APIs and the Spark Thrift Server:
• JDBC
• using Java code or via the Beeline JDBC client
• ODBC
• via the Simba ODBC driver

Features of SparkSQL

SparkSQL is one of Spark's modules, which provides SQL Interface to Spark. Below is a list of SparkSQL functionalities:

• Supports both schema-on-write and schema-on-read
• schema-on-write
• Requires data to be modeled before it can be stored (hint: traditional database systems)
• SparkSQL supports schema-on-write through columnar formats such as Parquet and ORC (Optimized Row Columnar)
• schema-on-read
• A schema is applied to data when it is read
• A user can store data in its native format without worrying about how it will be queried
• It not only enables agility but also allows complex evolving data
• One disadvantage of a schema-on-read system is that queries are slower than those executed on data stored in a schema-on-write system.^[19]
• Has a SQLContext and a HiveContext
• SQLContext (i.e., org.apache.spark.sql.SQLContext)
• Can read data directly from the filesystem
• This is useful when the data you are trying to analyze does not reside in Hive (for example, JSON files stored in HDFS).^[13]
• HiveContext (or org.apache.spark.sql.hive.HiveContext)
• Is a super set of the SQLContext (or HiveContext extends SQLContext)
• Read data by interacting with the Hive MetaStore
• For example, to start using ORC, you need to define a HiveContext instance first.^[12]
• Compatible with Apache Hive
• Not only supports HiveQL, but can also access Hive metastore, SerDes (i.e. Hive serialization and deserialization libraries), and UDFs (i.e., user-defined functions)
• HiveQL queries run much faster on Spark SQL than on Hive
• Existing Hive workloads can be easily migrated to Spark SQL.
• You can use Spark SQL with or without Hive
• Can be configured to read Hive metastores created with different versions of Hive
• Prepackaged with a Thrift/JDBC/ODBC server
• A client application can connect to this server and submit SQL/HiveQL queries using Thrift, JDBC, or ODBC interface
• Bundled with Beeline
• Which can be used to submit HiveQL queries

Architecture of SparkSQL

The architecture of SparkSQL contains three layers:

• Data Sources
• Schema RDD
• Language API

,

Data Sources

Usually the Data Source for spark-core is a text file, Avro file, etc. However, Spark SQL operates on a variety of data sources through the DataFrame (see details below). The default data source is parquet unless otherwise configured by spark.sql.sources.default, which is used when

• Reading or writing data in DataFrameWriter and DataFrameReader
• Creating external table from a path (in Catalog.createExternalTable)
• Streaming DataStreamReader and DataStreamWriter

Some of data sources supported by SparkSQL are listed below:

• JSON Datasets
• Spark SQL can automatically capture the schema of a JSON dataset and load it as a DataFrame.
• Hive Tables
• Hive comes bundled with the Spark library as HiveContext
• Parquet Files
• Use a columnar format
• Cassandra database

Read here for the methods of loading and saving data using the Spark Data Sources and options that are available for the built-in data sources.

Schema RDD (or DataFrame)

Spark Core is designed with special data structure called RDD (a native data structure of Spark). However, Spark SQL works on schemas, tables, and records via SchemaRDD, which was later renamed as “DataFrame” API.

With a SQLContext, applications can create DataFrame from an array of different sources such as:

• Hive tables
• Structured Data files
• External databases
• Existing RDDs.

It an also be registered as a temporary table. Registering a DataFrame as a table allows you to run SQL queries over its data.

Here is the summary of DataFrame API:

• DataFrame vs RDD
• DataFrame stores much more information about the structure of the data, such as the data types and names of the columns, than RDD.
• This allows the DataFrame to optimize the processing much more effectively than Spark transformations and Spark actions doing processing on RDD.
• Once data has been transformed into a Data Frame with a schema, It can then be stored in
• Hive (for persistence)
• If it needs to be accessed on a regular basis or registered
• Temp table(s)
• Which will exist only as long as the parent Spark application and it's executors (the application can run indefinitely)
• Conversions from RDD to DataFrame and vice versa
• Many of the Spark transformations and Spark actions that are typically applied on RDDs can also be applied on DataFrames.
• Spark SQL supports two different methods for converting existing RDDs into DataFrames
• Inferring the schema using reflection
• Programmatically specifying the schema
• Registration of DataFrames as Tables
• An existing RDD can be implicitly converted to a DataFrame and then be registered as a table.
• All of the tables that have been registered can then be made available for access as a JDBC/ODBC data source via the Spark thrift server.
• Supports big datasets (up to Petabytes)
• Supports different data formats and storage systems
• Data formats
• Avro, csv, elastic search, and Cassandra
• Storage systems
• HDFS, HIVE tables, mysql, etc.
• Provides language APIs for Python, Java, Scala, and R Programming
• It also achieves consistent performance of DataFrame API calls across languages using the state of art optimization and code generation through the Spark SQL Catalyst optimizer (tree transformation framework)

Language API

Spark SQL comes prepackaged with a Thrift/JDBC/ODBC server. A client application can connect to this server and submit SQL/HiveQL queries using Thrift, JDBC, or ODBC interface. It translates queries written using any of these interfaces into MapReduce, Apache Tez and Spark jobs.

Spark is compatible with different languages. All the supported programming languages of Spark can be used to develop applications using the DataFrame API of Spark SQL. For example, Spark SQL supports the following language APIs:

• Python
• Scala
• Java
• R
• HiveQL
• Is an SQL-like language with schema on read and transparently converts queries to MapReduce, Apache Tez and Spark jobs.
• An SQL-dialect with differences in structure and working.
• The differences are mainly because Hive is built on top of the Hadoop ecosystem and has to comply with the restrictions of Hadoop and MapReduce.

Finally, Spark SQL API consists of three key abstractions (as described above):

• SQLContext
• HiveContext
• DataFrame

References

1. Using Spark SQL (Hortonworks)
2. Setting Up HiveServer2 (Apache Hive)
3. HiveServer2 (slideshare.net)
4. Hive Metastore Administration (Apache)
5. HiveServer2 Overview (Apache)
6. SQLLine 1.0.2
7. Hadoop Cluster Maintenance
8. Big Data Analytics with Spark: A Practitioner’s Guide to Using Spark for Large-Scale Data Processing, Machine Learning, and Graph Analytics, and High-Velocity Data Stream Processing
9. Apache Hive—Hive CLI vs Beeline (Xml And More)
• Beeline is a JDBC client based on the SQLLine CLI — although the JDBC driver used communicates with HiveServer2 using HiveServer2’s Thrift APIs.
10. Apache Hive Essentials
11. Three Benchmarks for SQL Coverage in HiBench Suite ― a Bigdata Micro Benchmark Suite
12. Accessing ORC Files from Spark (Hortonworks)
13. Using the Spark DataFrame API
14. Spark Shell — spark-shell shell script
15. Tuning Spark (Hortonworks)
16. Spark SQL, DataFrames and Datasets Guide (Spark 1.6.1)
17. spark.sql.sources.default (default: parquet)
• Sets the default data source to use in input/output
• Is used when reading or writing data in DataFrameWriter and DataFrameReader, when creating external table from a path (in Catalog.createExternalTable) and in the streaming DataStreamReader and DataStreamWriter.
18. Mastering Apache Spark
19. Three Benchmarks for SQL Coverage in HiBench Suite ― a Bigdata Micro Benchmark Suite
20. Deep Dive Into Catalyst: Apache Spark 2.0’s Optimizer (slideshare.net)
21. Accessing Spark SQL through JDBC and ODBC (Hortonworks)
22. Using Spark to Virtually Integrate Hadoop with External Systems
• This article focuses on how to use SparkSQL to integrate, expose, and accelerate multiple sources of data from a single "Federation Tier".

Spark on YARN: Sizing Executors and Other Tuning Ideas

Spark on YARN leverages YARN services for resource allocation, runs Spark executors in YARN containers, and supports workload management and Kerberos security features. It supports two modes:^[2]

• YARN-cluster mode
• Optimized for long-running production jobs
• YARN-client mode
• Best for interactive use such as prototyping, testing, and debugging
• Spark shell and the Spark Thrift server run in YARN-client mode. 

In this article, we will use YARN-cluster mode for illustration (see Figure 1).  Before jump in the tuning part, you should read [1] first.

Figure 1.  YARN-clsuter mode

Spark Application

When tuning Spark applications, it is important to understand how Spark works and what types of resources your application requires. For example, machine learning tasks are usually CPU intensive, whereas extract-transform-load (ETL) operations are I/O intensive.

Using PageRank benchmark in HiBench as an example, a spark application can be submitted with following command line:^[6]

$ /usr/hdp/current/spark-client/bin/spark-submit  

--properties-file ./BDCSCE-HiBench/report/pagerank/spark/scala/conf/sparkbench/spark.conf --class org.apache.spark.examples.SparkPageRank --master yarn-cluster --num-executors 17 --executor-cores 5 --executor-memory 19G --driver-memory 2G ./BDCSCE-HiBench/src/sparkbench/target/sparkbench-5.0-SNAPSHOT-MR2-spark1.6-jar-with-dependencies.jar 

"properties-file" option  allow you to provide a path to a file from which to load extra properties.  However, the most important tuning parts (shown on the command line) are to size up your executors right:

• --num-executors
• Number of Spark executors to launch (default: 2)
• --executor-cores
• Number of cores per executor. (Default: 1 in YARN mode, or all available cores on the worker in standalone mode)
• --executor-memory
• Memory per executor (e.g. 1000M, 2G) (Default: 1G).

Figure 2.  spark.shuffle.manager = sort^[5]

Tuning Guidelines

To do our tuning exercise, we use the following YARN cluster as an example:

• Cluster Size: Total 6 Nodes
• 16 Cores each
• 64 GB or RAM each
• Total Cores
• 96 
• Total Memory
• 384 GB

Using [3] as our reference, here are the tuning guidelines:

• Want to run multiple tasks in the same JVM (see Figure 2)
• Need to leave some memory overhead for OS/Hadoop daemons
• Need some overhead for off heap memory
• Configured by spark.yarn.executor.memory.overhead
• Default: max (384MB, .07 * spark.executor.memory)
• YARN Application Master needs a core in both  YARN modes:
• YARN-client mode
• Spark Driver is run in the Application client process
• YARN-cluster mode
• Spark Driver in run in the Spark ApplicationMaster process
• Optimal HDFS I/O Throughput
• Best is to keep 5 cores per executor
• No Spark shuffle block can be greater than 2 GB
• Read [3] for explanation
• Spark SQL
• Especially problematic for Spark SQL
• Default number of partitions to use when doing shuffle is 200
• Low number of partitions leads to
• high shuffle block size and could exceed the 2GB limit
• not making good use of parallelism
• Rule of thumb is around 128 MB per partition

Calculations

Based on the guidelines, here are the steps to size executors:

• 5 cores per executor
• For max HDFS throughput
• Cluster has 6 x 15 = 90 cores in total
• After taking out Hadoop/YARN daemon cores
• 90 cores / 5 cores per executor = 18 executors
• 1 executor for ApplicationMaster => 17 executors
• Each node has 3 executors
• 63 GB/3 = 21 GB per executor
• 21 x (1-0.07) ~ 19 GB (counting off heap overhead)

Correct Answer

The final settings are shown in our previous "spark-sumbit" command line, which has the following values:

• 17 executors
• Note that Executors are not be released until the job finishes, even if they are no longer in use. Therefore, do not overallocate executors above the estimated requirements
• 19 GB memory each
• 5 cores each

Note also that you should use the settings above as the starting point and fine tune them and/or other parameters further based on your network/disk capabilities and other factors (e.g., CPU intensive job or I/O intensive job).

Extra Tuning Ideas

Based on [4,6,9], here are more tuning ideas:

• --executor-cores or spark.executor.cores
• HDFS client could have trouble with tons of concurrent threads. A rough guess is that at most 5  tasks per executor can achieve full write throughput, so it’s good to keep the number of cores per executor equal to or below 5.
• --executor-memory or spark.executor.memory
• Controls the executor heap size, but JVMs can also use some memory off heap, for example for interned Strings and direct byte buffers. 
• spark.yarn.executor.memoryOverhead
• The value of the spark.yarn.executor.memoryOverhead property is added to the executor memory to determine the full memory request to YARN for each executor.
• Need to increase memoryOverhead size if you see:
• Container killed by YARN for exceeding memory limits. 7.0 GB of 7 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead
• --driver-memory
• Driver memory does not need to be large if the job does not aggregate much data (as with a collect() action)
• --num-executors vs --executor-memory
• There are tradeoffs between num-executors and executor-memory:
• Large executor memory does not imply better performance, due to JVM garbage collection. Sometimes it is better to configure a larger number of small JVMs than a small number of large JVMs.
• Running executors with too much memory often results in excessive garbage collection delays. 
• 64GB is a rough guess at an upper limit for a single executor
• spark.serializer
• Consider switching from the default serializer to the Kryo serializer (i.e., org.apache.spark.serializer.KryoSerializer) to improve performance
• spark.executor.extraJavaOptions
• Always print out the details of GC events with the following settings to aid debugging:
• -XX:+PrintGCTimeStamps -XX:+PrintGCDetails -Xloggc:executor_gc.log 
• Note that executor_gc.log can be found on  Data Nodes where executors are running on
• spark.network.timeout
• Default timeout for all network interactions. This config will be used in place of 
• spark.core.connection.ack.wait.timeout
• spark.storage.blockManagerSlaveTimeoutMs
• spark.shuffle.io.connectionTimeout
• spark.rpc.askTimeout
• spark.rpc.lookupTimeout if they are not configured.
• If you run into the following timeouts:
• Executor heartbeat timed out after 121129 ms
• Consider set spark.network.tiemout to be higher value especially when you have heavy workloads^[9]
• spark.local.dir
• If you run into the following exception:
• java.io.IOException: No space left on device
• The best way to resolve this issue and to boost performance is to give as many disks as possible to handle scratch space disk IO
• Because Spark constantly writes to and reads from its scratch space, disk IO can be heavy and can slow down your workload
• Consider explicitly define parameter spark.local.dir in spark-defaults.conf configuration file to be something like:
• spark.local.dir   /data1/tmp,/data2/tmp, etc.
• yarn.nodemanager.resource.memory-mb 
• Controls the maximum sum of memory used by the containers on each node.
• yarn.nodemanager.resource.cpu-vcores 
• Controls the maximum sum of cores used by the containers on each node
• yarn.scheduler.minimum-allocation-mb 
• Controls the minimum request memory value
• yarn.scheduler.increment-allocation-mb 
• Control the increment request memory value
• Tuning parallelism
• Every Spark stage has a number of tasks, each of which processes data sequentially. In tuning Spark jobs, this number is probably the single most important parameter in determining performance.
• Read [4] for good tuning advice

References

1. Apache YARN一Knowing the Basics
2. Apache Spark Component Guide
3. Top 5 Mistakes to Avoid When Writing Apache Spark Applications
• How to adjust number of shuffle partitions
• Spark SQL
• use spark.sql.shuffle.partitions
• Others
• use rdd.repartition() or rdd.coalesce()
4. How-to: Tune Your Apache Spark Jobs (Part 2)
• The two main resources that Spark (and YARN) think about are CPU and memory. Disk and network I/O, of course, play a part in Spark performance as well, but neither Spark nor YARN currently do anything to actively manage them.
5. Spark Architecture: Shuffle
6. Tuning Spark (Hortonworkds)
7. Spark 2.1.0 Configuration
8. Spark 1.6.1 Configuration
9. Troubleshooting and Tuning Spark for Heavy Workloads
10. Untangling Apache Hadoop YARN, Part 4: Fair Scheduler Queue Basics (Cloudera)
11. Untangling Apache Hadoop YARN, Part 5: Using FairScheduler queue properties (Cloudera)
12. Untangling Apache Hadoop YARN, Part 3: Scheduler Concepts (Cloudera)
13. Tuning Spark (Apache Spark 2.1.0)

Apache YARN一Knowing the Basics

Apache Hadoop YARN is a modern resource-management platform that can host multiple data processing engines for various workloads like batch processing (MapReduce), interactive (Hive, Tez, Spark) and real-time processing (Storm). These applications can all co-exist on YARN and share a single data center in a cost-effective manner with the platform worrying about resource management, isolation and multi-tenancy.

In this article, we will use Apache Hadoop YARN provided by Hortonworks in the discussion.

Figure 1 Apache Hadoop YARN (Hortonworks)

Apache Hadoop YARN

HDFS and YARN form the data management layer of Apache Hadoop. YARN is the architectural center of Hadoop,   As its architectural center, YARN enhances a Hadoop compute cluster in the following ways:^[1]

• Multi-tenancy
• Cluster utilization
• Scalability
• Compatibility

For example, YARN takes into account all the available compute resources on each machine in the cluster. Based on the available resources, YARN will negotiate resource requests from applications (such as MapReduce or Spark) running in the cluster. YARN then provides processing capacity to each application by allocating Containers.

YARN runs processes on a cluster一two or more hosts connected by a high-speed local network一similarly to the way an operating system runs processes on a standalone computer. Here we will present YARN from two perspectives:

• Resource management
• Process management

Figure 2. Basic Entities in YARN

Resource Management

In a Hadoop cluster, it’s vital to balance the usage of RAM, CPU and disk so that processing is not constrained by any one of these cluster resources.  YARN is the foundation of the new generation of Hadoop.

In its design, YARN split up the two old responsibilities of the JobTracker and TaskTracker (see the diagram in [2]) into separate entities:

• A global ResourceManager (instead of a cluster manager)
• A per-application master process一ApplicationMaster (instead of a dedicated and short-lived JobTracker)
• Started on a container by the ResourceManager's launcher
• Per-node worker process一NodeManager (instead of TaskTracker)
• Per-application Containers
• Controlled by the NodeManagers
• Can run different kinds of tasks (also Application Masters)
• Can be configured to have different sizes (e.g., RAM, CPU)

The ResourceManager is the ultimate authority that arbitrates resources among all applications in the system. The ApplicationMaster is a framework-specific entity that negotiates resources from the ResourceManager and works with the NodeManager(s) to execute and monitor the component tasks.

Containers are an important YARN concept. You can think of a container as a request to hold resources on the YARN cluster. Currently, a container hold request consists of vcore and memory, as shown in the Figure 3 (left).  Once a hold has been granted on a node, the NodeManager launches a process called a task which runs in a container . The right side of Figure 3 shows the task running as a process inside a container.

Figure 3.  Resource-to-Process Mapping from Container's Perspective

Process Management

Let us see how YARN manages its processes/tasks from two perspectives:

• YARN scheduler
• By default, there are 3 schedulers currently provided with YARN:
• FIFO, Capacity and Fair 
• YARN model of computation
• Will use Apache Spark to Illustrate how a Spark job fits into the YARN model of computation

Figure 4.  Application Submission in YARN

YARN Scheduler^[6]

The ResourceManager has a scheduler, which is responsible for allocating resources to the various applications running in the cluster, according to constraints such as queue capacities and user limits. The scheduler schedules based on the resource requirements of each application.

An application (e.g. Spark application) is a YARN client program which is made up of one or more tasks.  For each running application, a special piece of code called an ApplicationMaster helps coordinate tasks on the YARN cluster. The ApplicationMaster is the first process run after the application starts.

Here are the sequence of events happening when an application starts (see Figure 4):

1. The application starts and talks to the ResourceManager for the cluster
• The ResourceManager makes a single container request on behalf of the application
2. The ApplicationMaster starts running within that container
3. The ApplicationMaster requests subsequent containers from the ResourceManager that are allocated to run tasks for the application. 
4. The ApplicationMaster launches tasks in the container and coordinates the execution of all tasks within the application 
• Those tasks do most of the status communication with the ApplicationMaster
5. Once all tasks are finished, the ApplicationMaster exits. The last container is de-allocated from the cluster.
6. The application client exits

Figure 5.  A Spark Job Consists of Multiple Tasks

YARN Model of Computation

In the Spark paradigm, an application consists of Map tasks, Reduce tasks, etc.^[7] Spark tasks align very cleanly with YARN tasks.

At the process level, a Spark application is run as:

• A single driver process 
• Manages the job flow and schedules tasks and is available the entire time the application is running
• Could be the same as the client process (i.e., YARN-client mode) or run in the cluster (i.e, YARN-cluster mode; see Figure 6)
• A set of executor processes 
• Scattered across nodes on the cluster (see Figure 5)
• Are responsible for executing the work in the form of tasks, as well as for storing any data that the user chooses to cache
• Multiple tasks can run within the same executor

Deploying these processes on the cluster is up to the cluster manager in use (YARN in this article), but the driver and executor themselves exist in every Spark application.

Recap

Tying two aspects (i.e., Resource and Process Management) together, we conclude that:

• A global ResourceManager runs as a master daemon, usually on a dedicated machine, that arbitrates the available cluster resources among various competing applications. 
• The ResourceManager tracks how many live nodes and resources are available on the cluster and coordinates what applications submitted by users should get these resources and when. 
• The ResourceManager is the single process that has the above information so it can make its allocation (or rather, scheduling) decisions in a shared, secure, and multi-tenant manner (for instance, according to an application priority, a queue capacity, ACLs, data locality, etc.)
• Each ApplicationMaster (i.e., a per-application master process) has responsibility for negotiating appropriate resource containers from the scheduler, tracking their status, and monitoring their progress. 
• The NodeManager is the per-node worker process, which is responsible for launching the applications’ containers, monitoring their resource usage (cpu, memory, disk, network) and reporting the same to the ResourceManager.

Figure 6 presents the consolidated view of how a spark application is run in YARN-cluster mode:

Figure 6.  Spark in YARN-cluster mode^[3]

References

1. Apache Hadoop Yarn (Hortonworks)
2. Hadoop MapReduce一Knowing the Basics (Xml and More)
3. Apache Spark Resource Management and YARN App Models (Cloudera)
4. HOW TO PLAN AND CONFIGURE YARN AND MAPREDUCE 2 IN HDP 2.0 (Hortonworks)
5. Untangling Apache Hadoop YARN, Part 1: Cluster and YARN Basics (Cloudera)
6. Introduction to Apache YARN Schedulers & Queues
7. Spark Architecture Shuffle
8. Apache Spark (Hortonworks)
9. Introduction to YARN (IBM)

Hadoop MapReduce一Knowing the Basics

MapReduce can means two different things:

• Programming Model
• If you can rewrite algorithms into Maps and Reduces, and your problem can be broken up into small pieces solvable in parallel, then MapReduce might be a potential distributed problem solving approach to your large datasets.
• See [1] for a sample MapReduce application
• Software Framework
• A framework such as Hadoop MapRedue breaks up large data into smaller parallelizable chunks and handles scheduling
• Alternative一Apache Tez can process certain workloads more efficiently than MapReduce

In this article, we will learn two kinds of Hadoop MapReduce frameworks provided on Apache Hadoop:

• MapReduce 1 (MR1) 
• YARN (MR2)

We will start the introduction of Hadoop MapReduce using MR1 and then briefly with MR2.

MapReduce Job

A MapReduce job usually splits the input data-set into independent chunks which are processed by the map tasks in a completely parallel manner. The framework sorts the outputs of the maps, which are then input to the reduce tasks. Typically both the input and the output of the job are stored in a file-system (e.g., HDFS on Apache Hadoop). The framework takes care of scheduling tasks, monitoring them and re-executes the failed tasks.

(input)  -> map -> -> combine -> -> reduce ->  (output)

MapReduce job is useful for batch processing on terabytes or petabytes of data stored in Apache Hadoop and has the following characteristics:^[6]

• Cannot control the order in which the maps or reductions are run
• For maximum parallelism, you need Maps and Reduces to not depend on data generated in the same MapReduce job (i.e. stateless) 
• A database with an index will always be faster than a MapReduce job on unindexed data
• Reduce operations do not take place until all Maps are complete (or have failed then been skipped) 
• General assumption that the output of Reduce is smaller than the input to Map一large datasource used to generate smaller final values

Hadoop MapReduce Framework

It's easy to execute MapReduce applications on Apache Hadoop.  Other than simplicity, scalability, and performance, Hadoop MapReduce framework also provides additional benefits such as:

• Failure and recovery
• Minimal data motion

Failure & Recovery

The framework takes care of failures. It is designed to detect and handle failures at the application layer, so delivering a highly-available service on top of a cluster of servers, each of which may be prone to failures.

If a server with one copy of the data is unavailable, another server has a copy of the same key/value pair, which can be used to solve the same sub-task. The JobTracker (see below) keeps track of it all.

Minimal data motion

Typically the compute nodes and the storage nodes are the same, that is, the MapReduce framework and the Hadoop Distributed File System are running on the same set of nodes (see the above diagram). This configuration allows the framework to effectively schedule tasks on the nodes where data is already present, resulting in lower network I/O and very high aggregate bandwidth across the cluster.

Job Tracker & Task Tracker

JobTracker

Applications using MapReduce framework are required to

• Specify Job configuration
• Specify input/output locations
• Supply map, combine and reduce functions

The Hadoop job client then submits the job (jar/executable etc.) and configuration to the JobTracker. In MR1, JobTracker manages and monitors both the resources and the MapReduce Job and task scheduling, which makes the JobTracker a single point of failure (SPOF) in a cluster.  Also, the cluster cannot be scaled efficiently.

TaskTracker 

The JobTracker will first determine the number of splits from the input path and select some TaskTracker based on their network proximity to the data sources, then the JobTracker send the task requests to those selected TaskTrackers.

The TaskTracker spawns a separate JVM processes to do the actual work; this is to ensure that process failure does not take down the task tracker. The TaskTracker monitors these spawned processes, capturing the output and exit codes. When the process finishes, successfully or not, the tracker notifies the JobTracker.

The TaskTrackers also send out heartbeat messages to the JobTracker, usually every few minutes, to reassure the JobTracker that it is still alive. These message also inform the JobTracker of the number of available slots, so the JobTracker can stay up to date with where in the cluster work can be delegated.

YARN一MR2

To overcome the drawbacks of MR1, YARN (or MR2) was introduced and provided for:^[9]

• Better scalability 
• As the JobTracker functionality is split between the ResourceManager and ApplicationMaster.  Hadoop may be run on much larger clusters on MR2.
• Better cluster utilization 
• As the resource capacity configured for each node may be used by both Map and Reduce tasks
• Non-MapReduce clusters 
• May be run on the same cluster concurrently
• Higher throughput 
• Uses finer-grained resource scheduling

References

1. Volume Rendering using MapReduce
2. MapReduce Tutorial (Apache Hadoop)
3. Hadoop MapReduce Framework (Hortonworks)
4. How Hadoop Map/Reduce works
5. Mapper 
• Maps input key/value pairs to a set of intermediate key/value pairs
6. Object-oriented framework presentation (CSCI 5448 Casey McTaggart)
7. YARN Architecture Guide
• The MapReduce framework consists of a single master ResourceManager, one slave NodeManager per cluster-node, and MRAppMaster per application (see tutorial).
• The ResourceManager has two main components: Scheduler and ApplicationsManager.
• Two scheduler plug-ins
• CapacityScheduler and the FairScheduler
8. Mastering Apache Spark 2
9. Practical Hadoop Ecosystem: A Definitive Guide to Hadoop-Related Frameworks
10. MapReduce Tutorial (Apache Hadoop)
• User Interfaces
11. All Cloud-related articles on Xml and More