Apache Storm

Overview

Apache Storm is a distributed real-time computation system designed for processing unbounded streams of data with guaranteed message processing. Originally created at BackType by Nathan Marz and later acquired by Twitter in 2011, Storm became an Apache top-level project in 2014.

Storm is to real-time processing what Hadoop is to batch processing — it provides a simple, reliable way to process streams of data at massive scale. Storm processes millions of tuples per second per node with millisecond latency.

Key capabilities include:

• Real-time stream processing: True event-at-a-time processing with sub-second latency
• Guaranteed message processing: At-least-once semantics (exactly-once with Trident)
• Horizontal scalability: Add nodes to increase throughput linearly
• Fault tolerance: Automatic worker restart and task reassignment
• Language agnostic: Topologies can use any language via multi-lang protocol
• Simple programming model: Spouts (data sources) and Bolts (processing units) connected into topologies

Architecture

Cluster Architecture

┌─────────────────────────────────────────────────┐
│                    Nimbus                         │
│  (Master node - job distribution & monitoring)   │
└──────────────────────┬──────────────────────────┘
                       │
              ┌────────┴────────┐
              ▼                 ▼
      ┌──────────────┐  ┌──────────────┐
      │  ZooKeeper   │  │  ZooKeeper   │
      │   Ensemble   │  │   Ensemble   │
      └──────┬───────┘  └──────┬───────┘
             │                  │
    ┌────────┼──────────────────┼────────┐
    ▼        ▼                  ▼        ▼
┌────────┐ ┌────────┐    ┌────────┐ ┌────────┐
│Superv. │ │Superv. │    │Superv. │ │Superv. │
│  ┌───┐ │ │  ┌───┐ │    │  ┌───┐ │ │  ┌───┐ │
│  │W1 │ │ │  │W3 │ │    │  │W5 │ │ │  │W7 │ │
│  │W2 │ │ │  │W4 │ │    │  │W6 │ │ │  │W8 │ │
│  └───┘ │ │  └───┘ │    │  └───┘ │ │  └───┘ │
└────────┘ └────────┘    └────────┘ └────────┘
 Worker     Worker        Worker     Worker
  Node       Node          Node       Node

Core Components

1. Nimbus (Master)

• Central coordination daemon (similar to Hadoop’s JobTracker)
• Distributes code (topology JARs) to Supervisors
• Assigns tasks to workers across the cluster
• Monitors worker health and reassigns tasks on failure
• Stateless — all state stored in ZooKeeper (Nimbus can be restarted without affecting running topologies)

2. Supervisors (Worker Nodes)

• Run on each worker machine in the cluster
• Listen for work assigned by Nimbus via ZooKeeper
• Start and stop worker processes as needed
• Each Supervisor manages multiple worker processes

3. Worker Processes

• JVM processes spawned by Supervisors
• Each worker runs a subset of a topology
• Contains multiple executors (threads)
• Each executor runs one or more tasks (instances of spouts/bolts)

4. ZooKeeper

• Coordinates between Nimbus and Supervisors
• Stores cluster state, task assignments, and heartbeat data
• Enables Nimbus to be stateless and restartable

Component Hierarchy

Cluster
  └── Nimbus (1 master)
  └── Supervisors (N worker machines)
        └── Worker Processes (M per supervisor)
              └── Executors (threads)
                    └── Tasks (spout/bolt instances)

Programming Model

Topologies

A topology is a graph of computation — a directed graph where nodes are spouts or bolts and edges are streams of tuples. Topologies run indefinitely until killed.

TopologyBuilder builder = new TopologyBuilder();

// Set spout with parallelism of 2
builder.setSpout("sentences", new SentenceSpout(), 2);

// Set bolts with parallelism and grouping
builder.setBolt("split", new SplitBolt(), 4)
       .shuffleGrouping("sentences");
builder.setBolt("count", new CountBolt(), 6)
       .fieldsGrouping("split", new Fields("word"));

