Apache Spark is a cluster computing platform designed to be fast, expresive, high level, general-purpose, fault-tolerante and compatible with Hadoop (Spark can work directly with HDFS, S3 and so on). Spark can also be defined as a framework for distributed processing and analisys of big amounts of data. People from databricks (the company behind Spark) called it a distributed executing engine for large scale analytics.

Spark improves efficiency over Hadoop because it uses in-memory computing primitives. According to the Apache Spark site, it can run programs up to 100x faster than Hadoop MapReduce in memory, or 10x faster on disk.

It also claims to improve usability through rich Scala, Python and Java APIs as well as an interactive shell, in Scala and Python. Spark is written in Scala.

Spark Architecture

Spark has three main components

Apache Spark stack

Spark Core (API)

A high level programming framework that allows programmers to focus on the logic and not the plumbing of distributing programming, that is, the steps to be done without worrying of coordinating tasks, networking of so on.

These steps are define by RDD (Resilient Distributed Datasets), the main programming abstraction that represent a collection of items distributed across many compute nodes that can be manipulated in parallel.

Spark clustering

Spark itself doesn’t manage the cluster, but it supports three cluster managers:

• Standalone: a simple cluster manager included in Spark itself called the Standalone Scheduler.
• Hadoop YARN: see my introduction to Apache Hadoop.
• Apache Mesos.

Spark stack

Finally, Spark provides high level specialized components that are closely integrated in order to provide one great platform.

The current components are:

• Spark SQL: for querying data via SQL.
• Spark Streaming: for real-time processing of live streams of data.
• GraphX: a library for manipulating graphs and performing graph-parallel computations.
• MLLib: a library for machine learning providing algorithms for doing so (classification, regression, …)

Spark Usage

There are two ways to work with Spark:

• The Spark interactive shells
• Spark standalone applications

Spark Shell

It’s an interactive shell from the command line that has two implementations, one in Python and the other in Scala, an RPEL that is very useful for learning the API or for data exploration.

Spark’s shells allow you to interact with data not only on your single machine, but on disk or in memory across many machines, thanks to the distributed nature of Spark.

Spark Applications

The other way to work with Spark is by creating standalone applications either in Python, Scala or Java. Use them for large scale data processing.

Spark main concepts

Driver program

It’s the program that launches the distributed operations on a cluster.

The Spark shell is a driver program.

The application that you write, with its main function that defines de datasets and applies operations on them is a driver program.

Spark Context (sc)

It’s the main entry point to the Spark API.

When using the shell, a preconfigured SparkContext is automatically created and it’s available in the variable called sc.

When writing applications, the first thing that you need to create is your own instance of the SparkContext.

Resilient Distributed Dataset (RDD)

The goal of Spark is to allow you to operate in datasets in a single machine and that these operations work in the same way in a distributed cluster.

For achieving this, Spark offers the Resilient Distributed Dataset (RDD), they are immutable collections (dataset) of objects that Spark distributes (distributed) through the cluster. They are loaded from a source of data and, since they are immutable, RDDs are also created as a result of transformation on existing RDDs (map, filters, etc.). Finally, Spark automatically rebuilds them in a node if there is a failure in another node (resilient)

There are two types of RDD operations on RDDs:

• Transformations: lazy operations to build RDDs based on the current RDD.
• Actions: return a result or write the RDD to storage. It implies a computation that actually applies the pending transformation that were lazily defined.

In the Spark jargon, this is called a Direct Acyclic Graph (DAG) of operations. The RDDs track the series of transformations used to build them by maintaining a pointer to its parents.

Spark Installation

Go to https://spark.apache.org/downloads.html and then:

1. Choose a Spark release (1.2.1 is the last at the time of this writing)
2. Choose a package type: select the package type of “Pre-built for Hadoop 2.4 and later”
3. Choose a download type: Direct Download is OK, but the default Apache Mirror works well.
4. Click on the link after Download Spark, for instance spark-1.2.1.tgz, to download Spark.

Unpack the downloaded file and move into that directory in order to use the interactive shell:

$ tar -xf spark-1.2.0-bin-hadoop2.4.tgz
$ cd spark-1.2.0-bin-hadoop2.4

Using the Shell

The Python version of the Spark shell is available via the command bin/pyspark and the Scala version of the shell by using bin/spark-shell.

Note: the shell accept code completion with the Tab key.

Let’s try the Scala shell:

$ bin/spark-shell
Spark assembly has been built with Hive, including Datanucleus jars on classpath
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
15/02/26 17:23:45 INFO SecurityManager: Changing view acls to: Javier
15/02/26 17:23:45 INFO SecurityManager: Changing modify acls to: Javier
15/02/26 17:23:45 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(Javier); users with modify permissions: Set(Javier)
15/02/26 17:23:45 INFO HttpServer: Starting HTTP Server
15/02/26 17:23:45 INFO Utils: Successfully started service 'HTTP class server' on port 46130.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.2.1
      /_/

Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_31)
Type in expressions to have them evaluated.
Type :help for more information.
15/02/26 17:23:50 WARN Utils: Your hostname, xxx resolves to a loopback address: 127.0.1.1; using 192.168.2.49 instead (on interface eth0)
15/02/26 17:23:50 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
15/02/26 17:23:50 INFO SecurityManager: Changing view acls to: Javier
15/02/26 17:23:50 INFO SecurityManager: Changing modify acls to: Javier
15/02/26 17:23:50 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(Javier); users with modify permissions: Set(Javier)
15/02/26 17:23:51 INFO Slf4jLogger: Slf4jLogger started
15/02/26 17:23:51 INFO Remoting: Starting remoting
15/02/26 17:23:51 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@xxx.malsolo.lan:55248]
15/02/26 17:23:51 INFO Utils: Successfully started service 'sparkDriver' on port 55248.
15/02/26 17:23:51 INFO SparkEnv: Registering MapOutputTracker
15/02/26 17:23:51 INFO SparkEnv: Registering BlockManagerMaster
15/02/26 17:23:52 INFO DiskBlockManager: Created local directory at /tmp/spark-1420fe71-6907-408a-b44c-9547ba1a2c49/spark-909fad01-a3df-484b-bd30-1ea6006396e9
15/02/26 17:23:52 INFO MemoryStore: MemoryStore started with capacity 265.1 MB
15/02/26 17:23:52 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/02/26 17:23:52 INFO HttpFileServer: HTTP File server directory is /tmp/spark-82586699-a230-47e4-8148-2cc4dcc741ec/spark-72f09be4-797a-4612-a845-e4fd1e578e76
15/02/26 17:23:52 INFO HttpServer: Starting HTTP Server
15/02/26 17:23:52 INFO Utils: Successfully started service 'HTTP file server' on port 41493.
15/02/26 17:23:53 INFO Utils: Successfully started service 'SparkUI' on port 4040.
15/02/26 17:23:53 INFO SparkUI: Started SparkUI at http://xxx.malsolo.lan:4040
15/02/26 17:23:53 INFO Executor: Starting executor ID  on host localhost
15/02/26 17:23:53 INFO Executor: Using REPL class URI: http://192.168.2.49:46130
15/02/26 17:23:53 INFO AkkaUtils: Connecting to HeartbeatReceiver: akka.tcp://sparkDriver@xxx.malsolo.lan:55248/user/HeartbeatReceiver
15/02/26 17:23:53 INFO NettyBlockTransferService: Server created on 40938
15/02/26 17:23:53 INFO BlockManagerMaster: Trying to register BlockManager
15/02/26 17:23:53 INFO BlockManagerMasterActor: Registering block manager localhost:40938 with 265.1 MB RAM, BlockManagerId(, localhost, 40938)
15/02/26 17:23:53 INFO BlockManagerMaster: Registered BlockManager
15/02/26 17:23:53 INFO SparkILoop: Created spark context..
Spark context available as sc.

