Back

Unit_5_Service_Models_and_Data_Intensive

Loading views...
← PreviousNext →

Unit 5: Service Models and Data-Intensive Computing

5.1 What is Data-Intensive Computing?

Data-intensive computing is concerned with the production, manipulation, and analysis of large-scale data — from hundreds of megabytes (MB) up to petabytes (PB) and beyond.

Dataset: A collection of information elements relevant to one or more applications. Datasets are stored in repositories that support storage, retrieval, and indexing. Metadata (data about data) is attached to datasets to help with classification and search.

5.1.1 Scale of Data in the Real World

Domain Data Scale Example
Astronomy PB/year Telescopes mapping the sky produce hundreds of GB/second
Bioinformatics TB Mining genome databases
Earthquake Simulation TB Recording Earth vibrations globally
Telecom 10–100 TB Customer data per company (used for billing + trend analysis)
US Mobile Traffic (2015 projection) 327 PB/month Actual mobile data traffic
Google 24 PB/day Data processed by Google's infrastructure
Facebook 36 PB (total) Uncompressed data in distributed infrastructure
Facebook inbox search 150 TB Data crawled per inbox search operation
Zynga (gaming) 1 PB/day Data moved daily from game events (FarmVille, FrontierVille)

5.1.2 Characteristics of Data-Intensive Applications

Data-intensive applications are NOT just about big storage — they also have compute-intensive properties:

                HIGH DATA COMPLEXITY (PB, distributed, heterogeneous formats)
                         ^
                         |
          Data/Compute   |   Data-Intensive
          intensive     |   problems          <-- Our focus
          problems      |   (big + complex)
                         |
          ------------------------------------> HIGH COMPUTATIONAL COMPLEXITY
          
          Simple/Model  |   Compute-Intensive
          Solvers       |   problems
                         |
                   LOW DATA COMPLEXITY (MB/GB, homogeneous formats)

Key Properties of Data-Intensive Apps:

  • Datasets in scale of multiple TB to PB
  • Data in heterogeneous formats, distributed across many locations
  • Processing in multi-step analytical pipelines (transform → fuse → analyze)
  • Processing scales linearly with data size (more data = more nodes needed)
  • Embarrassingly parallel — data can be split and processed independently

5.1.3 Open Challenges in Data-Intensive Computing

Challenge Explanation
Scalable algorithms Must search/process massive datasets efficiently
Metadata management Handle complex, heterogeneous, distributed data sources
Petascale file systems High-performance, high-reliability distributed file systems
Data signature generation Reduce data size without losing meaningful content
Software mobility Move computation to where the data is (not the reverse)
Hybrid interconnects Filter multi-GB data streams from high-speed instruments
Software integration Combine modules from different platforms into pipelines
"Move computation to the data" — In traditional computing, we move data to where the program runs. In data-intensive computing, we move the program to where the data is stored (e.g., MapReduce). This avoids transferring TB of data across the network.

5.2 Historical Development of Data-Intensive Computing

Phase 1: High-Speed Wide Area Networking (1989–1996)

1989: First high-speed network experiments for remote scientific visualization
1991: SC91 (Supercomputing 1991) — remote MRI brain scan visualization
      (Pittsburgh Supercomputing Center → Albuquerque over wide-area network)
      
1993: Kaiser Project — WALDO system capabilities:
      - Auto-generation of metadata
      - Real-time data cataloguing while processing
      - Cooperative research via remote data access
      - Mechanisms to incorporate data into databases

1994: MAGIC Project (DARPA funded):
      - First data-intensive environment
      - Developed DPSS (Distributed Parallel Storage Systems)
      - DPSS later used for TerraVision (3D terrain navigation)

1996: Clipper Project — coordinated collection of services for data-intensive computing:
      - Managed large computing resources
      - Generated/consumed high-rate, high-volume data flows
      - Aggregated dispersed resources (archives, compute, cache, network)

Phase 2: Data Grids (Late 1990s)

What is a Data Grid?

