Much of the traditional software operations we deal with are online, we click a button, wait for a moment, and the transaction or operation is completed. But there is a big area that deals with software operations that require offline processing. For example, background processing of jobs, e.g., OpenAI training/improving its existing GPT models behind the scenes using the data it gathers from its users. Batch Processing Whenever such an offline system runs a job that typically generates output from a batch of inputs, we call that batch processing. Inputs here are immutable, which avoids side effects. Benefits of batch processing: You can time travel. In case of any failure or unintentional outputs, you can jump to the last input checkpoint before a batch processing job. This handling is often referred to as human fault tolerance. Using batch processing and offline systems, compute usage efficiency can be improved. For example, whenever a heavy computation needs to be done, it's better to do it in bulk on maybe a GPU compute rather than crashing the CPU host where the server is online. Though the boundary between online and batch processing is not always clear. For example, a long-running database query could also be categorised as batch processing. Another alternative to batch processing is stream processing, which we will understand in the next article. MapReduce MapReduce is a batch processing algorithm that is utilized by Hadoop, CouchDB, and MongoDB as well. It is a balanced approach that is less extreme than completely parallelizing the jobs. There are several other frameworks like this that are now replacing MapReduce. For example, DataFrames APIs, query languages, etc. We will see MapReduce in detail sometime later. Simulating Batch Processing with Unix Tools (Single Host) If you are a Linux user, this simulation could be very easy for you to grasp. If not, just put it in ChatGPT or any AI tool to understand the command in detail if interested. A typical Nginx command looks like this: 216.58.210.78 - - [27/Jun/2025:17:55:11 +0000] "GET /css/typography.css HTTP/1.1" 200 3377 "https://martin.kleppmann.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36" If we generalize it: $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent" To get the first 5 popular pages, we can run the following command: cat /var/log/nginx/access.log | # read the log file awk '{print $7}' | # separate each log line by space, then extract URL sort | # sort the URLs uniq -c | # count duplicates sort -r -n | # sort by count (-r = reverse/desc) head -n 5 # show top 5 A sample output looks like this: 4189 /favicon.ico 3631 /2016/02/08/how-to-do-distributed-locking.html 2124 /2020/11/18/distributed-systems-and-elliptic-curves.html 1369 / 915 /css/typography.css The sort utility in Unix or Linux can handle a huge amount of data even if it does not fit in memory. It utilizes disk space to optimally sort it. This is an example of batch processing. The only thing is, this is done on a single machine and not a distributed machine. Imagine doing this using any programming language: Pseudo code: file_content = read(/var/log/nginx/access.log) list_of_urls = GetListOfUrls(file_content) url_mapping = GetURlvsCountMapping(list_of_urls) PrintTopFiveUrls(url_mapping) The above approach will bring the whole hashmap of URLs into memory, while the program will crash if the size of data is greater than the available memory. In the Unix pipe-based approach, disk was utilised and an efficient batch processing job was achieved. We are aware that the Unix command which we handled above utilised several components, a storage device accessed by the host OS, a scheduler allocating CPU and resources at a particular time, and different Unix-written programs (e.g., sort, awk) connected together by pipes. A similar analogy exists in distributed batch processing frameworks, which we will see ahead. Batch Processing in Distributed Systems A distributed framework might contain a distributed filesystem accessed by different OSs on different hosts, each having its own scheduler, filesystem, and programs. Distributed Filesystems We must understand a basic filesystem that is provided by our OS. It has several layers. The lowest level is device drivers that speak directly to the disk. The above layers utilize this for I/O. Above this, there is a page cache layer that keeps the recently accessed blocks in memory/cache for quick access (LRU). The block layer is accessed by the block API, which is wrapped in the filesystem layer. So a large file is broken into blocks and read or written. To track metadata, metadata is provided to that file, which is called inodes. Ultimately, the OS exposes the filesystem to applications (such as sort or Python code) via the Virtual File System or VFS API. A rough idea is shown in the following image: A distributed filesystem (DFS) works in a similar way. Files are broken into blocks, which are distributed across several machines. Obviously, DFS blocks are larger than local blocks. It should also be understood that larger DFS blocks result in partial usage by a single file. For example, a 900 MB file stored with 128 MB blocks would have seven blocks that use 128 MB and one block that uses 4 MB. Since machines are different, network requests are required to the machine where a running daemon exposes the API that allows remote reading and writing of blocks on their local filesystem. These daemons are called data nodes. A distributed page cache is also implemented by DFS. Some even have client-side caching. Just like a local filesystem keeps track of metadata including inodes, free space, block locations, directory trees, permissions, etc., DFS also does that in different ways. For example, Hadoop uses a NameNode service that maintains the metadata for the cluster. As discussed earlier, batch processing systems should be able to read and write files in the DFS, so the DFS provides a VFS for that. This can be a protocol or an interface. For example, the S3 API. Some DFS implement POSIX-compliant filesystems. To make it simple, it is like an operating system's VFS for a local filesystem. Another very popular example of a VFS protocol is NFS. DFSs that are NFS-compatible are more scalable as these can be accessed over the network via different clients on a single server. NFS clients connect to one endpoint, but underneath, these systems communicate with distributed metadata services and data nodes to perform the read/write operation. Sometimes, replication is adopted. File blocks are replicated across multiple machines (simply keeping copies on several machines). Replication can improve the efficiency of read batch operations. Object Stores Object stores are seen as an alternative to DFS. A very popular example is S3, and another is Azure Blob Storage. Batch processing jobs can utilize these instead of DFS. Objects are immutable. Unlike files in a filesystem, objects once written can only be fully rewritten using a put call. An update is a full rewrite. A get call can be used to read the object. Each object in an object store has a URL. For example, there is an S3 URL for each S3 object. There are no file handle APIs in object stores. Objects are also not organized as directory trees. It's as simple as having a key. To represent an empty object, a zero-byte object can be used. For example, in the S3 URL s3://my-photo-bucket/2025/04/01/birthday.png, the key is /2025/04/01/birthday.png. Similarly, other features in DFS like hard links, symlinks, file locking, etc., are not present in object stores. They are different from filesystems. Linking and locks are not supported. Renames are also non-atomic in object stores. Unlike a filesystem, here renaming is deleting + copying. Renaming a directory is the same as renaming each object inside it. In object stores, storage and computation are kept separate, unlike DFS. For example, HDFS allows computing tasks to run on the machines that have the files stored (helpful when code is easier to transfer over the network than big data/files that need to be processed). Key Value Stores vs (Object Stores or DFS) KV stores are optimized for small values (typically kilobytes) and frequent, low-latency reads/writes. In contrast, distributed filesystems and object stores are generally optimized for large objects (megabytes to gigabytes) and less frequent, larger reads. Conclusion Batch processing is an essential part of modern data systems, allowing large volumes of data to be processed efficiently in the background. From simple Unix pipelines running on a single machine to distributed filesystems, object stores, and large-scale processing frameworks, the core idea remains the same: process a batch of immutable input data to generate output reliably and efficiently. In next week's article, we will explore how distributed jobs and workflows are orchestrated and look at different standard batch processing jobs and frameworks. Credits: Many of the concepts discussed in this article are inspired by Designing Data-Intensive Applications (DDIA) by Martin Kleppmann.