The Project | About Us | Contribute | Donations | License

HOME

In this module, we will explore two powerful tools for real-time data processing: Apache Flink and Apache Storm. Both of these frameworks are designed to handle large-scale data streams efficiently, but they have different architectures and use cases. By the end of this module, you will understand the core concepts, features, and differences between Flink and Storm, and you will be able to implement basic data processing tasks using these tools.

Introduction to Apache Flink
Introduction to Apache Storm
Comparison of Flink and Storm
Exercises
Summary

Introduction to Apache Flink

Apache Flink is a stream processing framework that can handle both batch and real-time data processing. It is known for its high throughput, low latency, and fault tolerance.

Key Features of Flink

Stream and Batch Processing: Flink can handle both stream and batch data processing.
Event Time Processing: Flink supports event time processing, which allows for accurate handling of out-of-order events.
Stateful Computations: Flink provides robust support for stateful computations, which is essential for complex event processing.
Fault Tolerance: Flink ensures fault tolerance through distributed snapshots and state recovery mechanisms.
Scalability: Flink can scale horizontally to handle large volumes of data.

Flink Architecture

Flink's architecture consists of several key components:

JobManager: Manages the execution of jobs, including scheduling tasks and coordinating checkpoints.
TaskManager: Executes the tasks assigned by the JobManager and manages the local state.
Client: Submits jobs to the JobManager.

Basic Flink Example

Let's look at a simple example of a Flink job that counts the occurrences of words in a text stream.

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;

public class WordCount {
    public static void main(String[] args) throws Exception {
        // Set up the execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // Get input data
        DataSet<String> text = env.readTextFile("path/to/input.txt");

        // Split up the lines in pairs (2-tuples) containing: (word,1)
        DataSet<Tuple2<String, Integer>> counts = text
            .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(new Tuple2<>(word, 1));
                }
            })
            .groupBy(0)
            .sum(1);

        // Execute and print result
        counts.print();
    }
}

Explanation

ExecutionEnvironment: The context in which the program is executed.
DataSet: Represents a collection of data.
flatMap: Splits lines into words and pairs them with the number 1.
groupBy: Groups the tuples by the word.
sum: Sums up the counts for each word.

Introduction to Apache Storm

Apache Storm is a distributed real-time computation system that processes streams of data with high reliability and fault tolerance.

Key Features of Storm

Real-Time Processing: Storm is designed for real-time data processing.
Scalability: Storm can scale horizontally to handle large volumes of data.
Fault Tolerance: Storm ensures fault tolerance through automatic task re-execution.
At-Least-Once Processing: Storm guarantees that each message will be processed at least once.
Extensibility: Storm can be extended with custom spouts and bolts.

Storm Architecture

Storm's architecture consists of several key components:

Nimbus: The master node that distributes code around the cluster, assigns tasks to machines, and monitors for failures.
Supervisor: Runs on worker nodes and listens for work assigned to its machine.
Worker: Executes a subset of a topology.
Topology: A network of spouts and bolts that process streams of data.

Basic Storm Example

Let's look at a simple example of a Storm topology that counts the occurrences of words in a text stream.

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.spout.ISpout;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.topology.base.BaseRichBolt;

import java.util.Map;
import java.util.StringTokenizer;

public class WordCountTopology {
    public static class WordSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;

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

        @Override
        public void nextTuple() {
            String sentence = "the cow jumped over the moon";
            collector.emit(new Values(sentence));
        }

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

    public static class SplitSentenceBolt extends BaseRichBolt {
        private OutputCollector collector;

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

        @Override
        public void execute(org.apache.storm.tuple.Tuple tuple) {
            String sentence = tuple.getStringByField("sentence");
            StringTokenizer tokenizer = new StringTokenizer(sentence);
            while (tokenizer.hasMoreTokens()) {
                String word = tokenizer.nextToken();
                collector.emit(new Values(word));
            }
        }

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

    public static class WordCountBolt extends BaseRichBolt {
        private OutputCollector collector;
        private Map<String, Integer> counts = new HashMap<>();

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

        @Override
        public void execute(org.apache.storm.tuple.Tuple tuple) {
            String word = tuple.getStringByField("word");
            Integer count = counts.get(word);
            if (count == null) {
                count = 0;
            }
            count++;
            counts.put(word, count);
            collector.emit(new Values(word, count));
        }

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

    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word-spout", new WordSpout());
        builder.setBolt("split-sentence-bolt", new SplitSentenceBolt()).shuffleGrouping("word-spout");
        builder.setBolt("word-count-bolt", new WordCountBolt()).fieldsGrouping("split-sentence-bolt", new Fields("word"));

