Distributed Resource Management, SQL-to-Hadoop Ingestion, and Streaming Event Collection

5.1 Lecture Overview, Progress Tracking, and MapReduce/Hive Foundation

5.1.1 Big Data Curriculum Roadmap and Syllabus Positioning

Enterprise big data architectures rely on a layered stack of storage engines, distributed execution frameworks, multi-tenant schedulers, and data ingestion pipelines. In the overall big data ecosystem maturity model, foundational infrastructure comprises two core capabilities:

• Infinite Distributed Storage — the ability to store files far larger than any single machine's disk. Provided by the Hadoop Distributed File System (HDFS), which splits large files into fixed-size block replicas (default 128 MB per block, three replicas each) distributed across master NameNodes and slave DataNodes.
• Infinite Parallel Processing — the ability to run computations across hundreds or thousands of machines simultaneously. Provided by the MapReduce (MR) programming paradigm, which executes data-localized map, shuffle, sort, and reduce operations across multi-node clusters.

Having established how HDFS handles block reads and writes and how MapReduce parallelizes key-value computations, the current curriculum focus shifts to two critical operational layers:

1. Resource orchestration — YARN (Yet Another Resource Negotiator), the cluster operating system that allocates CPU, memory, and disk to competing jobs.
2. Data ingestion — Sqoop (SQL to Hadoop) for bulk relational database imports, and Flume (streaming event ingestion) for continuous log collection.

These components bridge relational data stores and unstructured log streams with Hadoop storage and processing engines. Subsequent course modules progress to NoSQL database systems (such as HBase and Cassandra), followed by Apache Spark and distributed machine learning after the mid-semester evaluation. Although current class progression is officially one week behind the initial syllabus calendar, all core topics will be thoroughly covered.

5.1.2 Recap of MapReduce Execution Components and Coin-Counting Walkthrough

To contextualize YARN resource management, consider the physical mechanics of MapReduce execution established through the coin-counting classroom walkthrough. In that exercise, students were tasked with counting and tabulating mixed coin collections (aluminum, copper, silver, and gold coins) distributed across a classroom.

Choosing between single-reducer and multi-reducer topologies:

The choice depends on whether the computation requires global coordination:

• Single Reducer Topology — Required when a global aggregate or total ordering across all keys is necessary (such as identifying the overall top-selling product category or finding the global maximum coin count). All intermediate mapper outputs converge onto a single physical reducer node. This creates a bottleneck but guarantees a total order.

• Multi-Reducer Topology — Used when key groups can be processed completely independently (such as calculating separate coin subtotals for each metal type or computing department-wise sales). Each reducer independently handles a distinct partition of keys in parallel. The partitioner function (by default, HashPartitioner) determines which reducer receives which key.

5.1.3 Hive Architecture and SQL-to-MapReduce Query Translation

Writing native MapReduce programs directly in Java requires managing low-level job configurations, custom serializable data types (Writable), and complex mapper-reducer classes. To democratize big data processing for data analysts familiar with relational database queries, enterprise engineering teams at Facebook developed Apache Hive.

The compilation pipeline — from SQL to MapReduce:

When a user submits a HiveQL query, Hive compiles the SQL statement into a Directed Acyclic Graph (DAG) of underlying MapReduce jobs:

\[ \text{HiveQL Query} \longrightarrow \text{Query Parser / Compiler} \longrightarrow \text{Logical Plan} \longrightarrow \text{Physical MapReduce Job(s)} \]

The query engine parses the SQL syntax, converts relational operations into MapReduce mapper and reducer transformations, submits the generated job to the cluster, and returns the aggregated results to the user.

SELECT department, SUM(sales) AS total_sales
FROM store_sales
GROUP BY department
ORDER BY total_sales DESC;

5.1.4 Student Questions and Answers

5.2 YARN Architecture, Component Decoupling, and Container Abstractions

5.2.1 Evolution from Hadoop 1.0 to Hadoop 2.0 and Module Decoupling

In Hadoop 1.0 architecture, cluster resource management and job lifecycle tracking were tightly coupled inside the MapReduce framework itself. A single master daemon called the JobTracker managed both cluster resource allocation and the execution monitoring of every individual map and reduce task across slave TaskTrackers.

This monolithic architecture suffered from severe structural limitations:

• Scalability Bottlenecks: A single JobTracker could not scale beyond approximately 4,000 nodes or 40,000 concurrent tasks because tracking individual task heartbeats (one per second per task) overwhelmed master memory and CPU resources. The reference text (Tom White, Hadoop: The Definitive Guide) confirms these exact limits.
• Single Point of Failure (SPOF): If the JobTracker crashed, all active jobs failed immediately. There was no mechanism for another daemon to take over, because the rapidly changing in-memory state (each task's status, progress counters, and intermediate results) was too complex to replicate.
• Inflexible Workload Support: Clusters could run only Java MapReduce jobs. Non-MapReduce applications — such as graph processing (e.g., Pregel-style algorithms), iterative machine learning (which needs to loop over data many times), real-time streaming (Storm, Samza), or native C/C++ executables — could not run directly on the cluster infrastructure. Every computation had to be shoehorned into the map-shuffle-reduce paradigm.

YARN acts as a generic, multi-tenant cluster operating system that manages physical resources (CPU, memory, storage, network) independently of application frameworks. Under Hadoop 2.0 and 3.0 (which added YARN Timeline Service v2 for enhanced metrics logging), MapReduce became just one of many application engines running on top of YARN, alongside Apache Spark, Apache Tez, and custom C/C++/Python execution scripts.

Component mapping from Hadoop 1.0 to YARN:

5.2.2 Core Master-Slave Components: Resource Manager and Node Manager

YARN adopts a hierarchical master-slave architecture that mirrors the storage layer topology of HDFS:

\[ \text{Storage Layer: NameNode (Master)} \longleftrightarrow \text{DataNode (Slave)} \] \[ \text{YARN Compute Layer: Resource Manager (Master)} \longleftrightarrow \text{Node Manager (Slave)} \]

The Resource Manager resides on the master node to control global resource distribution, while Node Managers run on slave nodes to monitor local node health and manage physical compute containers.

5.2.3 Container Abstraction and Linux Cgroups Integration

A Container in YARN represents the fundamental, atomic unit of compute and memory resource allocation leased to an application. It encapsulates a specific slice of a physical slave machine's capacity that cannot be broken down further:

\[ \text{Container} = \{\text{VCores}, \text{RAM (MB)}, \text{Disk I/O Bounds}, \text{Network Bandwidth (MBps)}\} \]

A container binds specific quantities of CPU cores, memory, disk I/O, and network bandwidth into an isolated execution sandbox on a slave node.

Key characteristics of YARN containers:

• Resource Boundaries: Configured at the cluster level via Linux Control Groups (Cgroups). For example, a basic container definition might specify 1 VCore and 2 GB of RAM, or 4 VCores and 8 GB of RAM. Even if an application task requires only a fraction of CPU or memory, YARN allocates the full minimum container boundary.

• Dynamic Resource Enforcement: Modern YARN containers enforce multi-dimensional limits using Cgroups, restricting process memory usage, disk write rates, and network bandwidth consumption. This prevents one rogue process from starving its neighbors.

• Container Reusability: Containers are generic and reusable sandboxes. Once a container process completes a mapper task, the same container workspace can be reused to execute subsequent map or reduce operations, eliminating process initialization overhead. This is a significant improvement over Hadoop 1.0's fixed "slots," where a map slot sat idle while reduce tasks waited.

5.2.4 Container Launch Context and Application Master Architecture

Launching a process inside a YARN container requires an explicit specification bundle called the Container Launch Context (CLC). The CLC contains three categories of information:

1. Environment Variables & Local Resources: Specifies environment paths, command-line parameters, required library dependencies, JAR files, and remote resource URIs needed by the application binary.
2. Security Credentials & Tokens: Encapsulates authentication keys and security tokens issued by the Resource Manager, ensuring that only authorized application processes can execute within slave containers. This is a critical security boundary — without valid tokens, a container cannot communicate with other YARN services.
3. Execution Commands: Contains the exact OS-level launch command (such as running a Java class main() method, executing a compiled C++ binary, or invoking a Python script).

5.2.5 Student Questions and Answers

5.3 YARN Execution Flow, Security Handshake, and RM Web Console Administration

5.3.1 Detailed 10-Step YARN Job Lifecycle Walkthrough

When a client application submits a job to a YARN-enabled Hadoop cluster, the system executes a strict multi-stage protocol across the master Resource Manager, slave Node Managers, the Application Master, and task containers. Unlike a monolithic system where one daemon controls everything, YARN distributes responsibility so that no single component becomes a bottleneck.

[Client] ---> (1. Submit Job) ---> [Resource Manager (RM)]
                                          |
                                   (2. Select Slave)
                                          v
[App Master] <--- (4. Launch AM) <--- [Node Manager 1 (NM1)]
     |
     +---> (5. Status Updates) ---> [Client]
     |
     +---> (6. Request Containers) ---> [Resource Manager]
     |                                          |
     |                                   (7. Grant Leases)
     v                                          v
[Node Manager 2 & 3] <--- (8. CLC + Tokens) ---+
     |
     +---> (9. Launch Mappers/Reducers) ---> [Worker Containers]
     |
     +---> (10. Heartbeats & Teardown) ---> [Resource Manager]

5.3.2 Internal Resource Manager Components: Application Manager vs Scheduler

The Resource Manager daemon splits its operational workload between two primary internal components. Understanding the boundary between them is essential: the Application Manager handles application lifecycle, while the Scheduler handles resource allocation. These two concerns must never mix, because combining them (as the JobTracker did) creates the scalability bottleneck that forced the Hadoop 2.0 redesign.

Application Manager (App Mgr). The Application Manager is the gateway through which all client applications enter the YARN cluster. Its responsibilities include:

• Client connection handshake: When a client submits a job via the RM IPC port (8032), the Application Manager accepts the connection, authenticates the user (via Kerberos tickets or delegation tokens), and validates the submission parameters (valid queue name, sufficient resource request within queue limits, correct CLC format).
• Initial container negotiation: The Application Manager negotiates with the Scheduler to obtain the first container on a slave node for the Application Master process. This negotiation includes specifying the AM's memory and VCore requirements and any placement preferences.
• AppMaster lifecycle monitoring: After the AM is launched, the Application Manager monitors its health via periodic registration heartbeats. If the AM's host machine fails (hardware crash, network partition, or OS panic), the Application Manager detects the missed heartbeat, marks the AM as lost, and requests a new container from the Scheduler to restart the AM on a different healthy node. This is a form of per-application high availability — each application's AM can fail and recover independently without affecting other applications on the cluster.
• Application state management: The Application Manager maintains the state machine for each application: NEW → SUBMITTED → ACCEPTED → RUNNING → FINISHED (or FAILED / KILLED). This state is exposed via the RM Web UI on port 8088.

Scheduler. The Scheduler is the pure resource allocation engine. Its sole job is to divide the cluster's total memory and VCore capacity among competing applications and tenant queues. Key design properties:

• No task monitoring: The Scheduler does not track individual task progress, restart failed tasks, or collect execution counters. Those responsibilities belong to the Application Master. This strict separation ensures the Scheduler's CPU loops remain dedicated to allocation decisions, allowing it to handle tens of thousands of concurrent container requests per second.
• Pluggable policies: The Scheduler supports swappable scheduling policies — FIFO (simple queue), Fair Scheduler (equal sharing with dynamic rebalancing), and Capacity Scheduler (hard SLA partitions). The administrator selects the policy at cluster configuration time, and the Scheduler enforces it.
• Delay scheduling optimization: When the Scheduler cannot satisfy a node-local request immediately (because the target node is busy), it can hold the request briefly (typically 1–2 seconds) rather than immediately assigning an off-rack container. This small delay dramatically increases the fraction of node-local allocations, reducing network I/O and improving overall job throughput. Studies show that even a 2-second delay can improve locality from 50% to over 90%.
• Capacity accounting: The Scheduler maintains real-time tallies of allocated and available resources per node, per rack, and per queue. Each NM heartbeat updates these tallies, making every heartbeat a scheduling opportunity.

5.3.3 Asynchronous Execution, Multi-Task Convergence, and Job Configuration

In real-world big data workflows, tasks running inside containers execute asynchronously. Not all mappers start at the same time, and not all mappers finish at the same time — the duration of each task depends on the size and characteristics of its input split, the computational complexity of the logic, and the hardware speed of the host node.

Coordinating asynchronous tasks into a unified job outcome requires explicit convergence logic:

• Mapper-to-Reducer Convergence: Asynchronous mapper outputs must be collected, sorted, and shuffled before reducer execution begins. The shuffle phase transfers intermediate key-value pairs from mapper output directories (stored on local disk) across the network to the appropriate reducers. Each reducer pulls data from all mappers, sorts it by key, and groups values by key for the reduce function.
• Job Configuration (JobConf): Written in Java or specified via configuration XML, JobConf defines the exact orchestration parameters:
• Number of mapper tasks and reducer tasks (job.setNumReduceTasks(N)).
• Custom Mapper and Reducer class bindings.
• Convergence rules: Specifying whether a single reducer is required (global aggregation) or multiple parallel reducers are permitted (partitioned aggregation).

5.3.4 YARN Administration Ports and Resource Manager Web UI Console

Cluster administrators monitor YARN performance using command-line tools and web management consoles listening on standard HTTP and IPC ports:

Using the RM Web UI (http://master:8088). The RM Web UI provides several critical views for administrators:

• Cluster Overview: Total memory, allocated memory, total VCores, allocated VCores, active nodes, decommissioned nodes, and lost nodes.
• Applications Tab: Lists all running, pending, finished, failed, and killed applications with their application ID, user name, queue name, start time, elapsed time, and final status.
• Scheduler Tab: Visualizes the queue hierarchy, per-queue resource allocation percentages, pending applications per queue, and active container counts.
• Node Status: Per-node breakdown of available memory, VCores, running containers, and health status.

5.3.5 Live Troubleshooting Scenario: YARN Connection Failures during Tool Execution

ERROR tool.ImportTool: Import failed: java.net.ConnectException: 
Call From master_host/127.0.0.1 to localhost:8032 failed on connection exception: 
java.net.ConnectException: Connection refused

   start-yarn.sh

jps   # Should show ResourceManager and NodeManager in addition to NameNode/DataNode

5.3.6 Application Lifespan, Scalability, and Availability

YARN applications are not all short-lived batch jobs. The framework supports a spectrum of application lifespans:

• Short-lived applications: A single MapReduce job or Sqoop import that runs for seconds to minutes, then terminates and releases all containers.
• Per-session applications: A Spark shell or Tez session that starts a set of executors and reuses them across multiple interactive queries. The session may last hours, holding containers even between queries.
• Long-running applications: A shared proxy service (such as a Spark Streaming job or a HiveServer2 instance) that runs indefinitely and acts as a gateway for multiple users. These applications must handle their own internal multi-tenancy.

Scalability design targets. YARN was designed to manage clusters of up to 10,000 nodes and schedule 100,000 concurrent tasks. Achieving this requires the Scheduler to remain decoupled from task-level monitoring (handled by per-application AMs) and the NM heartbeat protocol to be lightweight (each heartbeat carries a compact summary of node state, not individual task details).

High availability — divide and conquer. YARN achieves availability through two independent mechanisms:

• Resource Manager HA: An active/standby RM pair with automatic failover. If the active RM crashes, the standby RM takes over within seconds using shared state stored in ZooKeeper.
• Application Master HA: Each application's AM can be restarted independently on a different node if its host fails. The AM checkpoint mechanism saves recovery metadata so the new AM instance can resume from the last checkpoint rather than restarting from scratch.

5.3.7 Student Questions and Answers

5.4 YARN Scheduling Algorithms, Multi-Tenancy Quotas, and Resource Allocation

5.4.1 FIFO Scheduler Mechanics and Multi-Tenant Limitations

Operational Logic:

1. A job is submitted and appended to the tail of the single FIFO queue.
2. The Resource Manager examines the job at the head of the queue.
3. If the cluster has sufficient resources, the head job receives containers across all available nodes.
4. The job monopolizes the cluster — no other job receives any resources until the head job finishes.
5. Once the head job completes, the next job in the queue becomes the new head and the cycle repeats.

The "Small Job Blocked Behind Large Job" Problem:

Consider this real-world scenario that illustrates why FIFO is unsuitable for shared production clusters:

The critical insight is that Job B required only 10 seconds of compute time, yet it experienced a 1,440x latency amplification (4 hours vs. 10 seconds) purely because of its position in the queue. In a production environment, this behavior is unacceptable — executive stakeholders expect sub-second SLA responses, not 4-hour waits.

Why FIFO Exists Despite Its Limitations:

• Simplicity: Zero configuration overhead — no weights, no queues, no policies to tune.
• Deterministic ordering: Jobs execute in exactly the order they arrive, making debugging and auditing straightforward.
• Single-user clusters: For a single data scientist running jobs sequentially on a personal development cluster, FIFO is perfectly adequate since there are no competing tenants.
• Historical context: FIFO was the default scheduler in early Hadoop 1.x (MapReduce v1) before multi-tenancy requirements drove the development of Fair and Capacity schedulers.

5.4.2 Fair Scheduler Mechanics, Dynamic Rebalancing, and Cloudera Defaults

Dynamic Rebalancing in Detail:

The Fair Scheduler operates on a principle of continuous reallocation. Here is how it works step by step:

1. Single-job scenario: When only one job (Job A) is running, it receives 100% of the cluster — the Fair Scheduler does not artificially limit a lone job.
2. Second job arrives: When Job B is submitted, the scheduler detects that two jobs now share the cluster. It identifies containers belonging to Job A that are running on TaskTrackers where Job B also needs slots, and preempts (kills) some of Job A's containers. Those freed resources are immediately reallocated to Job B. The result: each job now holds approximately 50% of cluster resources.
3. Third job arrives: The process repeats. Resources are further divided so each job holds roughly 33.3% of the cluster.
4. Job completes: When Job A finishes, its containers are released. The scheduler immediately redistributes those resources to the remaining jobs (B and C), bringing each up to approximately 50% again.

This dynamic rebalancing means that a short job submitted mid-way through a long-running batch will still complete in a reasonable timeframe — it doesn't have to wait for the long job to finish.

Idle Capacity Borrowing: When a department's queue has no active jobs (e.g., the Marketing team runs nothing on weekends), the Fair Scheduler temporarily lends Marketing's unused capacity to other active queues (e.g., Data Science). When Marketing submits a new job on Monday morning, the scheduler reclaims borrowed capacity from other queues, gradually returning Marketing to its configured fair share. This happens transparently — no manual intervention is required.

Fair Scheduler Policy Options: The Fair Scheduler supports multiple allocation policies that can be configured on a per-queue basis:

• Fair sharing (default): Each application gets an equal share of resources.
• FIFO within a queue: Jobs inside a single queue execute in arrival order, but different queues still share fairly across each other.
• Dominant Resource Fairness (DRF): When clusters have multiple resource dimensions (CPU + memory), DRF ensures fairness across both dimensions simultaneously (see Section 5.4.7).

Preemption Support: The Fair Scheduler can proactively kill containers belonging to queues that are consuming more than their fair share, freeing resources for starved queues. This is particularly important for ensuring SLA compliance — a low-priority batch job holding excess resources will have its containers killed in favor of a latency-sensitive interactive query that has fallen below its fair share.

5.4.3 Multi-Tenant Queue Hierarchies, Weights, and Quota Mathematical Formulations

Mathematical Formulation of Queue Resource Percentages:

Let \(Q = \{q_1, q_2, \dots, q_K\}\) be the set of top-level tenant queues, where each queue \(q_k\) is assigned a configurable relative integer weight \(w_k > 0\). The percentage share of total cluster resources \(R(q_k)\) allocated to queue \(q_k\) is defined as:

\[ R(q_k) = \frac{w_k}{\sum_{j=1}^{K} w_j} \times 100\% \]

In words: the resource percentage of a specific queue equals its individual weight divided by the sum of all queue weights, multiplied by one hundred percent.

Sub-Queue Hierarchy Breakdown:

Queues can contain nested sub-queues, forming a tree structure that mirrors organizational hierarchies. This is configured using parent-child relationships in the Fair Scheduler XML configuration.

Consider the Sales queue (15% total cluster share) split into two regional sub-queues:

Sub-queue share calculation: \[ R(\text{NA} \mid \text{Sales}) = \frac{w_{\text{NA}}}{w_{\text{NA}} + w_{\text{EU}}} = \frac{30}{30 + 30} = 50\% \quad \text{of Sales capacity} \] \[ R(\text{NA} \mid \text{cluster}) = 50\% \times 15\% = 7.5\% \]

\[ R(\text{EU} \mid \text{Sales}) = \frac{w_{\text{EU}}}{w_{\text{NA}} + w_{\text{EU}}} = \frac{30}{30 + 30} = 50\% \quad \text{of Sales capacity} \] \[ R(\text{EU} \mid \text{cluster}) = 50\% \times 15\% = 7.5\% \]

The hierarchy means sub-queue weights are always relative to their siblings within the same parent, and the parent's share is relative to its own siblings. This recursive structure allows arbitrarily deep organizational modeling.

5.4.4 Zero-Weight Queues and Best Effort Jobs

Configuration Example:

Under the Data Science queue, a sub-queue named best_effort_jobs is configured with a weight of zero:

<queue name="best_effort_jobs">
  <weight>0.0</weight>
</queue>

Operational Meaning of Zero Weight:

When weighted queues are busy, the zero-weight queue is starved — it literally receives 0% of the cluster. This is because: \[ R(\text{best\_effort}) = \frac{0}{\sum w_j} \times 100\% = 0\% \]

The zero-weight queue only receives resources when the denominator's active queues shrink to zero (i.e., all weighted queues become idle). At that point, the scheduler has surplus capacity with no competing demand, and the best-effort jobs fill the gap.

The CSS Flexbox Analogy:

The zero-weight pattern is directly analogous to CSS Flexbox's flex-grow property:

In CSS, an element with flex-grow: 0 does not expand to fill available space — it only takes its minimum content size. Similarly, a YARN queue with weight = 0 receives no share of the cluster's capacity. But just as a flex-grow: 0 element still occupies its intrinsic size when it is the only element in the container, a zero-weight YARN queue will receive the entire cluster when it is the only queue with active jobs.

This same design pattern appears in operating systems (background/low-priority process scheduling), cloud batch processing (AWS Batch job queue priorities), and Android XML layout weights (layout_weight = 0).

5.4.5 Resource Guarantees and Constraints: Min Resources and Max Resources

Minimum Resources (minResources):

• Configures a guaranteed floor of memory and VCores for a queue (e.g., minResources = 65% for Data Science).
• Even under extreme cluster load, YARN guarantees this baseline capacity will be available when a job is submitted to that queue.
• If the Data Science queue has minResources = 65%, then even if 10 other queues are fighting for resources, Data Science will always receive at least 65% of the cluster when it has active jobs.
• Think of it as a service-level guarantee — the queue's contractual minimum.

Maximum Resources (maxResources):

• Configures a hard ceiling capping maximum resource consumption (e.g., maxResources = 85% for Data Science).
• Even if all other tenant queues are completely idle, jobs in the Data Science queue cannot consume more than 85% of total cluster capacity.
• This reserves a 15% safety buffer for incoming interactive queries from other departments, ensuring they can launch immediately without waiting for the scheduler to rebalance.
• Think of it as a fairness constraint — no single queue can hog the entire cluster.

Why Both Are Needed:

5.4.6 Capacity Scheduler Overview and Hortonworks Defaults

Key Properties of the Capacity Scheduler:

1. Hard-Partitioned Queues with Guaranteed Capacity:

Cluster capacity is divided into sub-queues with strict minimum capacity guarantees. Each organization or department receives a dedicated, guaranteed fraction of the cluster. Unlike the Fair Scheduler where shares are fluid, Capacity Scheduler queues have hard minimums that other queues cannot encroach upon.

1. Queue Elasticity (Borrowing Unused Capacity):

While each queue has a guaranteed minimum, unused capacity within a queue can be temporarily borrowed by other active queues. If the Engineering queue's guaranteed 30% is only 60% utilized (using 18% of total cluster), the remaining 12% can be borrowed by other queues that need it. When Engineering's load increases, it reclaims its full guaranteed capacity.

1. Maximum Capacity Limits:

Each queue can be configured with a maximum-capacity parameter that caps how much borrowed capacity it can consume. This prevents any single queue from monopolizing the cluster even during idle periods.

1. FIFO Within Each Queue:

While different queues share capacity via the Capacity Scheduler, jobs within a single queue execute in strict FIFO order. This provides predictable ordering within a department while allowing resource sharing across departments.

1. Preemption:

The Capacity Scheduler supports preemption of containers when a queue needs to reclaim its guaranteed capacity. If Queue A's containers are running on resources borrowed from Queue B, and Queue B suddenly needs its guaranteed minimum back, the scheduler can kill Queue A's excess containers to free resources for Queue B.

1. Hierarchical Queues:

Queues can be nested to any depth, mirroring corporate organizational structures (division → department → team).

Capacity Scheduler Configuration Example:

A typical enterprise configuration might define a two-level hierarchy:

root
├── prod          (40% capacity guarantee)
└── dev           (60% capacity guarantee, max-capacity: 75%)
    ├── eng       (50% of dev = 30% of total)
    └── science   (50% of dev = 30% of total)

• prod queue: Guaranteed 40% of the cluster. Reserved for production ETL pipelines and SLA-critical workloads. Even if the dev queue is very busy, prod always gets at least 40%.
• dev queue: Guaranteed 60% of the cluster. Its max-capacity is set to 75%, meaning it can borrow up to 15% extra from prod (when prod is idle), but can never exceed 75% total. This ensures prod always retains at least 25% for burst production workloads.
• eng sub-queue: Receives 50% of the dev queue's capacity (which is 30% of the total cluster).
• science sub-queue: Receives the other 50% of the dev queue's capacity (30% of the total cluster).

The max-capacity on dev is the key safety valve: without it, if prod has no running jobs, dev could consume 100% of the cluster, and a sudden production job would have to wait for the scheduler to reclaim resources. With max-capacity = 75%, at least 25% is always available for prod bursts.

5.4.7 Dominant Resource Fairness (DRF) — Multi-Dimensional Resource Fairness

Why Single-Dimension Fairness Fails:

Consider a cluster with 100 CPUs and 10 TB of memory. Two applications run:

• App A (e.g., a data loading job): Requests 2 CPUs and 300 GB memory per container.
• App B (e.g., a machine learning training job): Requests 6 CPUs and 100 GB memory per container.

If we only look at CPU fairness, App B uses 3x more CPU per container, so we might give App A more containers. If we only look at memory fairness, App A uses 3x more memory per container. DRF resolves this by computing each application's dominant share.

DRF Calculation:

For each application, compute the fraction of each resource type it consumes per container:

App A's dominant resource is memory (3%) — memory is the resource dimension where App A consumes the largest fraction of the cluster's total.

App B's dominant resource is CPU (6%) — CPU is the resource dimension where App B consumes the largest fraction of the cluster's total.

DRF aims to equalize each application's dominant share. If App A gets \(n_A\) containers and App B gets \(n_B\) containers:

• App A's dominant share: \(n_A \times 3\%\)
• App B's dominant share: \(n_B \times 6\%\)

Setting them equal: \(n_A \times 3\% = n_B \times 6\% \implies n_A = 2 \times n_B\)

Result: App A gets twice as many containers as App B. In terms of actual resource consumption:

• App A: many containers × (2 CPU, 300 GB each) — it's a memory-heavy workload that gets lots of containers because each one uses little CPU.
• App B: half as many containers × (6 CPU, 100 GB each) — it's a CPU-heavy workload that gets fewer containers because each one demands significant CPU.

This ensures neither application dominates the cluster on any single resource dimension, which is the hallmark of true multi-resource fairness.

5.4.8 Comparison of YARN Schedulers

5.4.9 Student Questions and Answers

5.4.10 Key Takeaways

5.5 Sqoop Relational Data Ingestion Framework

5.5.1 Purpose, Architectural Overview, and Name Etymology

Architectural Execution Paradigm:

Although Sqoop is invoked as a command-line interface (CLI) tool, it does not execute data transfers through a single client process. Instead, it follows a four-stage pipeline that compiles a user's CLI arguments into a full distributed MapReduce job:

\[ \text{Sqoop CLI Command} \;\longrightarrow\; \text{Java Code Generation} \;\longrightarrow\; \text{Compile JAR} \;\longrightarrow\; \text{MapReduce Job on YARN} \]

Stage 1 — CLI Parsing: Sqoop reads connection parameters (--connect, --username, --table, --num-mappers, etc.) and validates them.

Stage 2 — Java Code Generation: Sqoop connects to the target database via JDBC, retrieves the table metadata (column names, SQL types, primary key), and generates a table-specific Java class that implements the DBWritable interface. This interface provides two critical serialization methods:

• readFields(ResultSet __dbResults) — populates Java fields from a JDBC ResultSet row (deserialization for import)
• write(PreparedStatement __dbStmt) — inserts Java fields into a PreparedStatement (serialization for export)

The generated class also implements Hadoop's Writable interface, allowing instances to be serialized across the network during MapReduce execution and stored in SequenceFiles.

Stage 3 — JAR Compilation: Sqoop compiles the generated Java source file into a JAR (.jar) artifact using the Java compiler bundled with the JDK.

Stage 4 — MapReduce Submission on YARN: Sqoop submits the compiled JAR as a Map-only MapReduce job to the YARN Resource Manager. Crucially, there are zero Reducer tasks by default—each mapper opens an independent JDBC socket connection to the database, reads a partition of rows defined by SQL WHERE clauses, and streams the data directly into HDFS block files.

Sqoop uses Hadoop's DataDrivenDBInputFormat as the InputFormat for the import job. This class is responsible for splitting the source table into partitions and assigning each partition to a mapper. The partitioning is based on a "splitting column" (typically the primary key) and uses the column's MIN and MAX values to construct bounding queries.

Sqoop Connectors:

Sqoop ships with built-in connectors for the most popular databases:

Third-party connectors (e.g., Teradata, Couchbase) extend Sqoop to NoSQL stores and enterprise data warehouses. The connector architecture is modular—each connector knows how to translate SQL metadata into Hadoop-compatible types and can optionally provide optimized bulk-transfer strategies.

5.5.2 Core Command Syntax and Parameter Reference

sqoop import \
  --connect jdbc:mysql://localhost:3306/retail_db \
  --username root \
  --password-file /user/hadoop/credentials/mysql.pwd \
  --table customer_profiles \
  --target-dir /user/hadoop/data/customer_profiles \
  --where "signup_date >= '2023-01-01'" \
  --num-mappers 4 \
  --split-by customer_id

Key Sqoop CLI Command Parameters:

5.5.3 Parallel Import Mechanics, Splitting Logic, and Primary Key Pitfalls

Understanding how Sqoop parallelizes data ingestion across --num-mappers N is critical for both performance tuning and preventing runtime job failures. Sqoop does not stream the entire table through a single socket—it divides the row range into equal slices and assigns each slice to an independent mapper.

Partition Splitting Algorithm — Worked Example:

ERROR tool.ImportTool: Error during import: No primary key could be found
for table <table_name>. Please specify one with --split-by or use
--num-mappers 1.

General Step Size Formula:

\[ \text{Step Size} = \frac{\text{MAX} - \text{MIN} + 1}{\text{num-mappers}} \]

where:

• \(\text{MIN}\) = minimum value of the splitting column (retrieved via SELECT MIN(col))
• \(\text{MAX}\) = maximum value of the splitting column (retrieved via SELECT MAX(col))
• \(\text{num-mappers}\) = the value of the --num-mappers (or -m) parameter

When the range is not evenly divisible by the number of mappers, the final mapper's range absorbs the remainder, ensuring no rows are lost.

5.5.4 Directory Management and Target Overwrite Rules

ERROR tool.ImportTool: Import directory /user/hadoop/data/customer_profiles already exists.

sqoop import --connect jdbc:mysql://localhost:3306/retail_db \
  --table orders --target-dir /user/hadoop/data/orders \
  --delete-target-dir --num-mappers 4

sqoop import --connect jdbc:mysql://localhost:3306/retail_db \
  --table orders --target-dir /user/hadoop/data/orders \
  --append --num-mappers 4 --where "order_date > '2023-06-01'"

5.5.5 Output Data File Formats

5.5.6 Live Demo Walkthrough: 220,000 Record MySQL Import

sqoop import \
  --connect jdbc:mysql://localhost:3306/profile_db \
  --username root \
  --password root \
  --table profiles \
  --target-dir /user/hadoop/sqoop_demo \
  --num-mappers 4

INFO tool.ImportTool: Beginning import...
INFO manager.SqlManager: Executing SQL statement: SELECT t.* FROM profiles AS t LIMIT 1
INFO manager.SqlManager: Executing SQL statement: SELECT t.* FROM profiles AS t WHERE 1=0
INFO orm.CompilationManager: Writing jar file: /tmp/sqoop-hadoop/compile/...
INFO mapreduce.ImportJobBase: Beginning import of profiles
INFO impl.YarnClientImpl: Submitted application application_...
INFO mapreduce.Job: Running job: job_...
INFO mapreduce.Job:  map 0% reduce 0%
INFO mapreduce.Job:  map 25% reduce 0%
INFO mapreduce.Job:  map 50% reduce 0%
INFO mapreduce.Job:  map 75% reduce 0%
INFO mapreduce.Job:  map 100% reduce 0%
INFO mapreduce.Job: Job job_... completed successfully
INFO mapreduce.ImportJobBase: Retrieved 220000 records.

hdfs dfs -ls /user/hadoop/sqoop_demo

-rw-r--r--   1 hadoop hadoop   18456789   /user/hadoop/sqoop_demo/part-m-00000
-rw-r--r--   1 hadoop hadoop   18523456   /user/hadoop/sqoop_demo/part-m-00001
-rw-r--r--   1 hadoop hadoop   18398123   /user/hadoop/sqoop_demo/part-m-00002
-rw-r--r--   1 hadoop hadoop   18412567   /user/hadoop/sqoop_demo/part-m-00003

hdfs dfs -cat /user/hadoop/sqoop_demo/part-m-00000 | head -5

1,john_doe,john@example.com,2023-01-15,active
2,jane_smith,jane@example.com,2023-02-20,active
3,bob_jones,bob@example.com,2023-03-10,inactive
4,alice_wang,alice@example.com,2023-04-05,active
5,charlie_brown,charlie@example.com,2023-05-18,active

Direct-Mode Import Alternative:

For MySQL specifically, the same import could be performed in direct mode:

sqoop import \
  --connect jdbc:mysql://localhost:3306/profile_db \
  --username root \
  --password root \
  --table profiles \
  --target-dir /user/hadoop/sqoop_demo \
  --direct \
  --num-mappers 4

In direct mode, Sqoop bypasses JDBC and instead spawns instances of mysqldump for each mapper, reading data from its output stream. This is typically much faster for MySQL but cannot handle CLOB or BLOB columns, and metadata is still queried through JDBC.

Hive Integration Alternative:

To import directly into a Hive table (bypassing the manual CREATE TABLE + LOAD DATA steps):

sqoop import \
  --connect jdbc:mysql://localhost:3306/profile_db \
  --username root \
  --password root \
  --table profiles \
  --hive-import \
  --num-mappers 4

This single command imports the data into HDFS and then auto-creates a Hive table definition based on the source database schema, loading the data directly into Hive's warehouse directory. Note that Hive's type system is less rich than most SQL systems, so some type narrowing (e.g., DATE → STRING) may occur with a warning.

5.5.7 Sqoop Exports: HDFS Back to RDBMS

Sqoop is not limited to imports—it can also export data from HDFS back into a relational database. The export process mirrors the import architecture: a MapReduce job reads HDFS files (text or SequenceFile format), parses records using a generated Java class, and builds batch INSERT statements to write rows into the target table.

sqoop export \
  --connect jdbc:mysql://localhost:3306/analytics_db \
  --username root \
  --password root \
  --table sales_by_zip \
  --export-dir /user/hive/warehouse/zip_profits \
  --input-fields-terminated-by '\0001'

Key export considerations:

• The target table must already exist in the database—Sqoop does not auto-create it during export.
• Column order in the target table must match the order of fields in the HDFS files.
• For Hive-sourced data, the default field delimiter is Ctrl-A (\0001), specified via --input-fields-terminated-by.
• Exports are not atomic—parallel mappers commit independently, so partial results may be visible before all mappers complete. Use --staging-table for transactional guarantees (data is written to a staging table first, then moved atomically to the destination).
• Direct-mode exports are also supported for MySQL (via mysqlimport) and PostgreSQL.

5.5.8 Student Questions and Answers

5.6 Flume: Unstructured Streaming Log Ingestion Framework

5.6.1 Purpose, Origin, and Comparison with Sqoop

Comparative Analysis: Sqoop vs Flume

5.6.2 Flume Agent Architecture: Source, Channel, and Sink

[External Event Generator]  --->  (Source)  --->  [Channel]  --->  (Sink)  --->  [External Destination]
     (web server,                                                    (HDFS, HBase,
      application,                                                    Solr, another
      sensor, etc.)                                                    Flume agent)

The Flume Event

Every piece of data flowing through Flume is represented as a Flume Event object. An event has exactly two parts:

1. Body: A byte array (byte[]) containing the raw payload data. For a web server log line, this would be the UTF-8 encoded text of the log entry. For audio data, it would be the raw audio bytes. Flume treats the body as opaque — it does not parse or interpret the contents.
2. Headers: A Map<String, String> of optional key-value string attributes. Headers carry metadata about the event (such as timestamps, hostnames, severity levels, or custom tags) and are used by interceptors and multiplexing selectors for routing decisions. Headers are not part of the payload and can be added, modified, or removed as the event flows through the agent.

Component 1: Source

The Source is the entry point of a Flume agent. It is responsible for receiving or actively pulling data from an external event generator, wrapping raw records into standardized Flume Event objects, and pushing those events into one or more connected Channels.

Component 2: Channel

The Channel is the passive buffer that sits between the Source and the Sink. It holds Flume Events in a queue until the Sink consumes and delivers them. Channels are transactional bridges: an event is not removed from the Channel until the Sink successfully confirms delivery to the destination. This ensures that if the Sink fails mid-delivery, the event remains in the Channel for retry.

Importantly, a Source can feed multiple Channels simultaneously (enabling fan-out), but a Sink is fed by exactly one Channel. However, multiple Sinks can consume from the same Channel (covered by Sink Groups in Section 5.6.4).

Component 3: Sink

The Sink removes events from the Channel and delivers them to an external repository or to another Flume agent's Source over the network.

5.6.3 Channel Storage Implementations: Memory Channel vs File Channel

Detailed Comparison: Memory Channel vs File Channel

Batching for Efficiency

Flume processes events in batches within each transaction rather than one at a time. The default batch size is 100 events per transaction (configurable per component). This is particularly important for the File Channel because every transaction requires a disk write and fsync call. Processing 100 events per transaction rather than 1 event per transaction dramatically reduces the number of disk syncs and improves throughput by orders of magnitude.

Similarly, the Spooling Directory Source reads files in batches of 100 lines (configurable via batchSize), and the Avro Sink attempts to send 100 events per RPC call to the downstream agent.

5.6.4 Advanced Topologies: Agent Chaining, Fan-Out, and Multiplexing

1. Agent Chaining (Multi-Hop Architecture)

Agent chaining connects multiple Flume agents in series, where each agent performs a stage of processing or aggregation. The link between agents is established using an Avro Sink on the upstream agent sending events over the network to an Avro Source on the downstream agent.

[Edge Servers]          [Aggregator Machine]           [HDFS]
                   
Edge Agent (Tier 1)     Aggregator Agent (Tier 2)
┌─────────────┐        ┌──────────────┐
│ SpoolDir    │        │ Avro Source   │
│   Source    │        │              │
│     ↓       │  RPC   │     ↓        │
│ File Channel├───────►│ File Channel │
│     ↓       │  :10000│     ↓        │
│ Avro Sink   │        │ HDFS Sink    │
└─────────────┘        └──────────────┘

How it works:

• Tier 1 (Edge Agents): Run on each web server or edge machine. They collect local log events using a Spooling Directory Source, buffer them in a File Channel, and forward them via an Avro Sink to a central aggregator.
• Tier 2 (Aggregator Agents): Run on a smaller set of powerful machines. They receive events from many Tier 1 agents through an Avro Source, buffer them in a File Channel, and write them to HDFS using an HDFS Sink.

Why this matters: Aggregation produces fewer, larger HDFS files (rather than many tiny files from individual edge machines), which dramatically improves MapReduce performance and reduces HDFS NameNode memory pressure. The Avro RPC connection between agents is transactional — the Avro Sink will not commit its Channel transaction until the downstream Avro Source's Channel transaction has committed, ensuring at-least-once delivery across the network hop.

Tier configuration detail: Each agent on the same machine must use distinct File Channel checkpoint and data directories to avoid filesystem collisions:

edge.channels.channel1.checkpointDir = /tmp/edge/file-channel/checkpoint
edge.channels.channel1.dataDirs = /tmp/edge/file-channel/data
aggregator.channels.channel2.checkpointDir = /tmp/aggregator/file-channel/checkpoint
aggregator.channels.channel2.dataDirs = /tmp/aggregator/file-channel/data

2. Fan-Out Topology (Replication)

Fan-out delivers events from a single Source into multiple Channels simultaneously, so each Channel feeds a different Sink and destination. This is the default behavior when a Source is connected to multiple Channels.

                    ┌──→ [File Channel]  →  [HDFS Sink]   → HDFS (batch analytics)
[SpoolDir Source] ──┤
                    └──→ [Memory Channel] → [Solr Sink]    → Solr (real-time search)

How it works: The Source replicates each incoming event into every connected Channel. Each Channel-Sink pair operates independently with its own transaction. If one Channel's transaction fails (e.g., the channel is full), the Source's transaction to that channel is rolled back and retried later.

Optional Channels: In a fan-out scenario, you can designate certain channels as optional using selector.optional. If an optional channel's transaction fails, it does not block the Source from committing events to the other (required) channels. This is useful when one destination is critical (HDFS via File Channel) and another is best-effort (Solr indexing via Memory Channel).

Real-world example — Near-Real-Time Indexing: A web log ingestion system fans out events to both an HDFS sink (main repository, required channel) and a MorphlineSolrSink (search index, optional channel). If Solr is temporarily unavailable, events continue flowing to HDFS without interruption.

3. Multiplexing (Header-Based Routing)

Multiplexing extends fan-out with selective routing: instead of replicating every event to every channel, a multiplexing selector inspects event header attributes and routes each event to a specific Channel based on configurable rules.

                    ┌──→ [Channel A]  →  [Alert Sink]     (events where level=ERROR)
[Syslog Source] ────┤
                    ├──→ [Channel B]  →  [HDFS Sink]      (events where level=INFO)
                    │
                    └──→ [Channel C]  →  [Archive Sink]   (events where level=DEBUG)

Configuration example:

edge.sources.source1.selector.type = multiplexing
edge.sources.source1.selector.header = level
edge.sources.source1.selector.mapping.ERROR = channelA
edge.sources.source1.selector.mapping.INFO = channelB
edge.sources.source1.selector.default = channelC

The multiplexing selector reads the level header from each event and routes it to the mapped channel. Events with unmapped header values are sent to the default channel. This pattern is essential for content-based routing where different event severities or types must be processed differently.

5.6.5 Flume Reliability Modes

1. Best Effort Mode

• Mechanism: Events are transmitted without transactional delivery guarantees. The Source pushes events into the Channel without waiting for commit confirmation, and the Sink delivers events to the destination without transactional rollback on failure.
• Failure Behavior: If a downstream network hop fails, a Channel queue overflows, or the agent crashes, buffered events are dropped silently with no retry.
• Configuration: Typically uses a Memory Channel (no disk persistence) with no retry logic.
• Use Case: High-volume telemetry, sensor data streams, or any scenario where approximate data is sufficient and maximum throughput is the priority.

2. Store-and-Forward Mode

• Mechanism: Combines disk-backed File Channels with the agent's local transaction system. Events are written to the File Channel (persisted to local disk with fsync) before the Source's transaction is committed. If the downstream destination (HDFS, another agent) is unavailable, events remain safely stored in the local File Channel until connectivity is restored, at which point the Sink resumes delivery.
• Failure Behavior: Events survive agent crashes, host reboots, and network partitions. On restart, the File Channel replays its checkpoint and data directories to restore all uncommitted events.
• Limitation: This mode guarantees delivery within a single agent or between two adjacent agents, but does not provide end-to-end acknowledgment from the original event source to the final destination.
• Use Case: Production log ingestion where zero data loss within the Flume infrastructure is required.

3. End-to-End Reliability Mode

• Mechanism: Implements explicit transactional handshakes across the entire chain from the initial event source to the final destination Sink. An event is acknowledged back to the generator only after the final Sink confirms a successful permanent write to the destination storage.
• Failure Behavior: If any component in the chain fails, the event remains at the last successfully committed stage and is retried until end-to-end delivery succeeds.
• Implementation: Achieved by chaining multiple agents with Avro Sink/Source pairs, where each hop uses transactional File Channels. The Avro Sink on Agent N will not commit its Channel transaction until it receives confirmation that Agent N+1's Channel transaction has committed. This cascading transactional guarantee propagates through every tier.
• Use Case: Mission-critical regulatory or audit data where at-least-once delivery from source to final destination must be guaranteed even across multiple network hops.

5.6.6 Student Questions and Answers

Exam Guidance Summary

MapReduce and Hive SQL Translation

• Practice converting declarative SQL queries containing GROUP BY, JOIN, and ORDER BY into multi-stage MapReduce execution diagrams. For each query, identify:
• Mapper output keys and intermediate key-value pairs
• Combiner pre-aggregations (and when they are safe vs. unsafe)
• Partitioned Reducers for parallel group aggregation
• Single global sorting Reducer for ORDER BY across all groups
• Know that Hive compiles HiveQL into a DAG of MapReduce jobs, and that the execution engine can be MapReduce, Tez, or Spark.

YARN Core Architecture and Execution Lifecycle

• Trace the complete 10-step YARN job submission sequence: Client → RM → NM1 → AppMaster → RM → NM2/NM3 → Worker Containers. Detail the interaction between Client, Resource Manager, Node Manager, Application Master, and worker Containers at each step.
• Explain why YARN was decoupled in Hadoop 2.0: JobTracker SPOF, 4,000-node / 40,000-task scalability ceiling, and inability to run non-MapReduce workloads.
• Memorize YARN network ports:
• 8088 — Resource Manager Web UI (HTTP console)
• 8032 — Resource Manager IPC / Admin (job submission port)
• 8030 — Resource Tracker (NM heartbeats to RM)
• 8042 — Node Manager Web UI

YARN Multi-Tenant Schedulers and Resource Calculations

• Solve multi-tenant queue resource percentage allocations using the relative weight formula:

\[ R(q_k) = \frac{w_k}{\sum_{j=1}^{K} w_j} \times 100\% \]

• Explain the operational function of:
• Zero-weight queues (weight = 0.0): zero guaranteed share, executes only when all higher-priority queues are idle
• minResources floors: guaranteed baseline under heavy cluster load
• maxResources ceilings: hard cap preventing monopolization
• Know the three scheduler types and when to use each:
• FIFO: simple, single-tenant only
• Fair: dynamic equal sharing, idle capacity borrowing, Cloudera default
• Capacity: hard SLA partitions, queue elasticity, Hortonworks default
• Self-Study Assignment: Study the Capacity Scheduler queue parameters and comparison matrix for upcoming quizzes and final examinations.

Sqoop Command Syntax and Splitting Logic

• Memorize key Sqoop CLI parameters: --connect, --table, --target-dir, --num-mappers, --split-by, --where, --password-file.
• Understand the parallel import splitting formula:

\[ \text{Step Size} = \frac{\text{MAX} - \text{MIN} + 1}{\text{num-mappers}} \]

• Trace mapper queries: given MIN, MAX, and num-mappers, write the exact WHERE clause for each mapper.
• Identify primary key failure modes: explain why multi-mapper Sqoop imports fail on tables without primary keys unless --split-by or --num-mappers 1 is specified.
• Know that Sqoop generates Map-only jobs (0 reducers) because raw ingestion needs no key aggregation.

Flume Streaming Component Architecture

• Diagram the Flume Agent architecture: Source → Channel → Sink.
• Compare Memory Channels (high throughput, volatile, events lost on crash) vs File Channels (disk-backed, zero data loss, store-and-forward).
• Differentiate Sqoop vs Flume:
• Sqoop: structured batch RDBMS import via MapReduce jobs on YARN
• Flume: unstructured real-time streaming log ingestion via long-running JVM agents
• Know the three reliability modes: Best Effort, End-to-End, Store-and-Forward.
• Understand agent chaining (multi-hop with Avro Sink/Source), fan-out (one source to multiple channels), and multiplexing (header-based routing).

Key Industry Applications

• Enterprise Data Lake Ingestion with Apache Sqoop — Financial institutions and retail corporations use Sqoop nightly batch workflows to extract transactional database tables from MySQL, Oracle, and DB2 instances into HDFS and Hive Data Lakes, enabling historical trend analytics across millions of customer records. For example, a bank might import all credit card transaction tables each night, then run Hive queries to detect anomalous spending patterns the next morning.

• Real-Time Web Server Log Aggregation with Apache Flume — Large-scale e-commerce platforms deploy Flume agents directly on hundreds of front-end web server nodes using Exec and Spooling Directory Sources to stream web access logs continuously into HDFS and Elasticsearch clusters for real-time fraud detection and search indexing. A typical deployment uses a two-tier agent topology: edge agents on web servers forward events via Avro Sink to central aggregator agents, which write to HDFS with time-partitioned directories.

• Multi-Tenant Enterprise Cluster Orchestration with YARN — Cloud data platforms utilize YARN Fair and Capacity Schedulers to partition multi-petabyte clusters across diverse corporate departments (Data Engineering, Data Science, Executive Analytics), ensuring critical batch ETL pipelines meet strict SLAs while granting interactive query access to data scientists. The Fair Scheduler's dynamic rebalancing ensures that a department's idle capacity is not wasted — it is temporarily borrowed by active queues and reclaimed when needed.