A Data Grid harnesses heterogeneous resources across different administrative domains to provide:

  1. High-performance, reliable file transfer for moving large datasets
  2. Scalable replica discovery and management for accessing distributed datasets
  DATA GRID REFERENCE SCENARIO:
  
  [Telescope/  ]    [Telescope/  ]
  [Particle    ]    [Particle    ]
  [Accelerator ]    [Accelerator ]
        |                  |
        v                  v
  [Storage     ]    [Storage     ]
  [Facility A  ]    [Facility B  ]
        |                  |
        +--------+---------+
                 |
         [Information &    ]
         [Discovery Service]  <-- Helps scientists find where data is
                 |
  +----------------------------------+
  |   COMPUTE SITES                  |
  |  [Site 1] [Site 2] [Site 3]      |
  |  (analyze/process data)          |
  +----------------------------------+
                 |
         [Scientist 1]  [Scientist 2]
         (anywhere in the world)

Unique Challenges of Data Grids:

Challenge Details
Massive Datasets GB to TB scale. Must minimize bulk-transfer latency, replicate content strategically
Shared Data Collections Repositories used for both reading AND writing simultaneously
Unified Namespace Single logical name for each data element, even though it's replicated on multiple physical nodes
Access Restrictions Fine-grained and coarse-grained access control across administrative boundaries

Real Data Grid Examples:

Project Domain What it does
LHC Grid (CERN) High-energy physics Stores/analyzes PB of data from Large Hadron Collider experiments
BIRN Biomedical Data-sharing infrastructure across US research labs (funded by NIH)
IVOA Astronomy Virtual Observatory — interoperating data archives for astronomers worldwide

Phase 3: Data Clouds and "Big Data" (2004–Present)

Big Data Definition:

"Datasets that grow so large they become complex to work with using on-hand database management tools, requiring massively parallel software running on tens, hundreds, or even thousands of servers."

Big Data is NOT just science anymore. It emerged in commercial domains:

Domain Big Data Source
Web logs Every HTTP request to Google, Amazon, etc.
RFID Supply chain tracking (billions of tags)
Sensor networks IoT devices (temperature, location, etc.)
Social networks Facebook posts, likes, connections
Search indexing Crawled web pages, inverted indexes
Medical records Patient data, imaging, genomics
E-commerce Transaction logs, user behavior tracking

How Cloud Computing Supports Big Data (3 Ways):

  1. Compute on demand: Provision thousands of VMs in parallel to process large datasets
  2. Optimized storage: Large blob storage (S3) and distributed data stores
  3. Programming frameworks: MapReduce, Hadoop — APIs optimized for data-at-scale

A Data Cloud = Storage System + Processing Framework:

  Google Data Cloud:  Google File System (GFS) + MapReduce
  Apache Hadoop:      HDFS (GFS-like) + Hadoop MapReduce
  Sector/Sphere:      Sector DFS + Sphere (UDF-based processing)
  Greenplum:          MPP architecture + MapReduce-like functionality
  Aster:              MPP data warehouse + MapReduce, targeting 1 PB

Phase 4: Distributed Databases Limitation

Relational Databases (ACID-compliant): Very robust, support distributed transactions, query optimization. BUT:

  • Designed for structured data (tables, rows, columns)
  • Enforcing ACID properties limits horizontal scalability
  • Cannot scale to Data Grid / Data Cloud levels
