Distributed File Systems

We look at very large scale data processing techniques with the following characteristics

• massive amounts of data
• mostly read (OLAP) workload
• efficiency (data propcessed per second) over consistency
• separating storage and computation

Let's say we want to process a data file of size 1TB. This will take ~1 hour to read from SSD (~500MB/s) and ~2.5 hours to read from HDD (~150MB/s). Since the data storage medium is a bottleneck, we might want to use multiple disks to support parallel reads. But on a single machine there is bus limitation. So we might want to use N machines with K disks. This would be cost-efficient using cheap machines but hard to program

File System

A file system stores files (data), along with their metadata (name, access permission, timestamps, file size, location of data blocks) in a data structure called inodes (index nodes)

One machine can have many disks. How should files be stored across disks?

RAID: redundant array of independent disks

RAID is a data storage virtualization technique that combines multiple physical disk drive components into one or more logical units for the purposes of data redundancy and/or performance improvement. Different RAID levels provide different balance among the key goals of availability, fault tolerance, performance and capacity.

RAID 0 (striping)

• Data is striped across multiple disks –> better read/write performace
• No redundancy or fault tolerate (data loss will occur if a single disk fails)

RAID 1 (Mirroring)

• Data is duplicated across two or more disks, providing fault tolerance through redundancy
• Improves read performance but needs more storage and slower writes

RAID 5 (Striping with parity)

• Data and parity information (the XOR of data chunks across disks) are striped across at least three disks
• Improves read performance fault tolerance (can handle 1 disk failure)

Parallel/Distributed File Systems

Features

1. Distributed architecture
   • multiple servers work together to store/manage/access data
2. Data chunking/striping
   • data is divided into smaller chunks and distributed across multiple storage devices
   • enables concurrent reads and writes for improved performance
3. Metadata management
   • metadata can be stored and managed separately from data (stored on a different node)
   • allows for efficient searching, indexing, and data organization
4. Scalability
   • designed to handle large-scale storage and computing
   • can easily expand storage and performance as needed
5. Fault tolerance & data redundancy
   • replication, RAID, erasure coding (Reed–Solomon: recover any 1 block from any N other blocks while withstanding up to K concurrent failures)
6. Load balancing
   • ensures optimal distribution of data and workload across available resources
   • minimizes bottleneck and maximizes performance

Design Considerations

1. Disk failure is the norm given the number of storage machines
2. Most files are large –> block size I/O operations
3. Different programming models with different physical architecture
   a) shared memory: multi-thread with locks on data structures
   b) massage passing: master-slave/producer-consumer

Examples

Google File System (GFS), Apache Hadoop Distributed File Sysmtem (HDFS)

Google File System (GFS)

Features

• Central themes in distributed systems: parallel performance, fault tolerance, replication, (weak) consistency
• MapReduce and BigTable are built on top of GFS
• Single master architecture
• Original paper: https://static.googleusercontent.com/media/research.google.com/en//archive/gfs-sosp2003.pdf

Motivation

• Commodity hardware: error is frequent in large setups
• Large files: 100Mb+
• File operations: read and append, mostly sequential reads

Design Principles

1. Fault tolerance
   • GFS is designed to handle hardware failures and network issues gracefully through replication, checksums, and automatic recovery mechanisms
2. Data replication
   • To ensure durability and availability, the system replicates each chunk across multiple (default=3) nodes (chunkservers)
3. High throughput
   • GFS is optimized for high sustained throughput over low latency (workloads typically involve large data processing tasks)
4. Consistency and atomicity
   • GFS provides a consistent view of the file system using write serializability, chunk versioning, atomic record appends, etc
5. Chunk-based storage
   • GFS stores files as fixed-size chunks and uses a single master server to manage the metadata associated with these chunks
   • Actual chunk data are stored on chunk servers distributed across the cluster
6. Scalability
   • Architecture designed to handle large numberes of files and a vast amount of storage
   • Runs on commodity hardware
7. Master-slave architecture
   • A single master server (SPOF) manages the file system metadata and handles garbage collection and chunk migration for load balancing and failure recovery

Master

1. Stores the full file system metadata in memory for fast operations and background scans
   • namespace (file system tree): hierarchical information about files and directories
   • filename: an array of chunk handlers ???
   • chunk handler: list of chunkserver + chunk version number info
   • access control information
   • current locations of chunks