scala>

To exit either shell, press Ctrl-D.

scala> Stopping spark context.
15/02/26 17:27:40 INFO SparkUI: Stopped Spark web UI at http://xxx.malsolo.lan:4040
15/02/26 17:27:40 INFO DAGScheduler: Stopping DAGScheduler
15/02/26 17:27:41 INFO MapOutputTrackerMasterActor: MapOutputTrackerActor stopped!
15/02/26 17:27:41 INFO MemoryStore: MemoryStore cleared
15/02/26 17:27:41 INFO BlockManager: BlockManager stopped
15/02/26 17:27:41 INFO BlockManagerMaster: BlockManagerMaster stopped
15/02/26 17:27:41 INFO SparkContext: Successfully stopped SparkContext
15/02/26 17:27:41 INFO RemoteActorRefProvider$RemotingTerminator: Shutting down remote daemon.
15/02/26 17:27:41 INFO RemoteActorRefProvider$RemotingTerminator: Remote daemon shut down; proceeding with flushing remote transports.
15/02/26 17:27:41 INFO RemoteActorRefProvider$RemotingTerminator: Remoting shut down.
$

It’s possible to control the verbosity of the logging by creating a conf/log4j.properties file (use the existing conf/log4j.properties.template, Currently, Spark uses log4j 1.2.17, so you can find more details at Apache log4j™ 1.2 website.) and then changing the line:

log4j.rootCategory=INFO, console

To:

log4j.rootCategory=WARN, console

Now, with the shell we can try some commands like examining the sc variable, create RDDs, filtering them and so on.

scala> sc
res0: org.apache.spark.SparkContext = org.apache.spark.SparkContext@76af34b5

scala> val lines = sc.textFile("README.md")
lines: org.apache.spark.rdd.RDD[String] = README.md MappedRDD[1] at textFile at :12

scala> lines.count()
res1: Long = 98                                                                 

scala> lines.first()
res2: String = # Apache Spark