// Submit topology
Config conf = new Config();
conf.setNumWorkers(3);
StormSubmitter.submitTopology("word-count", conf, builder.createTopology());

Spouts

Spouts are the data source components — they read data from external sources and emit tuples into the topology.

public class SentenceSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;

    @Override
    public void open(Map config, TopologyContext context,
                     SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        // Emit a tuple with a message ID for reliability
        collector.emit(new Values("the cow jumped over the moon"),
                       UUID.randomUUID().toString());
    }

    @Override
    public void ack(Object msgId) {
        // Called when tuple is fully processed
    }

    @Override
    public void fail(Object msgId) {
        // Called when tuple fails — replay it
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
    }
}

Common spout sources: Kafka, RabbitMQ, Kinesis, Twitter API, file systems, databases.

Bolts

Bolts process input tuples and optionally emit new tuples. They can filter, aggregate, join, interact with databases, or call external APIs.

public class SplitBolt extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map config, TopologyContext context,
                        OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String sentence = input.getStringByField("sentence");
        for (String word : sentence.split("\\s+")) {
            collector.emit(input, new Values(word));
        }
        collector.ack(input);  // Acknowledge processing
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }
}

Tuples and Streams

• Tuple: The main data structure — a named list of values (any serializable type)
• Stream: An unbounded sequence of tuples. Each spout/bolt can emit multiple named streams
• Default stream: Unnamed stream used when no stream ID is specified

Stream Groupings

Stream groupings define how tuples are routed between components:

Choosing the Right Grouping

• Use fields grouping when you need all tuples with the same key to go to the same bolt (e.g., word count)
• Use shuffle grouping for stateless operations to maximize parallelism
• Use all grouping sparingly — it multiplies traffic by the number of bolt tasks

Reliability & Message Guarantees

At-Least-Once Processing

Storm tracks the processing of each tuple through the topology using an acker system:

1. Spout emits a tuple with a unique message ID
2. Each bolt anchors new tuples to the input tuple and acks when done
3. The acker uses XOR of tuple IDs to efficiently track the entire tuple tree
4. If all tuples in the tree are acked, the spout receives an ack callback
5. If any tuple times out (default 30 seconds), the spout receives a fail callback and can replay

// In a bolt — anchoring ensures reliability
collector.emit(inputTuple, new Values(newData));  // anchored
collector.ack(inputTuple);  // acknowledge processing

Exactly-Once with Trident

Trident is a higher-level abstraction built on Storm that provides:

• Exactly-once processing via micro-batching and transactional state
• Stateful operations: aggregations, joins, grouping with persistent state
• Batch processing: Groups tuples into small batches for efficient state updates

TridentTopology topology = new TridentTopology();
topology.newStream("sentences", new SentenceSpout())
    .each(new Fields("sentence"), new SplitFunction(), new Fields("word"))
    .groupBy(new Fields("word"))
    .persistentAggregate(new MemoryMapState.Factory(), new Count(),
                         new Fields("count"));

Parallelism

Storm’s parallelism is controlled at three levels:

// 3 workers, 4 executor threads for split bolt, 8 tasks
Config conf = new Config();
conf.setNumWorkers(3);
builder.setBolt("split", new SplitBolt(), 4)  // 4 executors
       .setNumTasks(8)                          // 8 tasks
       .shuffleGrouping("sentences");

Setting tasks > executors allows you to increase parallelism later (via rebalance) without restarting the topology.

Dynamic Rebalancing

# Increase parallelism without restarting
storm rebalance my-topology -n 5 -e split=8 -e count=12

Fault Tolerance

Worker Failure

1. Supervisor detects worker process death
2. Supervisor restarts the worker
3. If restart keeps failing, Nimbus reassigns tasks to other workers

Supervisor Failure

1. Nimbus detects missing heartbeats via ZooKeeper
2. Tasks from failed supervisor are reassigned to healthy supervisors
3. Supervisor can be restarted — it will pick up new assignments