2. Manages chunk leases (locks), garbage collection (freeing unused chunks) and chunk migration (copying/moving chunks)
3. Fault tolerace achieved using operation log (oplog)
   • namespace and filename metadata are persisted by logging mutations to oplog on the master's local disk
   • oplog is repliacted periodically to another machine (shadow master)
   • new master can be started if the master fails by replaying the oplog
4. Periodically communicates with all chunkservers via heartbeat messages to get states and send commands
   • heartbeat: short message ensuring that the server is alive and collecting statistics on the server
5. Master does not store chunk location information persistently (only in memory)
   • needs heartbeats from chunkservers for chunk location and primary status info upon startup of (new) master

Chunkservers (slaves)

1. Stores file chunks on local disks as Linux files
2. Chunk size: default 64 MB, analogous to page size in RDBMS
3. Each chunk has a chunk handle: global unique identifier assigned by the master
4. Stores a 32-bit checksum (to detect data corruption) in memoery and logged to disk

Consistency

Operations are atomic:

• Namespace mutations are atomic (managed by master using namespace locking)
• Record append operations are atomic
• Master's operation log defines a global order

Ensuring consistency after mutation in the presence of chunk replication

• Applies mutations in order by designating a primary replica
• Uses chunk version numbers to detect stale replicas (managed by the master and garbage-collected when appropriate)

Write Operation

1. Lease acquisition [client –> master]: chunk version increment
2. Primary selection [master –> client]: need to identify a primary replica which contains the latest version of the data being modified
3. Data forwarding [client –> chunknodes]: send data to primary replica node and secondary replica nodes
4. Data buffering and acknoledgement [client –> primary replica]
5. Write command: primary mutation (write on primary replica node) –> secondary mutation
6. Primary acknowledgement [secondary replica nodes –> primary replica node]: secondary mutations are done
7. Client acknowledgement [primary replica –> client]: write operation done

If any step fails the client is informed and retries the write

Colossus: next generation GFS

• Better scalability and availability. Instead of using a single master node to manage metadata, Colossus architecture contains
  • Curators: distributed NoSQL database to store metadata
  • Custodians: responsible for storage management (space balancing, raid, etc.)
• Blog post: https://cloud.google.com/blog/products/storage-data-transfer/a-peek-behind-colossus-googles-file-system

Apache Hadoop Distributed File System (HDFS)

What is Apache Hadoop?

• A platform/ecosystem for big data processing, started by yahoo and inspired by MapReduce and GFS from Google, written mostly in Java
• Core components include
  • Hadoop Distributed File System (HDFS)
  • A Map Reduce implementation
  • Other management services such as YARN

HDFS Motivations

• Lots of cheap PCs but high failure rate
• Write-once and read many times data (logs, archives; processing is mostly read-oriented)
• Large streaming reads over random access (high sustained throughput over low latency)

HDFS Features

• A stand-alone general purpose distributed file system, used as a part of a Hadoop cluster
• Highly scalable (to Petabytes of data)
• Fault-tolerant
• Offers a simplified data acess
  • User sees their files as if they were on a single machine
  • Tuning is required for large deployments
• Limited to file level permission (Linux like security)

HDFS Basic Operations

1. When an input file is added to HDFS:
   • File is split into smaller blocks of fixed size
     • Block: single unit of storage, transparent to the end-use
     • Block size traditionally 64/128/256 MB: large enough to force a choice/tradeoff between a seek and a sequential disk read
       • Want to minimize seek_time/transfer_time where seek_time + transfer_time is the time to read a block
   • Each block is replicated to multiple host (machines)
   • Replication level (default=3) is configurable
2. When a host crashes or is decommissioned, all blocks ware replicated to a new host
3. when a new host is added, blocks are rebalanced to avoid data skew

HDFS Architecture

1. NameNode (master)
   • Manages the file system namespace by storing
     • metadata: file-to-blocks and block-to-DataNodes mappings
     • file/directory structure, access permission, etc.
   • Coordinate file operations
     • Direct clients to DataNodes for reads and writes
     • No data is moved through the NameNode
   • Maintains overall health
     • Periodic communication (heartbeat) with DataNodes for block re-replication, block rebalancing and garbage collection
   • Communicates with Secondary NameNode: checkpointing
   • Executes file system namespace operations
     • opening, closing, renaming file/directories, etc.