ACID Properties: Atomicity (all-or-nothing), Consistency (data integrity rules), Isolation (concurrent transactions don't interfere), Durability (committed transactions survive failures).

5.3 Why Relational Databases Are Not Enough (4 Reasons)

# Reason Explanation
1 Big Data growth Data-intensive workloads are now common across all industries, not just science
2 Data analytics = business value Mining user data is now a core revenue generator, not just a cost (e.g., Facebook)
3 Unstructured data dominates Web pages, blogs, sensor logs, social media — none of this fits neatly into tables
4 Cloud-native architectures Apps deployed on 1000s of volatile nodes need storage that matches — relational DBs can't handle this volatility

5.4 Storage Systems for Data-Intensive Computing

5.4.1 High-Performance Distributed File Systems

(a) Lustre File System

  +-----------------------+
  |    METADATA SERVER    |   <-- Contains file system metadata (file names, paths, permissions)
  +-----------------------+
            |
  +---------+---------+---------+
  |                   |         |
  [Object Storage  ] [Object  ] [Object  ]
  [Server A        ] [Server B] [Server C]
  (actual data stored in objects)
            |
  [Client 1] [Client 2] ... [Client N]
  (POSIX-compliant access — mounts like a normal filesystem)
Feature Detail
Scale Petabytes of storage, thousands of clients
Throughput Hundreds of GB/s
Usage Used by top 500 supercomputers (including world's #1 in June 2012)
Failover Robust failover and recovery — server failures transparent to clients

(b) IBM General Parallel File System (GPFS)

  • Built on shared disks — disks attached to file system nodes via switching fabric.
  • Stripes large files across disk array AND replicates portions for high availability.
  • Supports petabytes at high throughput without losing consistency.
  • Unique: distributes metadata across multiple nodes → no single point of failure for metadata.
  • Supports RS/6000 supercomputers and Linux clusters.

(c) Google File System (GFS)

Design Assumptions (unusual choices):

Assumption Implication
Commodity hardware that often fails Must treat failure as the normal case, not an exception
Modest number of LARGE files (multi-GB common) Optimized for large sequential reads/writes, not millions of tiny files
Workload = mostly large streaming reads + small random reads Design I/O path for sequential throughput
Many large sequential appends Append-optimized interface
High sustained bandwidth > low latency Batch processing focus, not interactive

GFS Architecture:

  +----------------+
  |  MASTER SERVER |  <-- Single master. Stores ALL metadata:
  |  (metadata)    |      - File namespace (names, permissions)
  +----------------+      - File→chunk mapping
          |                - Chunk→chunkserver location
          |  metadata ops
  +-------+-------+-------+-------+
  |               |               |
  [Chunk Server] [Chunk Server] [Chunk Server]
  (store actual  (each file is split into 64 MB chunks,
   data chunks)   chunks replicated on 3+ servers)
          |
  [CLIENT]  -- (1) asks master for chunk location
            -- (2) talks DIRECTLY to chunk server for data

Key Design Decisions:

  • Single Master: Simple design. Potential bottleneck, mitigated by: stateless daemon, replication of master node, extensive logging for recovery.
  • Chunk Size = 64 MB: Reduces metadata size (fewer chunks to track), reduces master bottleneck.
  • Replication = 3 copies by default: One fails → still 2 others available.
  • Snapshots and Record Append: Supported natively — crucial for Google's append-heavy workloads (log files).

(d) Sector Distributed File System

Feature Detail
Type User-space file system (runs without kernel modification)
Deployment Commodity hardware across wide area networks
Key difference Does NOT split files into blocks — replicates entire files on multiple nodes
Custom replication Users configure replication strategy for performance optimization
Protocol UDT (optimized for wide area networks) — NOT TCP
Architecture Security server + Master nodes + Slave nodes + Client machines

(e) Amazon S3 (Simple Storage Service)

S3 is NOT a traditional file system. It's an object store.

  AWS Account
    |
    +-- Bucket: "my-website-assets"
    |     |
    |     +-- Object: "logo.png"  (key="logo.png", value=image bytes)
    |     +-- Object: "style.css" (key="style.css", value=CSS text)
    |
    +-- Bucket: "user-uploads"
          |
          +-- Object: "user123/photo.jpg"
Feature Detail
Interface HTTP REST API (GET/PUT/DELETE) — no special library needed
Storage model Flat namespace — buckets contain objects (key → binary value)
Semantics Simple get/put — no directory hierarchy, no file system operations
Access control Per-bucket: private, shared with specific accounts, or public
Authenticated URLs Time-limited public URLs for controlled temporary sharing
Internal details Not publicly disclosed; claimed: high availability, reliability, infinite scale

5.4.2 General Distributed File System Architecture

All the above systems (except S3) share a common reference architecture:

  +---------------------+
  |  METADATA / MASTER  |  <-- Tracks where each file/chunk is located
  |  NODES              |      (GFS=Master, GPFS=distributed, Sector=Master nodes)
  +---------------------+
            |
  +---------+---------+---------+
  [Slave Node 1] [Slave Node 2] [Slave Node 3]
  (actual storage — data blocks/chunks)
            |
  [CLIENT LIBRARY]  <-- POSIX-compatible interface
  (mount it like a regular filesystem)

Common design goal across all systems: Fault-tolerant, highly available storage.

5.5 NoSQL — Not Only SQL

5.5.1 Why NoSQL?

Traditional RDBMS limitations for Big Data:

  RELATIONAL DB (e.g., MySQL, PostgreSQL):
  - Fixed schema: all rows must have same columns
  - ACID transactions: powerful but expensive (locking, coordination)
  - JOIN operations: require data co-location (doesn't scale horizontally)
  - SQL interface: great for structured data
  - Vertical scaling: add more RAM/CPU to one big server
  
  PROBLEM: When you have 100 TB of unstructured web logs, a social graph
  with 1 billion edges, or sensor readings from 10 million devices...
  ...relational model breaks down.

Two main drivers of NoSQL growth:

  1. Simple data models are often sufficient — many apps only need key→value lookups, not complex SQL joins.
  2. Volume of unstructured data has exploded — web pages, blogs, sensor readings, social content.

5.5.2 NoSQL Data Models (Classification)

NoSQL Type Structure Examples Best For
Document Store JSON/BSON documents (key → document) MongoDB, CouchDB, SimpleDB Content management, catalogs
Key-Value Store Simple key → value DynamoDB, Redis, Riak Session storage, caching
Column Family Table with rows + column families BigTable, HBase, Cassandra Time-series, wide tables
Graph DB Nodes + edges + properties Neo4j, AllegroGraph, FlockDB Social networks, recommendations
Object DB Persistent objects ObjectStore, JADE, ZODB Object-oriented apps
Tuple Store Tuples (ordered sequences) Apache River Coordination services
Multi-value DB Fields can have multiple values OpenQM, Rocket U2 Legacy data

5.5.3 MongoDB and CouchDB (Document Stores)

Both provide:

  • Schema-less storage: Documents can have different fields
  • Each document = collection of key-value fields (value can be string, int, float, array)
  • REST API interface + JSON data format
  • MapReduce for querying and indexing
  • JavaScript as query language (not SQL)
  • Replication + high availability

Difference:

  • CouchDB: Guarantees ACID properties per document.
  • MongoDB: Supports sharding (distributing collection data across multiple nodes for horizontal scaling).
  MongoDB Sharded Collection:
  
  Collection: "users" (10 million documents)
      |
  Shard 1: users where _id in [A-G]  → Node 1
  Shard 2: users where _id in [H-P]  → Node 2
  Shard 3: users where _id in [Q-Z]  → Node 3
  
  Each shard handles its subset; total throughput = sum of all shards.

5.5.4 Amazon Dynamo (Key-Value Store)

Goal: Provide incrementally scalable, highly available storage for Amazon's business services (serving 10 million requests/day across thousands of servers).

Key Design Trade-off:

  • Sacrifices ACID (consistency) for availability and reliability
  • Uses eventual consistency model: All replicas will eventually have the same data, but reads might return slightly stale data during a write

Dynamo Architecture:

  DYNAMO RING (Consistent Hashing):
  
  Each storage peer owns a range of keys (arc of the ring)
  
          [Peer A]
         /        \
    [Peer F]      [Peer B]
        |    KEY     |
    [Peer E]  RING  [Peer C]
         \        /
          [Peer D]
  
  - A key (e.g., "user:12345") is hashed to a point on the ring
  - The peer responsible for that arc handles the key
  - Keys are REPLICATED on the next N-1 peers (not adjacent, to avoid correlated failures)
  - Each peer has a Request Coordinator, Failure/Membership Detection, Pluggable Storage Engine

"Always Writable" Store:

  • Even when some replicas are unavailable, writes always succeed
  • Conflicting versions are resolved in the background using vector clocks and object versioning
  • Trade-off: Applications must handle potential data conflicts themselves

Limitation: No referential integrity, no join operations, no relations. Acceptable for Amazon's use cases.

5.5.5 Google Bigtable (Column-Family Store)

Goal: Scale to petabytes of data across thousands of servers, serving 60+ Google applications.

Data Model:

  Bigtable = Distributed, persistent, multidimensional sorted map
  
  Key structure: (row_key, column_family:column_qualifier, timestamp) → value
  
  Example Table "WebPages":
  
  Row Key     | contents:html          | anchor:cnnsi.com | anchor:my.look.ca
  ------------|------------------------|------------------|------------------
  com.cnn.www | <html>CNN...</html>    | "CNN"            | "CNN.com"
  com.bbc.www | <html>BBC...</html>    |                  | "BBC"
              
  - Row key: arbitrary string (up to 64 KB)
  - Column family: group of related columns (e.g., "contents", "anchor")
  - Each value has multiple TIMESTAMPED versions (auto or user-set)

Bigtable Infrastructure:

  +---------------+      [Chubby Lock Service]
  |  MASTER SERVER|  <-- Tracks tablet server status
  |               |      Assigns tablets to tablet servers
  +---------------+      Uses Chubby for coordination and consistency
         |
  +------+------+------+------+
  [Tablet   ] [Tablet   ] [Tablet   ] [Tablet   ]
  [Server 1 ] [Server 2 ] [Server 3 ] [Server 4 ]
  (each serves requests for a "tablet" = contiguous partition of rows)
         |
  [ GFS (Google File System) ]  <-- Actual data stored as tablet files + logs

Key Components:

  • Tablet Server: Serves 10–1000 tablets per server. Handles read/write for its tablets.
  • Master Server: Monitors tablet servers; reassigns tablets if a server fails.
  • Chubby: Distributed lock service. Manages replicas, provides consistency, monitors system.
  • GFS: Underlying storage — all data stored as files; updates logged for recovery.

Used by: Google Personalized Search, Google Analytics, Google Finance, Google Earth (60+ apps).

5.5.6 Apache Cassandra (Hybrid Column-Family Store)

Origin: Developed by Facebook for inbox search (crawling 150 TB of data). Now Apache open-source.

Key Insight: Cassandra is a combination of Dynamo + Bigtable:

  • From Dynamo: Fully distributed ring architecture, gossip-based membership, eventual consistency
  • From Bigtable: Column Family data model

Architecture:

  Cassandra Ring (like Dynamo):
  - Nodes organized in a ring sharing a key space
  - Each node manages multiple, discontinuous portions of the key space
  - Replication aware of rack, data center, or geography
  - Gossip protocol for node membership and state dissemination
  
  Data model (like Bigtable):
  - Table = distributed multi-dimensional map indexed by key
  - Rows have columns grouped into column families
  - APIs: insert (row-level), retrieve/delete (column-level)

Durability Mechanism:

  • Every write is logged to disk first (commit log) BEFORE being applied in memory
  • In-memory data flushed to disk when size exceeds threshold
  • Read: check memory first, then disk
  • Speed optimization: each disk file has a key summary (avoids full file scan)

Scale: Largest known Cassandra deployment = 100 TB on 150 machines.

Users: Facebook, Digg, Twitter.

5.5.7 HBase (Hadoop's Bigtable)

  • Distributed database for the Hadoop ecosystem
  • Inspired by Google Bigtable (same data model: row + column family + timestamp → value)
  • Provides real-time read/write for tables with billions of rows and millions of columns
  • Backed by HDFS (Hadoop Distributed File System) — which mirrors GFS's design

5.6 Comparison: Distributed File Systems

System Master Nodes Storage File Handling Protocol Use Case
Lustre Metadata Server Object Storage Servers File blocks POSIX HPC Supercomputers
GPFS Distributed metadata Shared disk array File striping + replication POSIX IBM supercomputers, clusters
GFS Single master Chunk servers (64 MB chunks) Large sequential files Custom Google internal apps
Sector Master nodes Slave nodes Full-file replication UDT WAN data-intensive apps
S3 Invisible Object buckets Objects (key→bytes) HTTP REST Cloud blob storage

5.7 Comparison: NoSQL Systems

System Model Consistency Scale Key Feature
MongoDB Document Eventual (configurable) Horizontal (sharding) Rich queries, JSON, sharding
CouchDB Document ACID (per document) Moderate Full ACID on docs, REST API
Amazon Dynamo Key-Value Eventually consistent Massive Always writable, ring topology
Google Bigtable Column-Family Strong (via Chubby) Petabyte Sorted map, timestamped versions
Cassandra Column-Family (hybrid) Tunable consistency Massive Dynamo+Bigtable hybrid, no SPOF
HBase Column-Family Strong Large Real-time on Hadoop/HDFS

5.8 MapReduce Programming Model

5.8.1 What is MapReduce?

MapReduce is a programming platform introduced by Google for processing large quantities of data. It expresses computation as just two simple functions — map and reduce — and hides all the complexity of distributed data management inside the runtime.

Core Philosophy: "Move computation to where the data is." Instead of copying TB of data across the network to a single analysis machine, send the map function to each node that already has a portion of the data. Process locally. Collect results.

Formal Definition:

  map    (k1, v1)         -->  list(k2, v2)
  reduce (k2, list(v2))   -->  list(v2)
  • The map function takes a key-value pair and produces a list of intermediate key-value pairs.
  • The reduce function takes a key and all its associated values and produces a final result.

5.8.2 Step-by-Step MapReduce Execution

  USER PROGRAM
       |
       | (1) Submit input files + specify map & reduce functions
       v
  +----------------------------------+
  |     MapReduce MASTER PROCESS     |
  |  - Partitions input into splits  |
  |  - Assigns splits to map workers |
  |  - Coordinates reduce workers    |
  +----------------------------------+
       |                        |
       | (2) Assign map tasks   | (2) Assign reduce tasks
       v                        v
  [Map Worker 1]           [Reduce Worker 1]
  [Map Worker 2]           [Reduce Worker 2]
  [Map Worker 3]
       |
       | (3) Read input split from distributed file system
       | (4) Run map() function on each record
       | (5) Write intermediate <k2,v2> pairs to local disk
       |
       |----> (5) Master notifies reduce workers of intermediate file locations
       |
  [Reduce Worker 1]
       | (6) Remote-reads intermediate files from map workers
       | (7) Sorts by key (groups same keys together)
       | (8) Runs reduce() on each (key, list_of_values)
       | (9) Writes output to distributed file system
       v
  OUTPUT FILES

Input splits: Input files are divided into 16–64 MB chunks ("splits"). Each split is the input to one map task.

5.8.3 Worked Example: Word Count

Problem: Count how many times each word appears across 10 million documents.

  INPUT:
  "cloud computing is great"
  "cloud is everywhere"

  MAP FUNCTION (applied to each line):
  "cloud computing is great" --> [("cloud",1), ("computing",1), ("is",1), ("great",1)]
  "cloud is everywhere"      --> [("cloud",1), ("is",1), ("everywhere",1)]

  AFTER SHUFFLE & SORT (MapReduce runtime groups same keys):
  ("cloud",     [1,1])
  ("computing", [1])
  ("everywhere",[1])
  ("great",     [1])
  ("is",        [1,1])

  REDUCE FUNCTION (sum the list):
  ("cloud",      2)
  ("computing",  1)
  ("everywhere", 1)
  ("great",      1)
  ("is",         2)

  OUTPUT: word count file

5.8.4 Classic MapReduce Use Cases

Use Case Map Function Reduce Function
Distributed Grep Emit line if pattern matches Collect all matching lines
URL Access Frequency For each log entry: emit <URL, 1> Sum all 1s per URL → <URL, total>
Reverse Web-Link Graph For each link on page: emit <target, source> Collect all sources per target → <target, list(sources)>
Term Vector per Host For each doc: emit <host, term_vector> Merge term vectors per host
Inverted Index For each word in doc: emit <word, doc_id> Collect all docids per word → <word, list(docids)>
Distributed Sort Emit <key, record> per record Copy-through (runtime does sorting)

5.8.5 When Can You Use MapReduce?

Any computation with these two phases:

Phase Description MapReduce Stage
Analysis Operate directly on input data. Embarrassingly parallel — no ordering needed MAP
Aggregation Aggregate/sum/process intermediate results into final form REDUCE

Multi-stage MapReduce: Complex algorithms decompose into a chain of MapReduce jobs:

  Input --> [MapReduce 1] --> Intermediate --> [MapReduce 2] --> Final Output

5.8.6 Fault Tolerance in MapReduce

Failure Type Recovery Strategy
Worker failure Master detects via heartbeat. Re-assigns map/reduce task to another worker. Already-completed MAP tasks re-executed (output on local disk, now inaccessible). Completed REDUCE tasks don't need re-execution (output is in GFS).
Intermediate data loss If a map worker fails, its completed tasks are re-executed.
Master failure Checkpointing: Master periodically saves state to disk. Job can be restarted from last checkpoint.

5.9 MapReduce Variations and Extensions

5.9.1 Apache Hadoop

Hadoop = Open-source implementation of Google MapReduce + GFS

  HADOOP ECOSYSTEM:
  
  +------------------------------------------+
  |          HADOOP MAPREDUCE                 |  <-- Open-source MapReduce
  +------------------------------------------+
  |     HDFS (Hadoop Distributed File System) |  <-- Open-source GFS equivalent
  +------------------------------------------+
  |         Commodity Linux Servers           |
  +------------------------------------------+

HDFS Architecture (mirrors GFS):

  • NameNode = Master (metadata: file→block mapping)
  • DataNodes = Slaves (store actual 128 MB data blocks)
  • Files replicated 3× by default across different racks

Hadoop Ecosystem Projects:

Project Purpose
HDFS Distributed file system (GFS equivalent)
MapReduce Batch processing engine
Pig High-level language (Pig Latin) → compiles to MapReduce jobs
Hive SQL-like interface → generates MapReduce jobs; data warehouse on HDFS
HBase Bigtable-equivalent NoSQL database on HDFS
ZooKeeper Distributed coordination service (like Chubby)
Sqoop Import/export data between HDFS and relational databases

Scale: Yahoo! runs the world's largest Hadoop cluster — 40,000 machines, 300,000+ cores, available to academic institutions.

5.9.2 Apache Pig

Problem: Writing MapReduce programs in Java is verbose and low-level.

Solution: Pig provides a high-level language called Pig Latin — SQL-like, but designed for data transformation pipelines.

  PIG LATIN PROGRAM:
  
  data = LOAD 'access.log' USING PigStorage(' ');
  urls = FOREACH data GENERATE $4 AS url;
  grouped = GROUP urls BY url;
  counts = FOREACH grouped GENERATE group, COUNT(urls);
  STORE counts INTO 'url_counts';
  
  This Pig Latin script is compiled into a sequence of MapReduce jobs
  that run on top of Hadoop!

Advantages over raw MapReduce:

  • Much less code
  • SQL-familiar syntax (LOAD, FILTER, GROUP BY, JOIN)
  • Compiler handles optimization and job chaining

5.9.3 Apache Hive

  • Provides a data warehouse on top of Hadoop MapReduce
  • Supports HiveQL — a SQL dialect for querying HDFS data
  • Converts HiveQL queries into MapReduce jobs automatically
  • Not for OLTP (real-time transactions) — high query latency
  • Perfect for batch analytics: summarization, ad-hoc queries, large dataset analysis
  • Advantage: Scales out (backed by Hadoop); no new infra needed if Hadoop already in place
  HiveQL query:
  SELECT url, COUNT(*) as hits
  FROM web_logs
  WHERE status = 200
  GROUP BY url
  ORDER BY hits DESC
  LIMIT 100;
  
  Hive compiles this into MapReduce jobs running on HDFS.

5.9.4 Map-Reduce-Merge

An extension to MapReduce that adds a third phase:

  Standard MapReduce:  MAP --> [shuffle/sort] --> REDUCE
  Map-Reduce-Merge:    MAP --> [shuffle/sort] --> REDUCE --> MERGE
  • Merge phase: Efficiently merges data already partitioned/sorted by Map and Reduce
  • Simplifies handling heterogeneous related datasets
  • Allows expressing standard relational algebra operators (including various JOIN algorithms)
  • Removes the need to chain multiple MapReduce jobs for operations that naturally involve merging

5.9.5 Twister (Iterative MapReduce)

Problem: Standard MapReduce is not designed for iterative algorithms (like machine learning training loops that run MapReduce dozens of times).

Solution: Twister adds a loop construct:

  1. Configure Map
  2. Configure Reduce
  3. While condition holds:
       a. Run MapReduce
       b. Apply Combine operation to result  <-- new aggregation step
       c. Update condition
  4. Close

Additional Twister features:

  • Map/Reduce tasks can access static in-memory data (shared across iterations without reloading)
  • Combine phase: Aggregates output at end of each MapReduce round before the next iteration
  • Tools for data management across iterations

Use cases: K-means clustering, PageRank, machine learning training loops.

5.10 Alternatives to MapReduce

5.10.1 Sphere (Stream Processing)

  • Built on top of the Sector Distributed File System (SDFS)
  • Processing model: SPMD (Single Program Multiple Data) — same function runs on all data segments
  • Developers write User Defined Functions (UDFs) that process streams
  • A stream = data structure providing access to segments of files in SDFS
  • Sphere Process Engines (SPEs) execute UDFs across distributed nodes
  • Architecture: client-controlled master-slave model
  • Key difference from MapReduce: Sphere is more general — MapReduce is a special case of UDFs

5.10.2 All-Pairs

Abstraction:

  All-Pairs(A: set, B: set, F: function) --> M: matrix
  • Computes function F for every pair (a∈A, b∈B) — results form a matrix
  • Use cases: Biometrics similarity matrices (compare face images pairwise), data mining
  • Naive implementation (nested for-loop) is slow and ignores data distribution, dispatch latency, failures
  • All-Pairs engine: Models performance characteristics, optimally distributes data, dispatches batch jobs

4-phase execution:

  1. Model the system (performance model)
  2. Distribute data (optimally assign data to nodes)
  3. Dispatch batch jobs
  4. Clean up

5.10.3 DryadLINQ (Microsoft Research)

Dryad:

  • General-purpose DAG (Directed Acyclic Graph) computation engine
  • Developers express distributed applications as a DAG of sequential programs connected by channels
  • DAG nodes = programs; edges = data channels between programs
  • MapReduce is a special case of Dryad (Map node → Reduce node)
  • Supports dynamic modification of the computation graph at runtime
  • Auto-parallelizes programs — developers write sequential code, Dryad distributes it

DryadLINQ:

  • Programming environment built on Dryad
  • Uses LINQ (Language Integrated Query) extensions to C#
  • Results: Fully .NET-integrated framework for distributed computing
  • Can express MapReduce, SQL-style operations, and custom distributed algorithms
  C# code with LINQ:
  var result = data
      .Where(x => x.Status == 200)
      .GroupBy(x => x.Url)
      .Select(g => new { Url = g.Key, Count = g.Count() });
  
  DryadLINQ compiles this into a Dryad computation graph
  that runs across thousands of machines.

5.11 MapReduce Frameworks Comparison

Framework Model Input Languages Key Feature
Google MapReduce Map + Reduce Key-value pairs Java, C++, Python Original; uses GFS
Hadoop MapReduce Map + Reduce Key-value pairs Java, Python (Streaming) Open-source; uses HDFS
Pig Dataflow Files Pig Latin (HQL-like) High-level; compiles to MapReduce
Hive SQL-like HDFS files HiveQL (SQL dialect) Data warehouse on Hadoop
Map-Reduce-Merge Map+Reduce+Merge Key-value pairs Extended MR API Relational algebra support
Twister Iterative MapReduce Key-value pairs Java Iterative algorithms (ML, clustering)
Sphere SPMD/UDF streaming Sector file segments C++ UDFs Stream processing on SDFS
DryadLINQ DAG computation Any C# + LINQ Superset of MapReduce; .NET integrated

Navigation

← PreviousNext →