Batch Processing and Map-Reduce in Data-driven Applications

The web, and increasing numbers of HTTP/REST-based APIs, has made the request/response style of interaction so common. It’s important to note that it’s not the only way of building systems. Let’s distinguish 3 different types of systems:

Services (online systems)

A service waits for a request or instruction from a client to arrive. When one is received, the service tries to handle it as quickly as possible and sends a response back. Response time is usually the primary measure of performance of a service, and availability is often very important.

Batch processing systems (offline systems)

A batch processing system takes a large amount of input data, runs a job to process it, and produces some output data. Jobs often take from a few minutes to several days, so there normally isn’t a user waiting for the job to finish. Instead, batch jobs are often scheduled to run periodically (e.g., once a day). The primary performance measure of a batch job is usually throughput (the time it takes to crunch through an input dataset of a certain size).

Tools include Apache Spark and Hadoop.

Stream processing systems (near-real-time systems)

Stream processing is somewhere between online and offline processing. Like a batch processing system, a stream processor consumes inputs and produces outputs (rather than responding to requests). However, a stream job operates on events shortly after they happen, whereas a batch job operates on a fixed set of input data. This difference allows stream processing systems to have lower latency than batch systems.

Tools include Apache Flink and Kafka.

Batch processing is an important building block in the quest to build reliable, scalable, and maintainable applications. MapReduce, a batch processing algorithm published in 2004, was called “the algorithm that makes Google so massively scalable”.

MapReduce

MapReduce is a fairly low-level programming model compared to the parallel processing systems that were developed for data warehouses, but it was a major step forward for processing at scale on commodity hardware.

A single MapReduce job takes one or more inputs and produces one or more outputs. It does not modify the input and does not have any side effects other than producing the output.