2. DataNode (slaves)
   • Actual storage and management of data block on a single host
   • Provides client with access to data
   • Perform block creation, deletion and replication upon instruction from NameNode
   • Sends heartbeat (every 3 seconds) and blockreport (every 6 hours) to NameNode
3. Secondary NameNode: Checkpointing, NOT a failover NameNode
   • The NameNode keeps the current state of the HDFS in an fsimage (file system image) file, the last checkpoint of the namespace
   • All modifications to the HDFS are written to a log (edits)
   • During startup, NameNode merges fsimage and edits to get to the latest state
   • Secondary NameNode merges fsimage and edits periodically (default is every 1 hour or 1 million transcations)

Replication

• Maximize reliability, availability and bandwidth through opportunistic execution
• NameNode determins replica placement
  • First replica on the local node (where the writing process happens) or a random node
  • Second replica is usually on the same rack as the first replica to reduce cross-rack network I/O
  • Third replica on a different node in a different rack
  • Consider geo-replication for disaster recovery

Rebalancer & Block Placement Policy

Considerations

• Non-uniform distribution of blocks across hosts (e.g. when hosts are added/removed)
• Different hardware generation (different performace)
• Current hardware unitilization across hosts
• HDFS operates on block level and not on file level

The system administrator can perform periodic cluster-wide rebalancing of blocks:
hadoop balancer <threshold> (threshold X%, default=10)

• Each node's utilization ratio is within X% of the overall cluster utilization ratio (utilized_space/overall_capacity)
• Smaller threshold leads to more balanced distribution but might take a lot of time

HDFS allows for custom block placement policies, such as

• using servers with lower utilization first
• operating on file level
• assigning weights to DataNodes
• moving more blocks to newer generation hardware

Distributed Computation

Hadoop v1.0 (MapReduce)

What is MapReduce?

MapReduce can refer to

• the programming paradigm/model using map (group by) and reduce (aggregate) functions
• the execution framework or specific implementation of the paradigm: the data processing layer of Hadoop

In MapReduce, a job defines, schedules, monitors, and manages the execution of Hadoop Map Reduce .jar files. A job splits the input file(s) into chunks which are processed by multiple (map/reduce) tasks.

What can M/R do? e.g. estimate

$\pi$ using (quasi-)MC methods.

Hadoop 1.0 Architecture

1. Job Tracker
   • one Job Tracker per Hadoop instance
   • manages job life cycle
   • receives and accepts job submissions from clients
   • manages resources and scheduling for each task
   • communicates with task trackers to deploy and run tasks
   • attempts to assign tasks to support data locality
2. Task Tracker
   • one task tracker per node
   • runs and manages individual (map/reduce) tasks locally
     • a mapper/reducer is a node running a map/reduce task
   • communicates progress of tasks back to Job Tracker
3. MapReduce (data processing) is laid on top of HDFS (data storage/accessing)
   • One NameNode, one Secondary NameNode, one Job Tracker node
   • Each slave node contains a TaskTracker process and a DataNode process

MapReduce Implementation

Client provides Map and Reduce functions; everything else is handled by Hadoop M/R.

Steps

1. Input Reader
   • divide input into splits (data chunks) and send each split to a map task