Nimbus Failure

• Running topologies continue unaffected (workers keep processing)
• New topologies cannot be submitted, and failed workers won’t be reassigned
• Storm HA: Configure multiple Nimbus instances with leader election via ZooKeeper

Performance Characteristics

Tuning Tips

• Increase parallelism for bottleneck components (check Storm UI metrics)
• Use local or shuffle grouping to reduce network hops
• Tune topology.max.spout.pending to control backpressure
• Batch external writes in bolts using tick tuples
• Use Kryo serialization instead of Java serialization
• Size workers appropriately: 1-2 GB heap per worker is common

Storm vs Other Stream Processors

When to choose Storm:

• Need true sub-millisecond latency
• Simple event processing pipelines
• Existing Storm infrastructure
• Language-agnostic processing (multi-lang protocol)

When to choose alternatives:

• Need complex windowing or SQL → Flink
• Need unified batch + stream → Spark or Flink
• Need managed state and exactly-once → Flink

Common Use Cases

1. Real-time analytics: Click stream analysis, trending topics, live dashboards
2. Continuous ETL: Transform and load streaming data into data warehouses
3. Online machine learning: Real-time feature computation and model scoring
4. Monitoring and alerting: Infrastructure monitoring, anomaly detection
5. Distributed RPC: Parallelize computation of complex functions

Interview Questions

Conceptual

1. What is a topology in Storm?

   • A topology is a DAG (directed acyclic graph) of spouts and bolts connected by streams. It defines the computation logic and runs indefinitely until killed. It’s analogous to a MapReduce job but for real-time processing.

2. Explain the difference between spouts and bolts.

   • Spouts are data source components that read from external systems (Kafka, queues) and emit tuples. Bolts are processing components that receive tuples, perform computation (filter, aggregate, transform), and optionally emit new tuples.

3. How does Storm guarantee message processing?

   • Through the acker mechanism: each tuple is tracked using XOR of tuple IDs. Spouts assign message IDs, bolts anchor and ack tuples. If the entire tuple tree is acked, the message succeeds. If any tuple times out, the spout’s fail() is called for replay.

4. What is the difference between at-least-once and exactly-once in Storm?

   • At-least-once: Core Storm guarantees every tuple is processed at least once (may replay on failure). Exactly-once: Trident provides this via micro-batching and transactional state updates, ensuring each tuple affects state exactly once.

5. What are stream groupings and why do they matter?

   • Stream groupings define how tuples are distributed from one component to the next. They affect correctness (fields grouping ensures same keys go to same bolt for counting) and performance (shuffle grouping distributes load evenly).

Architecture

6. What happens when a worker dies?

   • The Supervisor detects the death and restarts the worker. The acker system detects timed-out tuples and triggers replays from the spout. If the Supervisor repeatedly fails to restart, Nimbus reassigns the tasks to other nodes.

7. Why is ZooKeeper needed in Storm?

   • ZooKeeper stores cluster state (assignments, heartbeats), coordinates Nimbus and Supervisors, enables Nimbus to be stateless and restartable, and supports Nimbus HA through leader election.

8. How would you handle data skew in Storm?

   • Use partial aggregation: first bolt uses shuffle grouping for partial counts, second bolt uses fields grouping for final aggregation. This distributes the load more evenly than sending all instances of a hot key to one bolt.

Performance

9. How do you tune Storm topology performance?

   • Check the Storm UI for component latency and capacity. Increase parallelism for components with capacity > 1.0. Use local-or-shuffle grouping. Tune max.spout.pending for backpressure. Batch external writes with tick tuples.

10. What is backpressure in Storm and how does it work?

    • Backpressure (since Storm 1.0) automatically slows down spouts when downstream bolts can’t keep up. When a bolt’s receive queue exceeds a threshold, it signals ZooKeeper, and spouts reduce their emission rate. This prevents tuple timeouts and OOM errors.