MapReduce jobs read and write files on a distributed filesystem across potentially thousands of machines using the write-once-read-many access model. In Hadoop’s implementation of MapReduce, that filesystem is called HDFS (Hadoop Distributed File System.

Hadoop is a programming framework with which you can write code to process large datasets in a distributed filesystem like HDFS.

Before we go into the MapReduce algorithm, it’s worth knowing how HDFS works.

Hadoop Distributed File System

HDFS is a scalable, fault-tolerant, and high-throughput distributed file system designed to store massive datasets across commodity hardware in Apache Hadoop clusters. It uses a master-slave architecture with NameNodes for metadata management and DataNodes for data storage, breaking large files into 128MB blocks. HDFS provides high reliability through data replication, with a default replication factor of 3.

NameNodes and DataNodes

An HDFS cluster consists of a single NameNode — a master server that manages the file system namespace. There are a number of DataNodes — usually one per node in the cluster, which manage storage attached to the nodes that they run on. Internally, a file is split into one or more blocks and these blocks are stored in a set of DataNodes.

The NameNode:

• Executes file system namespace operations like opening, closing, and renaming files and directories.
• Determines the mapping of blocks to DataNodes and file system properties using FsImage.
• Regulates access to files by clients.
• Manages replication.
• Maintain EditLog which keep all file modifications on file.

Secondary NameNode:

• Prevents EditLog from growing endlessly by merging it with the FsImage
• Replicate EditLog and FsImage of the Namenode
• Used as recovery node for the Namenode

The DataNodes:

• Serves read and write requests from the file system’s clients.
• Perform block creation, deletion, and replication upon instruction from the NameNode.
• Sends periodic heartbeat to the NameNode

HDFS Rack Awareness

A rack is a physical collection of 30–50 DataNodes connected to the same network switch. Rack awareness is a fault-tolerance feature that maps DataNodes to physical, network-connected racks, allowing the NameNode to distribute data replicas across different, distinct racks. This architecture prevents data loss during switch failures and optimizes network traffic by preferring to read data from closer nodes.

When writing data, the default policy is to:

• place the first replica on the local node (or same rack)
• the second on a different rack, and
• the third on a different node in that same second rack.

If a network switch fails, the entire rack becomes unavailable, but the data remains accessible from other, separate racks.

Imagine you are managing a small Hadoop cluster with the following specifications:

• Number of DataNodes (Nodes): 10
• Disk Capacity per Node: 4 TB
• Replication Factor: default
• Reserved Capacity: 20% (Reserved for OS, Hadoop metadata, and Temp space)

How can we find the maximum possible file size?

Calculate Total Raw Storage: 10 x 4TB = 40TB

Calculate Usable Storage (accounting for 20% reserved space): 40TB x (1–0.2) = 32TB

Divide by Replication Factor (to get actual file storage): 32TB / 3 = 10.66TB

Maximum Theoretical File Size: ~10.66 TB

MapReduce Job Execution

A MapReduce job has 4 main steps:

1. Input data splitting — Read a set of input files from a distributed file system (HDFS), and break it up into fixed-size, logical chunks called input splits or records.

2. Mapping —processes its assigned input split in parallel.

• A RecordReader converts the raw input data from the split into a set of key-value pairs. The user-defined Map function is applied to each key-value pair, generating zero or more intermediate key-value pairs as output.
• A Combiner (a local reducer) performs local aggregation to optimize performance by reducing the amount of data transferred over the network.

3. Shuffling and sorting — Partitions, shuffles and sorts key-value pairs:

• Partitioning determines which Reducer will receive which intermediate key-value pairs, typically based on a hash of the key.
• Shuffling involves the physical transfer of the data over the network to the appropriate Reducer nodes.
• Sorting groups all intermediate values with the same key together and sorts them by key, providing the Reducer with a list of all values for a given unique key.

4. Reducing —The user-defined Reduce function is called once for each unique key and its associated list of values. If there are multiple occurrences of the same key, the sorting has made them adjacent in the list, so it is easy to combine those values without having to keep a lot of state in memory.

Those four steps can be performed by one MapReduce job. Steps 2 (map) and 4 (reduce) are where you write your custom data processing code. Step 1 (breaking files into records) is handled by the input format parser. Step 3, the sort step, is implicit in MapReduce — the output from the mapper is always sorted before it is given to the reducer.

In Hadoop MapReduce, the mapper and reducer are each a Java class that implements a particular interface:

// Mapper
public class WordCountMapper extends MapReduceBase implements
Mapper<LongWritable,Text,Text,IntWritable>{
  private final static IntWritable one = new IntWritable(1);

  public void map(LongWritable key, Text value,OutputCollector<Text,IntWritable> output,
  Reporter reporter) throws IOException{
    String line = value.toString();
    StringTokenizer tokenizer = new StringTokenizer(line);

    while (tokenizer.hasMoreTokens()){
      output.collect(new Text(tokenizer.nextToken()), one);
    }
  }
}

// Reducer
public class WordCountReducer extends MapReduceBase implements
Reducer<Text,IntWritable,Text,IntWritable> {

  public void reduce(Text key, Iterator<IntWritable> values,OutputCollector<Text,IntWritable> output,
  Reporter reporter) throws IOException {
    int sum=0;

    while (values.hasNext()) {
      sum+=values.next().get();
    }
    output.collect(key,new IntWritable(sum));
  }
}

Example

Imagine we start with the following set of data: [Deer, Car, Deer, Car, River, Bear] stored in a file of 256MB. In the first step, we split the input into 2 chunks:

• Split 1: [Deer, Car, Deer]
• Split 2: [Car, River, Bear]

In the map phase, convert each split into key-value pairs:

• Mapper 1 output: [Deer: 1, Car: 1, Deer: 1]
• Mapper 2 output: [Car: 1, River: 1, Bear: 1]

In the shuffle and sort phase, the framework groups all values by key across all mappers:

(Deer, [1, 1]) (Car, [1, 1]) (River, [1]) (Bear, [1])

Then, each reducer processes one key and its list of values:

(Deer, 2) (Car, 2) (River, 1) (Bear, 1)

Overall, we created 2 splits. So the no.of mapper objects = 2 and map() is called once per input record (per element in the split). So no.of calls to map() = 6.

Here we configured 1 reducer. So there is only 1 reducer object. reduce() is called once per unique key. So there were 4 calls to reduce().

Note that if our data is skewed, then 1 or 2 reducers process most of the data while others sit idle, creating a reducer bottleneck.

MapReduce workflows

The range of problems you can solve with a single MapReduce job is limited. Thus, it is very common for MapReduce jobs to be chained together into workflows, such that the output of one job becomes the input to the next job.

The Hadoop MapReduce framework does not have any particular support for workflows, so this chaining is done implicitly by directory name: the first job must be configured to write its output to a designated directory in HDFS, and the second job must be configured to read that same directory name as its input. From the MapReduce framework’s point of view, they are two independent jobs.

A batch job’s output is only considered valid when the job has completed successfully (MapReduce discards the partial output of a failed job). Therefore, one job in a workflow can only start when the prior jobs (jobs that produce its input directories) have completed successfully. To handle these dependencies between job executions, various workflow schedulers for Hadoop have been developed (e.g., Airflow).

These schedulers also have management features that are useful when maintaining a large collection of batch jobs. (e.g., workflows consisting of 50 to 100 MapReduce jobs are common). Tool support is important for managing such complex dataflows. Various higher-level tools for Hadoop, (such as Hive) also set up workflows of multiple MapReduce stages that are automatically wired together appropriately.

Hadoop YARN

Hadoop YARN (Yet Another Resource Negotiator) is the core cluster management and resource scheduling technology in Hadoop 2.0 and later, designed to manage system resources across distributed nodes. By separating resource management from data processing, YARN allows multiple engines like Spark and MapReduce to run simultaneously.

Key Components of YARN Architecture:

• ResourceManager (RM): The master daemon that arbitrates cluster resources among all running applications. It consists of a Scheduler (allocates resources) and an ApplicationsManager (manages job submissions).
• NodeManager (NM): A per-machine agent responsible for launching and monitoring application containers, managing resource usage, and reporting node health to the ResourceManager.
• ApplicationMaster (AM): A framework-specific library that negotiates resources from the RM and works with the NMs to execute and monitor tasks for a specific application.
• Container: A fraction of physical resources on a node, allocated by the scheduler to a specific task.

Applications of Batch workflows

So what is the result of all of that processing, once it is done? OLTP queries generally look up a small number of records by key, using indexes, in order to present them to a user (for example, on a web page).

In contrast, analytic queries often scan over a large number of records, performing groupings and aggregations, and the output often has the form of a report: a graph showing the change in a metric over time, or the top 10 items according to some ranking, or a breakdown of some quantity into subcategories.

Where does batch processing fit in? It is not transaction processing, nor is it analytics.

Building Search Indexes

Google’s original use of MapReduce was to build indexes for its search engine. A full-text search index typically works like this: it is a file (the term dictionary) in which you can efficiently look up a particular keyword and find the list of all the document IDs containing that keyword.

If you need to perform a full-text search over a fixed set of documents, then a batch process is a very effective way of building the indexes: the mappers partition the set of documents as needed, each reducer builds the index for its partition, and the index files are written to the distributed filesystem. Building such document-partitioned indexes parallelizes very well.

Key-value stores as batch process output

Another common use for batch processing is to build machine learning systems such as classifiers (e.g., spam filters, anomaly detection, image recognition) and recommendation systems (e.g., people you may know, products you may be interested in, or related searches).

The output of those batch jobs is often some kind of database: for example, a database that can be queried by user ID to obtain suggested friends for that user. These databases need to be queried from the web application that handles user requests, which is usually separate from the Hadoop infrastructure. So how does the output from the batch process get back into a database where the web application can query it?

The most obvious choice might be to use the client library for your favorite database directly within a mapper or reducer, and to write from the batch job directly to the database server, one record at a time. This will work, but it is a bad idea for several reasons:

• Making a network request for every single record is orders of magnitude slower than the normal throughput of a batch task. Even if the client library supports batching, performance is likely to be poor.
• MapReduce jobs often run many tasks in parallel. If all the mappers or reducers concurrently write to the same output database, with a rate expected of a batch process, that database can easily be overwhelmed.

A much better solution is to build a brand-new database inside the batch job and write it as files to the job’s output directory in the distributed filesystem, just like the search indexes in the last section. Those data files are then immutable once written, and can be loaded in bulk into servers that handle read-only queries. Various key-value stores support building database files in MapReduce jobs (e.g., Voldemort).

Hadoop vs Distributed Databases

When the MapReduce paper was published, it was not at all new. All of the processing and parallel join algorithms that we discussed in the last few sections had already been implemented in massively parallel processing (MPP) databases.

The biggest difference is that MPP databases focus on parallel execution of analytic SQL queries on a cluster of machines, while the combination of MapReduce and a distributed filesystem provides something much more like a general-purpose operating system that can run arbitrary programs.

Storage Flexibility

HDFS accepts data in any format without requiring upfront schema design, enabling the “data lake” approach (collect now, structure later). MPP databases demand careful modeling before ingestion, which slows data collection. The tradeoff is that interpretation burden shifts from data producers to consumers (schema-on-read).

Processing Model Diversity

MPP databases are tightly integrated systems optimized for SQL queries, which works well for business analytics but poorly for tasks like machine learning or image analysis. Hadoop lets engineers run arbitrary code over data, and because multiple processing models (SQL, MapReduce, and others) can share the same cluster and files, there’s no need to move data into separate specialized systems.

Fault Tolerance Design

MPP databases handle failures by aborting and rerunning entire queries, which is acceptable for short queries. MapReduce instead retries at the individual task level and writes intermediate results to disk. This design originated from Google’s environment, where low-priority batch jobs could be preempted by higher-priority tasks at a ~5% hourly rate — far more often than hardware failures. This made granular fault recovery essential, though in environments without aggressive preemption, this overhead is harder to justify.

Thanks for reading 🎉