2. Map Task
   • apply the Map function to each split
   • each Map returns a list of key-value pairs
   • Map tasks run in parallel,
   • usually set to 1 mapper per block (at most #DataNode tasks)
3. Shuffle
   • the output of each mapper is sorted by key on the map instance and send to different reducers accordingly
     Shuffle and Sort on the Map Instance
     • Map outputs are buffered in memory using circular buffer (queue, 100 MB), and contents are spilled to disk when 80% full
     • After the mapper finishes, all spill files on the DataNode are merged into a single file, which is then sorted by key and partitioned based on the number of reducers
     • An optional combiner (local reducer that runs in memory after map phase) can be executed by the spilling thread before writing tuples to disk, reducing I/O and network traffic
   • records from across Map tasks with the same key are sent to the same reducer for aggregation
     Shuffle and Sort on the Reduce Instance
     • Map outputs are copied over to the reducer nodes
     • "Sort" is done by a multi-pass merge of map outputs: once a mapper finishes and sends the data to the reducer node, the chunk is combined with existing data in a sorted manner
     • The final merge pass goes directly into the reduce task
4. Reduce Task
   • apply the Reduce function to each key and output a list of key-value pairs
   • need to wait for all mappers to finish (bottleneck) before starting
   • set the number of reducers depending on the expected number of keys: too few create bottleneck while too many create extensive network traffic
     • can set to one reducer if we just want sorted output

Things to handle

• Scheduling: assign nodes to map/reduce tasks
• Data distribution: move processes to data (locality)
• Synchronization: gather, sort and shuffle intermediate data
• Error and faults
• Detect worker failures and restarts
• Everything happen on top of HDFS (intermediary results, shyffling partitions, etc.)

Hadoop v2

Changes over v1

1. YARN replacing old architecture
2. HDFS 2: NameNode improvements
   • High availability: add an additional passive (failover) NameNode which is updated periodically
   • Federation: multiple NameNodes responsible for subnamespaces
3. scheduling (FIFO, capacity scheduling, fair scheduling, delay scheduling)

Problems with Hadoop v1.0

• Only one JobTracker
  • limited horizontal scalability (of Task Trackers)
  • fault tolerance and maintenance: if JT dies/stops all jobs must restart
• Rigid programming model: only supports MapReduce

YARN: Yet Another Resource Negotiator

• Generalized cluster management
• Reponsible for resources in the clusters
  • resource: CPU/memory/etc.
    $\times$ time required by a program
• Any node can become a job tracker for a given job
• Tasks are now "containers" (packages of software that contain all of the necessary elements to run in any environment)

YARN Components

• Resource Manager
  • one node per cluster (can configure a failover node)
  • manages job scheduling and execution (but NO job tracking)
  • manages global resource allocation
• Application Master
  • one node per job, replaces JobTracker in Hadoop v1
  • manages task scheduling and execution for a given job
• Node Manager
  • One process per node, similar to TaskTracker in Hadoop v1
  • manages the lifecycle of task containers
  • reports to Resource Manager on health and resource usage

Execution flow of M/R on YARN

See p.10, lecture9_hadoops
Client submit application -> ResourceManager starts MRAppMaster (via NodeManager) ->MRAppMaster starts taskJVM (via NodeManager) to run M/R Task

Job Scheduling

Job scheduling should follow these resource management design principles:

• provide fast response times to small jobs in a multitenant Hadoop cluster (many users)
• improve utilization: proper load balancing, hotspot avoidance
• improve data locality: running task closer to data (same node/rack)

Hadoop has a taste for simplicity. YARN is a centralized resource negotiator with very basic scheduling and decision techniques:

1. FIFO: suffer from poor parallelism and small jobs starvation
2. Capacity schedular
   • Organize jobs into (hierarchical) queues
     • One queue for production (heavier) jobs (80% overall capacity)
     • One queue for test (lighter) jobs (20% overall capacity)
   • FIFO scheduling withing each queue
   • Suppoerts preemption: system can stop a job/task to maintain the promised capacity
   • Issue: underutilization
3. Fair scheduler
   • Conceptually similar to Capacity Scheduler
   • Group jobs into "pools" (implemented as queues) and assign to each pool a guaranteed minimum share of resources
   • On average every job in a pool will get an equal share overtime
   • Extra capacity is diveded evenly between pools
   • Implementation: maintain a priority queue of jobs (not tasks) by the number of running tasks (small is higher priority). When a resource is free, assign it to the head of queue (the job with the least number of running tasks)

However, the schedulings hurt data locality: the job picked by policy may not have data on free nodes

4. Delay Scheduling: trade time for locality
   • force jobs to wait for a short time if they cannot launch local tasks
   • Empirically 1-5 seconds helps reaching almost 100% map locality
   • It works well under two conditions
     • many tasks are short relative to jobs
     • there are many locations where a task can run efficiently (block repliaction)

Hadoop v3

Long-term resource planning (cloud computing)

• Resources: cumulative CPU and memory over time
  • Static capacity attribution managed by the system admin is not feasible in a cloud setup (millions of users)
  • resources underutilization: fluctuating max cluster capacity (hardware failure/maintenance) and fluctuating demands
  • when reaching max capacity: can move/prioritized jobs to the extent that SLA (serve-level agreement) allows

Improvements over v2

1. Better YARN timeline service which improves scalability and reliability
2. Intra DataNode balancing: alleviates skew within DataNode
3. Erasure coding replacing replication factor of 3: storage overhead (extra storage needed) is reduced to 50% as opposed to 200%
4. NameNode HA (high availability): supports 2 or more standby NameNodes as opposed to only one passive NameNode