There is an INFO message that informs of the URL of the Spark UI (INFO SparkUI: Started SparkUI at http://[ipaddress]:4040), so you can use it to see information about the tasks and clusters.

Spark UI

Spark Operations

Once we have the Spark shell, let’s use it to take a look to the available operations before we dive into creating applications.

Creating RDDs

You can turn an existing collection into a RDD (parallelize it), you can load an external file (several formats: text, JSON, CSV, SequenceFiles, objects) or even existing Hadoop InputFormat (with sc.hadoopFile())

scala> val numbers = sc.parallelize(List(1,2,3))
numbers: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :12

scala> val lines = sc.textFile("README.md")
lines: org.apache.spark.rdd.RDD[String] = README.md MappedRDD[2] at textFile at :12

scala>

Transformations

As we said earlier, transformations are lazy evaluated operations on RDDs that return a new RDD.

You can pass each element through a function (with map()) or keep elements that pass a predicate (with filter()) or produce zero or more elements for each element (with flatMap()) and so on.

scala> val squares = numbers.map(x => x*x)
squares: org.apache.spark.rdd.RDD[Int] = MappedRDD[3] at map at :14

scala> val spark = lines.filter(line => line.contains("Spark"))
spark: org.apache.spark.rdd.RDD[String] = FilteredRDD[4] at filter at :14

scala> val sequences = numbers.flatMap(x => 1 to x)
sequences: org.apache.spark.rdd.RDD[Int] = FlatMappedRDD[5] at flatMap at :14

scala> val words = lines.flatMap(line => line.split(" "))
words: org.apache.spark.rdd.RDD[String] = FlatMappedRDD[6] at flatMap at :14

scala>

Actions

They are the operations that return a final value to the driver program or write data to an external storage system that result in the evaluation of the transformations in the RDD.

For instance, retrieve the contents (collect()), return the first n elements (take()), count the number of elements (count()), combine elements with an associative function (reduce()) or write elements to a text file (saveAsTextFile())

scala> sequences.collect()
res0: Array[Int] = Array(1, 1, 2, 1, 2, 3)                                      

scala> squares.take(2)
res1: Array[Int] = Array(1, 4)

scala> words.count()
res2: Long = 524

scala> numbers.reduce((x, y) => x + y)
res4: Int = 6

scala> spark.saveAsTextFile("borrar.txt")

scala>

Key/Value Pairs

There is a special type of RDD, Pair RDDs, that that contain elements that are tuples, that is, a key-value pair, being key and value of any type.

They are very useful for perform aggregations, grouping, counting. They can be obtained from some initial ETL (extract, transform, load) operations.

The pair RDDS can be partitioned across nodes for improving speed by allowing similar keys to be accesible on the same node.

Regarding operations, Spark offers special operation for Pair RDDs that allow you to act on each key in parallel, for instance, reduceByKey() to aggregate data by key, join() to merge two RDDs by grouping elements with the same key, or even sortByKey().

scala> val pets = sc.parallelize(List(("cat", 1), ("dog", 1), ("cat", 2)))
pets: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[8] at parallelize at :12

scala> pets.collect()
res12: Array[(String, Int)] = Array((cat,1), (dog,1), (cat,2))

scala> pets.reduceByKey((a, b) => a + b).collect()
res9: Array[(String, Int)] = Array((dog,1), (cat,3))

scala> pets.groupByKey().collect()
res10: Array[(String, Iterable[Int])] = Array((dog,CompactBuffer(1)), (cat,CompactBuffer(1, 2)))

scala> pets.sortByKey().collect()
res11: Array[(String, Int)] = Array((cat,1), (cat,2), (dog,1))

scala>

Now let’s use the Shell to see how easily you can implement the MapReduce WordCount example in a single line:

scala> val counts = lines.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((x, y) => x + y)
counts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[18] at reduceByKey at :14

scala> counts.collect()
res16: Array[(String, Int)] = Array((package,1), (this,1), (Because,1), (Python,2), (cluster.,1), (its,1), ([run,1), (general,2), (YARN,,1), (have,1), (pre-built,1), (locally.,1), (changed,1), (locally,2), (sc.parallelize(1,1), (only,1), (several,1), (This,2), (basic,1), (first,1), (documentation,3), (Configuration,1), (learning,,1), (graph,1), (Hive,2), (["Specifying,1), ("yarn-client",1), (page](http://spark.apache.org/documentation.html),1), ([params]`.,1), (application,1), ([project,2), (prefer,1), (SparkPi,2), (,1), (engine,1), (version,1), (file,1), (documentation,,1), (MASTER,1), (example,3), (are,1), (systems.,1), (params,1), (scala>,1), (provides,1), (refer,2), (configure,1), (Interactive,2), (distribution.,1), (can,6), (build,3), (when,1), (Apache,1),...
scala>

Spark Applications

For writing a Spark Application it’s possible to use Scala, Python or Java. What I’m going to do is to use Java 8 to take advantage of the new features of the language in order to have a less verbose syntax.

Word count Java application

First, use the appropriate dependency. For instance, with maven:

<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-core_2.10</artifactId>
			<version>1.2.1</version>
		</dependency>

Then, you have to instantiate your own SparkContext, and it’s done via a SparkConf object. We use the minimal configuration: a name for the cluster URL (“local” to use a local cluster) and an application name to identify the application on the cluster:

SparkConf conf = new SparkConf().setMaster("local").setAppName("Word Count with Spark");
		JavaSparkContext sc = new JavaSparkContext(conf);

Now, before writing Java, code it’s necessary to explain the differences with Scala.

Spark is written in Scala and it takes full advantage of its features. But Java lacks of some of them. So Spark provides alternatives with interfaces or concrete classes.

Let’s see the Word Count example in Spark written in Scala:

val file = spark.textFile("file")
val counts = file.flatMap(line => line.split(" "))
                 .map(word => (word, 1))
                 .reduceByKey(_ + _)
counts.saveAsTextFile("out")

Java didn’t accept functions as parameters, so Spark provides interfaces in the org.apache.spark.api.java.function package to be implemented, either as anonymous inner classes or as named classes, to be passed as arguments of the functions (flatMap(), map(), reduceByKey(), …)

In our case, these are the functions that are needed:

• FlatMapFunction with the method Iterable call(T t) to return zero or more output records from each input record (t).
• PairFunction with the method Tuple2 call(T t) to return key-value pairs (Tuple2), and can be used to construct PairRDDs.
• Function2 with the method R call(T1 v1, T2 v2), a two-argument function that takes arguments of type T1 and T2 and returns an R.

Java doesn’t have a native implementation of Tuple (as Lukas Eder said On a side-note at here “Why the JDK doesn’t ship with built-in tuples like C#’s or Scala’s escapes me.”, in other words, “Functional programming without tuples is like coffee without sugar: A bitter punch in your face.”)

For that reason, Spark provides several implementations for Tuple in the scala package.

But, Java has evolved, and now functions are first class citizens, so it’s possible to pass them as parameters for other functions, thus it’s very easy to write the Java 8 version of the word count in Spark using lambdas (since the provided interfaces have a sole public method. And the result is almost as clear as the Scala version)

JavaRDD<String> lines = sc.textFile("file");
		JavaPairRDD<String, Integer> counts = lines.flatMap(line -> Arrays.asList(line.split(" ")))
			.mapToPair(word -> new Tuple2<String, Integer>(word, 1))
			.reduceByKey((x, y) -> x + y);
		counts.saveAsTextFile("out");

The complete source code is available at GitHub.

Build and run

Now, we only have to build the project (with maven) and submit it to Spark (with bin/spark-submit). From the root directory of the application (note: the out directory must not exists, so remove it previously if you need so with rm -r out):

$ mvn clean install
$ ~/Applications/spark-1.2.1-bin-hadoop2.4/bin/spark-submit --class com.malsolo.spark.examples.WordCount target/spark-examples-0.0.1-SNAPSHOT.jar

Finally, we can see the results to compare with the ones obtained using Hadoop:

$ cat out/part-00000 | grep President
(President,,26)
(President,72)
(President.,8)
(Vice-President,5)
(Vice-President,,5)
(Vice-President;,1)
(President;,3)
(Vice-President.,1)

$ cat out/part-00000 | grep United
(United,85)

$ cat out/part-00000 | grep State
(State,47)
(States,46)
(States.",1)
(State.,6)
(States,,55)
(State,,20)
(States:,2)
(States.,8)
(Statement,1)
(States;,13)
(State;,4)

Shared Variables

Spark closures and the variables they use are sent separately to the tasks running on the cluster, thus the variables created in the driver program are recieved in the tasks as a new copy, so updates on these copies are not propagated back to the driver.

Spark has two kinds of shared variables, accumulators and broadcast variables, to solve that problem as well as for solving issues related with the amount of data that is sent across the cluster.

Accumulators

Variables that can be used to aggregate values from worker nodes back to the driver program. In a nutshell:

• They are created with SparkContext.accumulator(initialValue) that returns an org.apache.spark.Accumulator[T] (with T, the type of initialValue)
• Worker code adds values with += in Scala or the function add() in Java.
• The driver program can access with value in Scala or value()/setValue() in Java (accessing from worker code throws an exception)
• The right value will be obtained after calling an action (remember that transformations are lazy operations)

Spark has built-in support for accumulators of type Integer, but you can create custom Accumulators by extending AccumulatorParam.

Let’s see an example that counts the empty lines in the file that we use to count words:

public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("Word Count with Spark");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		JavaRDD<String> lines = sc.textFile(INPUT_FILE_TEXT);
		
		final Accumulator<Integer> blankLines = sc.accumulator(0);
		
		@SuppressWarnings("resource")
		JavaPairRDD<String, Integer> counts = lines.flatMap(line -> 
			{
				if ("".equals(line)) {
					blankLines.add(1);
				}
				return Arrays.asList(line.split(" "));
			})
			.mapToPair(word -> new Tuple2<String, Integer>(word, 1))
			.reduceByKey((x, y) -> x + y);

		counts.saveAsTextFile(OUTPUT_FILE_TEXT);
		
		System.out.println("Blank lines: " + blankLines.value());
		
		sc.close();
	}

• In line 5 we create an Accumulator initialized to 0
• In lines 11 to 16 we modify the FlatMapFunction to add 1 if the input line is empty
• In line 22 we print the value of the content. After the saveAsTextFile() action.

Let’s try:

$ rm -r out
$ mvn clean install
$ ~/Applications/spark-1.2.1-bin-hadoop2.4/bin/spark-submit --class com.malsolo.spark.examples.WordCount target/spark-examples-0.0.1-SNAPSHOT.jar

Spark assembly has been built with Hive, including Datanucleus jars on classpath
15/03/02 15:26:22 WARN Utils: Your hostname, xxx resolves to a loopback address: 127.0.1.1; using yyy.yyy.y.yy instead (on interface eth0)
15/03/02 15:26:22 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
15/03/02 15:26:23 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Blank lines: 169
$

Broadcast variables

Shared variable to efficiently distribute large read-only values to all the worker nodes.

If you need to use the same same variable in multiple parallel operations, it’s likely you’d rather share it instead of letting Spark sends it separately for each operation.

In a nutshell:

• They are created with SparkContext.broadcast(initValue) on an object of type T that has to be Serializable.
• Access its value with value in Scala or value() in Java.
• The value shouldn’t be modified after creation, because the change will only happen in one node.

Let’s see an example with a list of words that have not to be included in the count (a short list, but enough to see the concept):

public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("Word Count with Spark");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		JavaRDD<String> lines = sc.textFile(INPUT_FILE_TEXT);
		
		final Accumulator<Integer> blankLines = sc.accumulator(0);
		
		final Broadcast<List<String>> wordsToIgnore = sc.broadcast(getWordsToIgnore());
		
		@SuppressWarnings("resource")
		JavaPairRDD<String, Integer> counts = lines.flatMap(line -> 
			{
				if ("".equals(line)) {
					blankLines.add(1);
				}
				return Arrays.asList(line.split(" "));
			})
			.filter(word -> !wordsToIgnore.value().contains(word))
			.mapToPair(word -> new Tuple2<String, Integer>(word, 1))
			.reduceByKey((x, y) -> x + y);
		
		counts.saveAsTextFile(OUTPUT_FILE_TEXT);
		
		System.out.println("Blank lines: " + blankLines.value());
		
		sc.close();
	}

	private static List<String> getWordsToIgnore() {
		return Arrays.asList("the", "of", "and", "for");
	}

• In line 9 we create the broadcast variable: a list of words to ignore. In lines 30 to 31 we only return 4 words, but it’s easy to see that the list could be big enough.
• In line 19 we access the broadcast variable with the value() method and use it in a filter method.

Resources

• Source code at GitHub
• Learning Spark. By Holden Karau, Andy Konwinski, Patrick Wendell, Matei Zaharia (O’Reilly Media)
• Cloudera Developer Training for Apache Spark. By Diana Carroll (Cloudera training)
• Parallel Programming with Spark (Part 1 & 2). By Matei Zaharia ((UC Berkeley AMPLab YouTube channel))
• Advanced Spark Features. By Matei Zaharia (UC Berkeley AMPLab YouTube channel)
• A Deeper Understanding of Spark Internals. By Aaron Davidson (Apache Spark YouTube channel)

Hadoop is an open source framework for distributed fault-tolerant data storage and batch processing. It allows you to write applications for processing really huge data sets across clusters of computers using simple programming model with linear scalability on commodity hardware. Commodity hardware means cheaper hardware than the dedicated servers that are sold by many vendors. Linear scalability means that you have only to add more machines (nodes) to the Hadoop cluster.

The key concept for Hadoop is move-code-to-data, that is, data is distributed across the nodes of the Hadoop cluster and the applications (the jar files) are later sent to that nodes instead of vice versa (as in Java EE where applications are centralized in a application server and the data is collected to it over the network) in order to process the data locally.

At its core, Hadoop has two parts:

· Hadoop Distributed File System (HDFS™): a distributed file system that provides high-throughput access to application data.
· YARN (Yet Another Resource Negotiator): a framework for job scheduling and cluster resource management.

As you can see in the very definition of the Apache Hadoop website (what is Apache Hadoop?), Hadoop offers as a third component Hadoop MapReduce, a batch-based, distributed computing framework modeled after Google’s paper on MapReduce. It allows you to parallelize work over a large amount of raw data by splitting the input dataset into independent chunks which are processed by the map tasks (initial ingestion and transformation) in parallel, whose outputs are sorted and then passed to the reduce tasks (aggregation or summarization).

In the previous version of Hadoop (Hadoop 1), the implementation of MapReduce was based on a master JobTracker, for resource management and job scheduling/monitoring, and per-node slaves called TaskTracker to launch/teardown tasks. But it had scalability problems, specially when you wanted very large clusters (more than 4,000 nodes).

So, MapReduce has undergone a complete overhaul and now is called MapReduce 2.0 (MRv2), but it is not a part by itself, currently, MapReduce is a YARN-based system. That’s the reason why we can say that Hadoop has two main parts: HDFS and YARN.

Hadoop ecosystem

Besides the two core technologies, the distributed file system (HDFS) and Map Reduce (MR), there are a lot of projects that expand Hadoop with additional useful technologies, in such a way that we can consider all of them an ecosystem around Hadoop.

Next, a list of some of these projects, organized by some kind of categories:

• Data Ingestion: to move data from and into HDFS
  • Flume: a system for moving data into HDFS from remote systems using configurable memory-resident daemons that watch for data on those systems and then forward the data to Hadoop. For example, weblogs from multiple servers to HDFS.
  • Sqoop: a tool for efficient bulk transfer of data between structured data stores (such as relational databases) and HDFS.
• Data Processing:
  • Pig: a procedural language for querying and data transform with scripts in a data flow language call PigLatin.
  • Hive: a declarative SQL-like kanguage.
  • Spark: an in-memory distributed data processing that breaks problems up over all of the Hadoop nodes, but keeps the data in memory for better performance and can be rebuilt with the details stored in the Resilient Distributed Dataset (RDD) from an external store (usually HDFS).
  • Storm: a distributed real-time computation system for processing fast, large streams of data.
• Data Formats:
  • Avro: a language-neutral data serialization system. Expressed as JSON.
  • Parquet: a compressed columnar storage format that can efficiently store nested data
• Storage:
  • HBase: a scalable, distributed database that supports structured data storage for large tables.
  • Accumulo: a scalable, distributed database that supports structured data storage for large tables.
• Coordination:
  • Zookeeper: a high-performance coordination service for distributed applications.
• Machine Learning:
  • Mahout: a scalable machine learning and data mining library: classification, clustering, pattern mining, collaborative filtering and so on.
• Workflow Management:
  • Oozie: a service for running and scheduling workflows of Hadoop jobs (including Map-Reduce, Pig, Hive, and Sqoop jobs).

Hadoop installation

To install Hadoop on a single machine to try it out, just download the compressed file for the desired version and unpack it on the filesystem.

Prerequisites

There is some required software for running Apache Hadoop:

• Java. It’s also necessary to inform Hadoop where Java is via the environment variable JAVA_HOME

$ java -version
java version "1.8.0_31"
Java(TM) SE Runtime Environment (build 1.8.0_31-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.31-b07, mixed mode)
$ echo $JAVA_HOME
/usr/lib/jvm/java-8-oracle

• ssh: I have Ubuntu 14.04 that comes with ssh, but I had to manually install a server.

$ which ssh
/usr/bin/ssh
$ which sshd
/usr/sbin/sshd

• On Mac OSX, make sure Remote Login (under System Preferences -> File Sharing) is enabled for the current user or for all users.
• On Windows, the best option is to follow the instructions at the Wiki: Build and Install Hadoop 2.x or newer on Windows.

Download and install

To get a Hadoop distribution, download a recent stable release from one of the Apache Download Mirrors.

There are several directories, for the current, last stable, last v1 stable version and so on. Basically, you’ll download a tar gzipped file named hadoop-x.y.z.tar.gz, for instance: hadoop-2.6.0.tar.gz.

You can unpack it wherever you want and then point the PATH to that directory. For example:

$ tar xzf hadoop-2.6.0.tar.gz
$
$ export HADOOP_HOME=~/Applications/hadoop-2.6.0
$ export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

Now you can verify the installation by typing hadoop version:

$ hadoop version
Hadoop 2.6.0
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r e3496499ecb8d220fba99dc5ed4c99c8f9e33bb1
Compiled by jenkins on 2014-11-13T21:10Z
Compiled with protoc 2.5.0
From source with checksum 18e43357c8f927c0695f1e9522859d6a
This command was run using ~/Applications/hadoop-2.6.0/share/hadoop/common/hadoop-common-2.6.0.jar
$

Configuration

Hadoop has three supported modes:

• Local (Standalone) Mode: a single Java process with daemons running. For development testing and debugging.
• Pseudo-Distributed Mode: each Hadoop daemon runs in a separate Java process. For simulating a cluster on a small scale.
• Fully-Distributed Mode: the Hadoop daemons run on a cluster of machines. If you want to take a look, see the oficial documentation: Hadoop MapReduce Next Generation – Cluster Setup.

In standalone mode, there is no further action to take, the default properties are enough and there are no daemons to run.

In pseudodistributed mode, you have to set up your computer as described at Hadoop MapReduce Next Generation – Setting up a Single Node Cluster. But let’s review the steps needed.

You need at least a minimum configuration with four files in HADOOP_HOME/etc/hadoop/:

• core-site.xml. Common configuration, default values at Configuration: core-default.xml

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
	<property>
		<name>fs.defaultFS</name>
		<value>hdfs://localhost:8020</value>
	</property>
</configuration>

fs.defaultFS replaces the deprecated fs.default.name whose default value is file://

• hdfs-site.xml. HDFS configuration, default values at Configuration: hdfs-default.xml

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
	<property>
		<name>dfs.replication</name>
		<value>1</value>
	</property>
</configuration>

dfs.replicationis the default block replication, unless other is specified at creation time. The default value is 3, but we use 1 because we have only one node.

Other useful values are:

dfs.namenode.name.dir, local path for storing the fsimage by the NN (defaults to file://${hadoop.tmp.dir}/dfs/name with hadoop.tmp.dir configurable at core-site.xml with default value /tmp/hadoop-${user.name})

dfs.datanode.data.dir, local path for storing blocks by the DN (defaults to file://${hadoop.tmp.dir}/dfs/data)

• mapred-site.xml. MapReduce configuration, default values at Configuration: mapred-default.xml

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
	<property>
		<name>mapreduce.framework.name</name>
		<value>yarn</value>
	</property>
</configuration>

mapreduce.framework.name, the runtime framework for executing MapReduce jobs: local, classic or yarn.

Other useful values are:

mapreduce.jobtracker.system.dir, the directory where MapReduce stores control files (defaults to ${hadoop.tmp.dir}/mapred/system).

mapreduce.cluster.local.dir, the local directory where MapReduce stores intermediate data files (defaults to ${hadoop.tmp.dir}/mapred/local)

• yarn-site.xml. YARN configuration, default values at Configuration: yarn-default.xml

<?xml version="1.0"?>
<configuration>
	<property>
		<name>yarn.resourcemanager.hostname</name>
		<value>localhost</value>
	</property>
	<property>
		<name>yarn.nodemanager.aux-services</name>
		<value>mapreduce_shuffle</value>
	</property>
</configuration>

yarn.resourcemanager.hostname, the host name of the Resource Manager.

yarn.nodemanager.aux-services, list of auxiliary services executed by the Node Manager. The value mapreduce_shuffle is for the Suffle/Sort in MapReduce that is an auxilary service in Hadoop 2.x.

Configuring SSH

Pseudodistributed mode is like fully distributed mode with a single host: localhost. In order to start the daemons on the set of hosts in the cluster, SSH is used. So we’ll configure SSH to log in without password.

Remember that you need to have SSH installed and a server running. On Ubuntu, try this if you need so:

$ sudo apt-get install ssh

Now create a SSH key with an empty passprhase

$ ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa_hadoop
$ cat ~/.ssh/id_rsa_hadoop.pub >> ~/.ssh/authorized_keys

Finally, test if you can connect without a pasword by trying:

$ ssh localhost

First steps with HDFS

Before using HDFS for the first time, some steps must be performed:

Formatting the HDFS filesystem

Just run the following command:

$ hdfs namenode -format

Starting the daemons

To start the HDFS, YARN, and MapReduce daemons, type:

$ start-dfs.sh
$ start-yarn.sh
$ mr-jobhistory-daemon.sh start historyserver

You can check what processes are running with the Java’s jps command:

$ jps
25648 NodeManager
25521 ResourceManager
25988 JobHistoryServer
25355 SecondaryNameNode
25180 DataNode
26025 Jps
$

Stopping the daemons

Once it’s over, you can stop the daemons with:

$ mr-jobhistory-daemon.sh stop historyserver
$ stop-yarn.sh
$ stop-dfs.sh

Creating A User Directory

You can create a home directory for a user with the next command:

$ hadoop fs -mkdir -p ~/Documents/hadoop-home/

Other Hadoop installations

There are another way to get installed Hadoop, that is, using companies that provide products that include Apache Hadoop or some kind of derivatives:

• Amazon Elastic MapReduce (Amazon EMR)
• Cloudera’s Distribution including Apache Hadoop (CDH)
• Hortonworks Data Platform Powered by Apache Hadoop (HDP)
• MapR Technologies
• Pivotal HD

Hadoop Distributed File System (HDFS)

The HDFS filesystem designed for distributed storage of very large files (hundreds of megabytes, gigabytes, or terabytes in size) and distributed processing using commodity hardware. It is a hierarchical UNIX-like file system, but internally it splits large files into blocks (with size from 32MB to 128MB, being 64MB the default), in order to perform a distribution and a replication of these blocks among the nodes of the Hadoop cluster.The applications that use HDFS usually write data once and read data many times.

The HDFS has two types of nodes:

• The master NameNode (NN), that stores the filesystem tree and the metadata for locating the files and directories in the tree that are actually located in the DataNodes. It stores this information in memory, however, to ensure against data loss, it’s also saved to disk using two files: the namespace image and the edit log.
  • fsimage: a point in time snapshot of what HDFS looks like.
  • edit log: the deltas or changes to HDFS since the last snapshot.

  Both are prediodically merged.

• The DataNodes (DN), that are responsible for serving the actual file data (once the client knows which one to use after contacting the NameNode). They also sends heatbeats every 3 seconds (by default) to the NN and block reports every 1 hour (by default) to the DN both for maintenance purposes.

There is also a node poorly named Secondary NameNode that is not a failover node nor a backup node, it periodically merges the namespace image with the edit log to prevent the edit log from becoming too large. Thus, the best name for it is Checkpoint Node.

The Command-Line Interface

Once you have installed Hadoop, you can interact with HDFS, as well as other file systems that Hadoop supports (local filesystem, HFTP FS, S3 FS, and others), using the command line. The FS shell is invoked by:

$ hadoop fs 

Provided to you have hadoop in the PATH as we saw above.

You can find a list of available commands at File System Shell.

You can perform operations like:

• copyFromLocal (putting files into HDFS)
• copyToLocal (getting files form HDFS)
• mkdir (creating directories in HDFS)
• ls (list files in HDFS)

Data exchange with HDFS

Hadoop is mainly written in Java, being the core class for HDFS the abstract class org.apache.hadoop.fs.FileSystem, that represents a filesystem in Hadoop. The several concrete subclasses provide implementations from local filesystem (fs.LocalFileSystem) to HDFS (hdfs.DistributedFileSystem), or Amazon S3 (fs.s3native.NativeS3FileSystem) and many more (read-only HTTP, FTP server, …)

Reading data

Reading data using the Java API involves to obtain the abstract FileSystem via one of the factory methods (get()), or the convenient method for retrieving the local filesystem (getLocal()):

public static FileSystem get(Configuration conf) throws IOException
public static FileSystem get(URI uri, Configuration conf) throws IOException
public static FileSystem get(URI uri, Configuration conf, String user)
throws IOException
public static LocalFileSystem getLocal(Configuration conf) throws IOException

And then obtain an input stream for a file (that can be later be closed):

public FSDataInputStream open(Path f) throws IOException
public abstract FSDataInputStream open(Path f, int bufferSize) throws IOException

With these methods on hand, the flow of the data in HDFS will be the next:

Reading from HDFS

1. The client calls the open() method to read a file. A DistributedFileSystem is returned.
2. The DistributedFileSystem asks the NameNode for the block locations. The NameNode returns an ordered list of the DataNodes that have a copy of the block (sorted by proximity to the client). The DistributedFileSystem returns a FSDataInputStream to the client for it to read data from
3. The client calls read() on the input stream
4. The FSDataInputStream reads data for the client from the DataNode until there is no more data in that node.
5. The FSDataInputStream will manage the closing and opening connection to DataNodes for serving data to the client in a transparently way. It also manages validation (checksum) and errors (by trying to read data from a replica)
6. When the client has finished reading, it calls close() on the FSDataInputStream

Writing data

The Java API allows you to create files with create methods (that, by the way, also create any parent directories of the file that don’t already exist). The API also includes a Progressable interface to be notified of the process of the data being written to the datanodes. It’s also possible to append data to an existing file, but this functionality is optional (S3 doesn’t support it from the time being)

public FSDataOutputStream create(Path f) throws IOException
public FSDataOutputStream append(Path f) throws IOException

The output stream will be used for writing the data. Furthermore, the FSDataOutputStream can inform of the current position in the file.

The flow of the data written to HDFS with these methods is the next:

Writing to HDFS

1. The client creates the file by calling create() on DistributedFileSystem.
2. DistributedFileSystem makes an RPC call to the namenode to create a
   new file in the filesystem’s namespace, with no blocks associated with it. The NameNode checks file existence and permission, throwing IOException if theres is any problem, otherwise, it returns a FSDataOutputStream for writing data to.
3. The data written by the client is splitted into packets that are sent to a data queue.
4. The data queue is consumed by the Data Streamer which streams the packets to a pipeline of DataNodes (one per replication factor). Each DataNode stores the packet and send it to the next DataNode in the pipeline.
5. There is another queue, ack queue, that contains the packets that are waiting for acknowledged by all the datanodes in the pipeline. If a DataNode fails in a write operation, the pipeline will be re-arranged transparently for the client.
6. When the client has finished writing data, it calls close() on the stream.
7. The remaining packets are flushed and, after receiving all the acknowledgments, the NameNode is notified that the write to the file is completed.

Apache YARN (Yet Another Resource Negotiator)

YARN is Hadoop’s cluster resource management system. It provides provides APIs for requesting and working with cluster resources to be used not by user code, but for higher level APIs, like MapReduce v2, Spark, Tez…

YARN separates resource management and job scheduling/monitoring into separate daemons. In Hadoop 1.x these two functions were performed by the JobScheduler, that implies a bottleneck for scaling the Hadoop nodes in the cluster.

YARN Components

There are five major component types in a YARN cluster:

• Resource Manager (RM): a global per-cluster daemon that is solely responsible for allocating and managing resources available within the cluster.
• Node Manager (NM): a per-node daemon that is responsible for creating, monitoring, and killing containers.
• Application Master (AM): This is a per-application daemon whose duty is the negotiation of resources from the ResourceManager and to work with the NodeManager(s) to execute and monitor the tasks.
• Container: This is an abstract representation of a resource set that is given to a particular application: memory and cpu. It’s a computational unit (one node runs several containers, but a container cannot cross a node boundary). The AM is a specialized container that is used to bootstrap and manage the entire application’s life cycle.
• Application Client: it submits applications to the RM and it specifies the type of AM needed to execute the application (for instance, MapReduce).

Anatomy of a YARN Request

These are the steps involved in the submission of a job to the YARN framework.

YARN architecture

1. The client submits a job to the RM asking to run an AM process (Job Submission in the picture above).
2. The RM looks for resources to acquire a container on a node to launch an instance of the AM.
3. The AM registers with the RM to enable the client to query the RM for details about the AM.
4. Now the AM is running, and it could run the computation returning the result to the client, or it could request more containers to the RM to run a distributed computation (Resource Request in the picture above)
5. The application code executing in the launched container (tasks) reports its status to the AM through an application-specific protocol (MapReduce status in the picture above, that it’s assuming that the YARN application being executed is MapReduce).
6. Once the application completes execution, the AM deregisters with the RM, and the containers used are released back to the system.

This process applies for each client that submits jobs. In the picture above there are two clients (the red one and the blue one)

Hadoop first program: WordCount MapReduce

MapReduce is a paradigm for data processing that uses two key phases:

1. Map: it performs a transformation on input key-value pairs to generate intermediate key-value pairs.
2. Reduce: it performs a summarize function on intermediate key-value groups to generate the final output of key-value pairs.
3. The groups that are the input of the Reduce phase are created by sorting the output of the Map phase in an operation called as Short/Shuffle (in YARN, is an auxiliary service)

Writing the program

For writing a MapReduce program in Java for running it in Hadoop you need to provide a Mapper class, a Reducer class, and a driver program to run a job.

Let’s begin with the Mapper class, it will separate each word with a count of 1:

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
	
	private final static IntWritable ONE = new IntWritable(1);
	private Text word = new Text();
	
	@Override
	protected void map(LongWritable key, Text value,
			Mapper<LongWritable, Text, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		StringTokenizer tokenizer = new StringTokenizer(line);
		while (tokenizer.hasMoreTokens()) {
			word.set(tokenizer.nextToken());
			context.write(word, ONE);
		}
	}

}

Highlights here are the parameters of the Mapper class, in this case:

1. The input key, a long that will be ignored
2. The input value, a line of text
3. The output key, the word to be counted
4. The output value, the count for the word, always one, as we said before.

As you can see, instead of using Java types, it’s better to use Hadoop basic types that are optimized for network serialization (available in the org.apache.hadoop.io package)

The basic approach is to override the map() method and make use of the key and value input parameters, as well as the instance of a Context to write the output to: the words with its count (one, for the moment being)

Let’s continue with the Reducer class.

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Reducer<Text, IntWritable, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		context.write(key, new IntWritable(sum));
	}
	
}

The intermediate result from the Mapper will be partitioned by MapReduce in such a way that the same reducer will receive all output records containing the same key. MapReduce will also sort all the map output keys and will call each reducer only once for each output key along with a list of all the output values for this key.

Thus, to write a Mapper class, you override the method reduce that has as parameters the only key, the list of values as an iterable and an instance of the Context to write the final result to.

In our case, the reducer will sum the count that each words carry (always one) and it will write the result to the context.

Finally, the Driver class, the class that runs the MapReduce job.

public class WordCountDriver {
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] myArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (myArgs.length != 2) {
			System.err.println("Usage: WordCountDriver <input path> <output path>");
			System.exit(-1);
		}
		Job job = Job.getInstance(conf, "Classic WordCount");
		job.setJarByClass(WordCountDriver.class);
		
		FileInputFormat.addInputPath(job, new Path(myArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(myArgs[1]));
		
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		
		//job.setMapOutputKeyClass(Text.class);
		//job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

First, create a Hadoop Configuration (the default values are enough for this example) and use the GenericOptionsParser class to parse only the generic Hadoop arguments.

To configure, submit and control the execution of the job, as well as to monitor its progress, use a Job object. Take care of configuring it (via its set() methods) before submitting the job or an IllegalStateException will be thrown.

In a Hadoop cluster, the JAR package will be distributed around the cluster, to allow Hadoop to locate this JAR we pass a class in the Job ’s setJarByClass() method.

Next, we specify the input and output paths by calling the static addInputPath() (or setInputPaths) method on FileInputFormat() (with a file, directory or file pattern) and the static setOutputPath() method on FileOutputFormat (with a non-existing directory, in order to avoid data loss from another job) respectively.

Then, the job is configured with the Mapper class and the Reducer class.

There is no need for specifying the map output types because they are the same than the ones produced by the Reducer class, but we need to indicate the output types for the reduce function.

Finally, the waitForCompletion() method on Job submits the job and waits for it to finish. The argument is a flag for verbosity in the generated output. The return value indicates success (true) or failure (false). We use it for the the program’s exit code (0 or 1).

Running the program

The source code is available at github. You can download it, go to the directory and just run the next commands:

$ mvn clean install
$ export HADOOP_CLASSPATH=target/mapreduce-0.0.1-SNAPSHOT.jar
$ hadoop com.malsolo.hadoop.mapreduce.WordCountDriver data/the_constitution_of_the_united_states.txt out

You will see something like this:

15/02/25 15:30:47 INFO Configuration.deprecation: session.id is deprecated. Instead, use dfs.metrics.session-id
15/02/25 15:30:47 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
15/02/25 15:30:47 INFO input.FileInputFormat: Total input paths to process : 1
15/02/25 15:30:47 INFO mapreduce.JobSubmitter: number of splits:1
15/02/25 15:30:47 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_local284822998_0001
15/02/25 15:30:47 INFO mapreduce.Job: The url to track the job: http://localhost:8080/
15/02/25 15:30:47 INFO mapreduce.Job: Running job: job_local284822998_0001
15/02/25 15:30:47 INFO mapred.LocalJobRunner: OutputCommitter set in config null
15/02/25 15:30:47 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
15/02/25 15:30:48 INFO mapred.LocalJobRunner: Waiting for map tasks
15/02/25 15:30:48 INFO mapred.LocalJobRunner: Starting task: attempt_local284822998_0001_m_000000_0
15/02/25 15:30:48 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]
15/02/25 15:30:48 INFO mapred.MapTask: Processing split: file:.../mapreduce/data/the_constitution_of_the_united_states.txt:0+45119
15/02/25 15:30:48 INFO mapred.MapTask: (EQUATOR) 0 kvi 26214396(104857584)
15/02/25 15:30:48 INFO mapred.MapTask: mapreduce.task.io.sort.mb: 100
15/02/25 15:30:48 INFO mapred.MapTask: soft limit at 83886080
15/02/25 15:30:48 INFO mapred.MapTask: bufstart = 0; bufvoid = 104857600
15/02/25 15:30:48 INFO mapred.MapTask: kvstart = 26214396; length = 6553600
15/02/25 15:30:48 INFO mapred.MapTask: Map output collector class = org.apache.hadoop.mapred.MapTask$MapOutputBuffer
15/02/25 15:30:48 INFO mapred.LocalJobRunner: 
15/02/25 15:30:48 INFO mapred.MapTask: Starting flush of map output
15/02/25 15:30:48 INFO mapred.MapTask: Spilling map output
15/02/25 15:30:48 INFO mapred.MapTask: bufstart = 0; bufend = 75556; bufvoid = 104857600
15/02/25 15:30:48 INFO mapred.MapTask: kvstart = 26214396(104857584); kvend = 26183792(104735168); length = 30605/6553600
15/02/25 15:30:48 INFO mapred.MapTask: Finished spill 0
15/02/25 15:30:48 INFO mapred.Task: Task:attempt_local284822998_0001_m_000000_0 is done. And is in the process of committing
15/02/25 15:30:48 INFO mapred.LocalJobRunner: map
15/02/25 15:30:48 INFO mapred.Task: Task 'attempt_local284822998_0001_m_000000_0' done.
15/02/25 15:30:48 INFO mapred.LocalJobRunner: Finishing task: attempt_local284822998_0001_m_000000_0
15/02/25 15:30:48 INFO mapred.LocalJobRunner: map task executor complete.
15/02/25 15:30:48 INFO mapred.LocalJobRunner: Waiting for reduce tasks
15/02/25 15:30:48 INFO mapred.LocalJobRunner: Starting task: attempt_local284822998_0001_r_000000_0
15/02/25 15:30:48 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]
15/02/25 15:30:48 INFO mapred.ReduceTask: Using ShuffleConsumerPlugin: org.apache.hadoop.mapreduce.task.reduce.Shuffle@1bd34bf7
15/02/25 15:30:48 INFO reduce.MergeManagerImpl: MergerManager: memoryLimit=334338464, maxSingleShuffleLimit=83584616, mergeThreshold=220663392, ioSortFactor=10, memToMemMergeOutputsThreshold=10
15/02/25 15:30:48 INFO reduce.EventFetcher: attempt_local284822998_0001_r_000000_0 Thread started: EventFetcher for fetching Map Completion Events
15/02/25 15:30:48 INFO reduce.LocalFetcher: localfetcher#1 about to shuffle output of map attempt_local284822998_0001_m_000000_0 decomp: 90862 len: 90866 to MEMORY
15/02/25 15:30:48 INFO reduce.InMemoryMapOutput: Read 90862 bytes from map-output for attempt_local284822998_0001_m_000000_0
15/02/25 15:30:48 INFO reduce.MergeManagerImpl: closeInMemoryFile -> map-output of size: 90862, inMemoryMapOutputs.size() -> 1, commitMemory -> 0, usedMemory ->90862
15/02/25 15:30:48 INFO reduce.EventFetcher: EventFetcher is interrupted.. Returning
15/02/25 15:30:48 INFO mapred.LocalJobRunner: 1 / 1 copied.
15/02/25 15:30:48 INFO reduce.MergeManagerImpl: finalMerge called with 1 in-memory map-outputs and 0 on-disk map-outputs
15/02/25 15:30:48 INFO mapred.Merger: Merging 1 sorted segments
15/02/25 15:30:48 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 90857 bytes
15/02/25 15:30:48 INFO reduce.MergeManagerImpl: Merged 1 segments, 90862 bytes to disk to satisfy reduce memory limit
15/02/25 15:30:48 INFO reduce.MergeManagerImpl: Merging 1 files, 90866 bytes from disk
15/02/25 15:30:48 INFO reduce.MergeManagerImpl: Merging 0 segments, 0 bytes from memory into reduce
15/02/25 15:30:48 INFO mapred.Merger: Merging 1 sorted segments
15/02/25 15:30:48 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 90857 bytes
15/02/25 15:30:48 INFO mapred.LocalJobRunner: 1 / 1 copied.
15/02/25 15:30:48 INFO Configuration.deprecation: mapred.skip.on is deprecated. Instead, use mapreduce.job.skiprecords
15/02/25 15:30:48 INFO mapred.Task: Task:attempt_local284822998_0001_r_000000_0 is done. And is in the process of committing
15/02/25 15:30:48 INFO mapred.LocalJobRunner: 1 / 1 copied.
15/02/25 15:30:48 INFO mapred.Task: Task attempt_local284822998_0001_r_000000_0 is allowed to commit now
15/02/25 15:30:48 INFO output.FileOutputCommitter: Saved output of task 'attempt_local284822998_0001_r_000000_0' to file:.../out/_temporary/0/task_local284822998_0001_r_000000
15/02/25 15:30:48 INFO mapred.LocalJobRunner: reduce > reduce
15/02/25 15:30:48 INFO mapred.Task: Task 'attempt_local284822998_0001_r_000000_0' done.
15/02/25 15:30:48 INFO mapred.LocalJobRunner: Finishing task: attempt_local284822998_0001_r_000000_0
15/02/25 15:30:48 INFO mapred.LocalJobRunner: reduce task executor complete.
15/02/25 15:30:48 INFO mapreduce.Job: Job job_local284822998_0001 running in uber mode : false
15/02/25 15:30:48 INFO mapreduce.Job:  map 100% reduce 100%
15/02/25 15:30:48 INFO mapreduce.Job: Job job_local284822998_0001 completed successfully
15/02/25 15:30:48 INFO mapreduce.Job: Counters: 33
	File System Counters
		FILE: Number of bytes read=283490
		FILE: Number of bytes written=809011
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
	Map-Reduce Framework
		Map input records=872
		Map output records=7652
		Map output bytes=75556
		Map output materialized bytes=90866
		Input split bytes=175
		Combine input records=0
		Combine output records=0
		Reduce input groups=1697
		Reduce shuffle bytes=90866
		Reduce input records=7652
		Reduce output records=1697
		Spilled Records=15304
		Shuffled Maps =1
		Failed Shuffles=0
		Merged Map outputs=1
		GC time elapsed (ms)=8
		CPU time spent (ms)=0
		Physical memory (bytes) snapshot=0
		Virtual memory (bytes) snapshot=0
		Total committed heap usage (bytes)=525336576
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=45119
	File Output Format Counters 
		Bytes Written=17405
$

And you can take a look to the result using:

$ sort -k2 -h -r out/part-r-00000 | head -20
the	663
of	494
shall	293
and	256
to	183
be	178
or	157
in	139
by	101
a	94
United	85
for	81
any	79
President	72
The	64
have	64
as	64
States,	55
such	52
State	47
$

Regarding this example, I have to mention a couple of things:

1. The code includes a data directory containing a text file (yes, the constitution of the USA)
   • Yes, the program need some improvements for not considering the commas and something like that.
   • It’s funny to see the most repeated words (the, of, shall, and, to, be, or, in, by, a) and the most important words (United, States, President)
2. Due to a problem with Hadoop and IPv6, it doesn’t work with pseudistributed mode due to a connection exception (java.net.ConnectException: Call […] to localhost:8020 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused). For this example is enough to use local mode (just recover the original core-site.xml, hdfs-site.xml, mapred-site.xml and yarn-site.xml and you can also stop the HDFS, YARN, and MapReduce daemons. See above)

Resources

• Hadoop: The Definitive Guide, 4th Edition. By Tom White (O’Reilly Media)
• Pro Apache Hadoop, 2nd Edition. By Sameer Wadkar, Madhu Siddalingaiah, Jason Venner (Apress)
• Mastering Hadoop. By Sandeep Karanth (Packt publishing)
• Hadoop in Practice, Second Edition. By Alex Holmes (Manning publications)
• Hadoop in Action, Second Edition. By Chuck P. Lam and Mark W. Davis (Manning publications)
• Hadoop – Just the Basics for Big Data Rookies. By Adam Shook (SpringDeveloper YouTube channel)
• Getting started with Spring Data and Apache Hadoop. By Thomas Risberg, Janne Valkealahti (SpringDeveloper YouTube channel)
• Hadoop 201 — Deeper into the Elephant. By Roman Shaposhnik (SpringDeveloper YouTube channel)
• Hadoop MapReduce Next Generation – Setting up a Single Node Cluster.
• The File System (FS) shell
• Primeros pasos con Hadoop: instalación y configuración en Linux. By Juan Alonso Ramos (Adictos al trabajo)
• How-to: Install a Virtual Apache Hadoop Cluster with Vagrant and Cloudera Manager. By Justin Kestelyn (@kestelyn Cloudera blog()
• How to build a Hadoop VM with Ambari and Vagrant. By Saptak Sen (Hortonworks blog)
• Hadoop HDFS Data Flow IO Classes. By Shrey Mehrotra (Hadoop Ecosystem : Hadoop 2.x)