        Config config = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word-count-topology", config, builder.createTopology());
        Thread.sleep(10000);
        cluster.shutdown();
    }
}

Explanation

Spout: Emits sentences.
SplitSentenceBolt: Splits sentences into words.
WordCountBolt: Counts the occurrences of each word.
TopologyBuilder: Defines the topology.
LocalCluster: Runs the topology locally.

Comparison of Flink and Storm

Feature	Apache Flink	Apache Storm
Processing Model	Stream and Batch Processing	Stream Processing
Fault Tolerance	Exactly-once semantics	At-least-once semantics
State Management	Built-in state management	Requires external state management
Latency	Low latency	Low latency
Scalability	High scalability	High scalability
Ease of Use	Higher-level APIs	Lower-level APIs
Event Time Processing	Supported	Limited support

Exercises

Exercise 1: Implement a Flink Job

Create a Flink job that reads a text file, counts the occurrences of each word, and writes the results to an output file.

Exercise 2: Implement a Storm Topology

Create a Storm topology that reads sentences from a spout, splits them into words using a bolt, and counts the occurrences of each word using another bolt.

Solutions

Solution 1: Flink Job

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;

public class WordCount {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<String> text = env.readTextFile("path/to/input.txt");
        DataSet<Tuple2<String, Integer>> counts = text
            .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(new Tuple2<>(word, 1));
                }
            })
            .groupBy(0)
            .sum(1);
        counts.writeAsCsv("path/to/output.csv", "\n", " ");
        env.execute("Word Count Example");
    }
}

Solution 2: Storm Topology

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.spout.ISpout;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.topology.base.BaseRichBolt;

import java.util.Map;
import java.util.StringTokenizer;

public class WordCountTopology {
    public static class WordSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;

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

        @Override
        public void nextTuple() {
            String sentence = "the cow jumped over the moon";
            collector.emit(new Values(sentence));
        }

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

    public static class SplitSentenceBolt extends BaseRichBolt {
        private OutputCollector collector;

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

        @Override
        public void execute(org.apache.storm.tuple.Tuple tuple) {
            String sentence = tuple.getStringByField("sentence");
            StringTokenizer tokenizer = new StringTokenizer(sentence);
            while (tokenizer.hasMoreTokens()) {
                String word = tokenizer.nextToken();
                collector.emit(new Values(word));
            }
        }

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

    public static class WordCountBolt extends BaseRichBolt {
        private OutputCollector collector;
        private Map<String, Integer> counts = new HashMap<>();

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

        @Override
        public void execute(org.apache.storm.tuple.Tuple tuple) {
            String word = tuple.getStringByField("word");
            Integer count = counts.get(word);
            if (count == null) {
                count = 0;
            }
            count++;
            counts.put(word, count);
            collector.emit(new Values(word, count));
        }

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

    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word-spout", new WordSpout());
        builder.setBolt("split-sentence-bolt", new SplitSentenceBolt()).shuffleGrouping("word-spout");
        builder.setBolt("word-count-bolt", new WordCountBolt()).fieldsGrouping("split-sentence-bolt", new Fields("word"));

        Config config = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word-count-topology", config, builder.createTopology());
        Thread.sleep(10000);
        cluster.shutdown();
    }
}

Summary

In this module, we explored Apache Flink and Apache Storm, two powerful tools for real-time data processing. We covered their key features, architectures, and provided basic examples to illustrate their usage. We also compared Flink and Storm to highlight their differences and use cases. Finally, we provided practical exercises to reinforce the learned concepts. Understanding these tools will enable you to handle large-scale data streams efficiently and build robust data processing applications.

Flink and Storm

Table of Contents

Introduction to Apache Flink

Key Features of Flink

Flink Architecture

Basic Flink Example

Explanation

Introduction to Apache Storm

Key Features of Storm

Storm Architecture

Basic Storm Example

Explanation

Comparison of Flink and Storm

Exercises

Exercise 1: Implement a Flink Job

Exercise 2: Implement a Storm Topology

Solutions

Solution 1: Flink Job

Solution 2: Storm Topology

Summary

Massive Data Processing

Module 1: Introduction to Massive Data Processing

Module 2: Storage Technologies

Module 3: Processing Techniques

Module 4: Tools and Platforms

Module 5: Storage and Processing Optimization

Module 6: Massive Data Analysis

Module 7: Case Studies and Practical Applications

Module 8: Best Practices and Future of Massive Data Processing