首页 首页 大数据 查看内容

Storm 的入门介绍和原理(官方文档)

木马童年 2019-4-29 14:30 29 0

In this tutorial, you’ll learn how to create Storm topologies and deploy them to a Storm cluster. Java will be the main language used, but a few examples will use Python to illustrate Storm’s multi- ...

In this tutorial, you’ll learn how to create Storm topologies and deploy them to a Storm cluster. Java will be the main language used, but a few examples will use Python to illustrate Storm’s multi-language capabilities.

Preliminaries

This tutorial uses examples from thestorm-starterproject. It’s recommended that you clone the project and follow along with the examples. ReadSetting up a development environmentandCreating a new Storm projectto get your machine set up.

Components of a Storm cluster

A Storm cluster is superficially similar to a Hadoop cluster. Whereas on Hadoop you run “MapReduce jobs”, on Storm you run “topologies”. “Jobs” and “topologies” themselves are very different – one key difference is that a MapReduce job eventually finishes, whereas a topology processes messages forever (or until you kill it).

There are two kinds of nodes on a Storm cluster: the master node and the worker nodes. The master node runs a daemon called “Nimbus” that is similar to Hadoop’s “JobTracker”. Nimbus is responsible for distributing code around the cluster, assigning tasks to machines, and monitoring for failures.

Each worker node runs a daemon called the “Supervisor”. The supervisor listens for work assigned to its machine and starts and stops worker processes as necessary based on what Nimbus has assigned to it. Each worker process executes a subset of a topology; a running topology consists of many worker processes spread across many machines.

Storm 的入门介绍和原理(官方文档)

All coordination between Nimbus and the Supervisors is done through aZookeepercluster. Additionally, the Nimbus daemon and Supervisor daemons are fail-fast and stateless; all state is kept in Zookeeper or on local disk. This means you can kill -9 Nimbus or the Supervisors and they’ll start back up like nothing happened. This design leads to Storm clusters being incredibly stable.

Topologies

To do realtime computation on Storm, you create what are called “topologies”. A topology is a graph of computation. Each node in a topology contains processing logic, and links between nodes indicate how data should be passed around between nodes.

Running a topology is straightforward. First, you package all your code and dependencies into a single jar. Then, you run a command like the following:

storm jar all-my-code.jar backtype.storm.MyTopology arg1 arg2

This runs the classbacktype.storm.MyTopologywith the argumentsarg1andarg2. The main function of the class defines the topology and submits it to Nimbus. Thestorm jarpart takes care of connecting to Nimbus and uploading the jar.

Since topology definitions are just Thrift structs, and Nimbus is a Thrift service, you can create and submit topologies using any programming language. The above example is the easiest way to do it from a JVM-based language. SeeRunning topologies on a production cluster] for more information on starting and stopping topologies.

Streams" >Streams

The core abstraction in Storm is the “stream”. A stream is an unbounded sequence of tuples. Storm provides the primitives for transforming a stream into a new stream in a distributed and reliable way. For example, you may transform a stream of tweets into a stream of trending topics.

The basic primitives Storm provides for doing stream transformations are “spouts” and “bolts”. Spouts and bolts have interfaces that you implement to run your application-specific logic.

A spout is a source of streams. For example, a spout may read tuples off of aKestrelqueue and emit them as a stream. Or a spout may connect to the Twitter API and emit a stream of tweets.

A bolt consumes any number of input streams, does some processing, and possibly emits new streams. Complex stream transformations, like computing a stream of trending topics from a stream of tweets, require multiple steps and thus multiple bolts. Bolts can do anything from run functions, filter tuples, do streaming aggregations, do streaming joins, talk to databases, and more.

Networks of spouts and bolts are packaged into a “topology” which is the top-level abstraction that you submit to Storm clusters for execution. A topology is a graph of stream transformations where each node is a spout or bolt. Edges in the graph indicate which bolts are subscribing to which streams. When a spout or bolt emits a tuple to a stream, it sends the tuple to every bolt that subscribed to that stream.

Storm 的入门介绍和原理(官方文档)

Links between nodes in your topology indicate how tuples should be passed around. For example, if there is a link between Spout A and Bolt B, a link from Spout A to Bolt C, and a link from Bolt B to Bolt C, then everytime Spout A emits a tuple, it will send the tuple to both Bolt B and Bolt C. All of Bolt B’s output tuples will go to Bolt C as well.

Each node in a Storm topology executes in parallel. In your topology, you can specify how much parallelism you want for each node, and then Storm will spawn that number of threads across the cluster to do the execution.

A topology runs forever, or until you kill it. Storm will automatically reassign any failed tasks. Additionally, Storm guarantees that there will be no data loss, even if machines go down and messages are dropped.

Data model

Storm uses tuples as its data model. A tuple is a named list of values, and a field in a tuple can be an object of any type. Out of the box, Storm supports all the primitive types, strings, and byte arrays as tuple field values. To use an object of another type, you just need to implementa serializerfor the type.

Every node in a topology must declare the output fields for the tuples it emits. For example, this bolt declares that it emits 2-tuples with the fields “double” and “triple”:

```java public class DoubleAndTripleBolt extends BaseRichBolt { private OutputCollectorBase _collector;

@Overridepublic void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) { _collector = collector;}@Overridepublic void execute(Tuple input) { int val = input.getInteger(0); _collector.emit(input, new Values(val*2, val*3)); _collector.ack(input);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("double", "triple"));} } ```

ThedeclareOutputFieldsfunction declares the output fields["double", "triple"]for the component. The rest of the bolt will be explained in the upcoming sections.

A simple topology

Let’s take a look at a simple topology to explore the concepts more and see how the code shapes up. Let’s look at theExclamationTopologydefinition from storm-starter:

java TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("words", new TestWordSpout(), 10); builder.setBolt("exclaim1", new ExclamationBolt(), 3) .shuffleGrouping("words"); builder.setBolt("exclaim2", new ExclamationBolt(), 2) .shuffleGrouping("exclaim1");

This topology contains a spout and two bolts. The spout emits words, and each bolt appends the string “!!!” to its input. The nodes are arranged in a line: the spout emits to the first bolt which then emits to the second bolt. If the spout emits the tuples [“bob”] and [“john”], then the second bolt will emit the words [“bob!!!!!!”] and [“john!!!!!!”].

This code defines the nodes using thesetSpoutandsetBoltmethods. These methods take as input a user-specified id, an object containing the processing logic, and the amount of parallelism you want for the node. In this example, the spout is given id “words” and the bolts are given ids “exclaim1” and “exclaim2”.

The object containing the processing logic implements theIRichSpoutinterface for spouts and theIRichBoltinterface for bolts.

The last parameter, how much parallelism you want for the node, is optional. It indicates how many threads should execute that component across the cluster. If you omit it, Storm will only allocate one thread for that node.

setBoltreturns anInputDeclarerobject that is used to define the inputs to the Bolt. Here, component “exclaim1” declares that it wants to read all the tuples emitted by component “words” using a shuffle grouping, and component “exclaim2” declares that it wants to read all the tuples emitted by component “exclaim1” using a shuffle grouping. “shuffle grouping” means that tuples should be randomly distributed from the input tasks to the bolt’s tasks. There are many ways to group data between components. These will be explained in a few sections.

If you wanted component “exclaim2” to read all the tuples emitted by both component “words” and component “exclaim1”, you would write component “exclaim2”’s definition like this:

java builder.setBolt("exclaim2", new ExclamationBolt(), 5) .shuffleGrouping("words") .shuffleGrouping("exclaim1");

As you can see, input declarations can be chained to specify multiple sources for the Bolt.

Let’s dig into the implementations of the spouts and bolts in this topology. Spouts are responsible for emitting new messages into the topology.TestWordSpoutin this topology emits a random word from the list [“nathan”, “mike”, “jackson”, “golda”, “bertels”] as a 1-tuple every 100ms. The implementation ofnextTuple()in TestWordSpout looks like this:

java public void nextTuple() { Utils.sleep(100); final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"}; final Random rand = new Random(); final String word = words[rand.nextInt(words.length)]; _collector.emit(new Values(word)); }

As you can see, the implementation is very straightforward.

ExclamationBoltappends the string “!!!” to its input. Let’s take a look at the full implementation forExclamationBolt:

```java public static class ExclamationBolt implements IRichBolt { OutputCollector _collector;

public void prepare(Map conf, TopologyContext context, OutputCollector collector) { _collector = collector;}public void execute(Tuple tuple) { _collector.emit(tuple, new Values(tuple.getString(0) + "!!!")); _collector.ack(tuple);}public void cleanup() {}public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word"));}public Map getComponentConfiguration() { return null;} } ```

Thepreparemethod provides the bolt with anOutputCollectorthat is used for emitting tuples from this bolt. Tuples can be emitted at anytime from the bolt – in theprepare,execute, orcleanupmethods, or even asynchronously in another thread. Thisprepareimplementation simply saves theOutputCollectoras an instance variable to be used later on in theexecutemethod.

Theexecutemethod receives a tuple from one of the bolt’s inputs. TheExclamationBoltgrabs the first field from the tuple and emits a new tuple with the string “!!!” appended to it. If you implement a bolt that subscribes to multiple input sources, you can find out which component theTuplecame from by using theTuple#getSourceComponentmethod.

There’s a few other things going in in theexecutemethod, namely that the input tuple is passed as the first argument toemitand the input tuple is acked on the final line. These are part of Storm’s reliability API for guaranteeing no data loss and will be explained later in this tutorial.

Thecleanupmethod is called when a Bolt is being shutdown and should cleanup any resources that were opened. There’s no guarantee that this method will be called on the cluster: for example, if the machine the task is running on blows up, there’s no way to invoke the method. Thecleanupmethod is intended for when you run topologies inlocal mode(where a Storm cluster is simulated in process), and you want to be able to run and kill many topologies without suffering any resource leaks.

ThedeclareOutputFieldsmethod declares that theExclamationBoltemits 1-tuples with one field called “word”.

ThegetComponentConfigurationmethod allows you to configure various aspects of how this component runs. This is a more advanced topic that is explained further onConfiguration.

Methods likecleanupandgetComponentConfigurationare often not needed in a bolt implementation. You can define bolts more succinctly by using a base class that provides default implementations where appropriate.ExclamationBoltcan be written more succinctly by extendingBaseRichBolt, like so:

```java public static class ExclamationBolt extends BaseRichBolt { OutputCollector _collector;

public void prepare(Map conf, TopologyContext context, OutputCollector collector) { _collector = collector;}public void execute(Tuple tuple) { _collector.emit(tuple, new Values(tuple.getString(0) + "!!!")); _collector.ack(tuple);}public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word"));} } ```

Running ExclamationTopology in local mode

Let’s see how to run theExclamationTopologyin local mode and see that it’s working.

Storm has two modes of operation: local mode and distributed mode. In local mode, Storm executes completely in process by simulating worker nodes with threads. Local mode is useful for testing and development of topologies. When you run the topologies in storm-starter, they’ll run in local mode and you’ll be able to see what messages each component is emitting. You can read more about running topologies in local mode onLocal mode.

In distributed mode, Storm operates as a cluster of machines. When you submit a topology to the master, you also submit all the code necessary to run the topology. The master will take care of distributing your code and allocating workers to run your topology. If workers go down, the master will reassign them somewhere else. You can read more about running topologies on a cluster onRunning topologies on a production cluster].

Here’s the code that runsExclamationTopologyin local mode:

```java Config conf = new Config(); conf.setDebug(true); conf.setNumWorkers(2);

LocalCluster cluster = new LocalCluster(); cluster.submitTopology(“test”, conf, builder.createTopology()); Utils.sleep(10000); cluster.killTopology(“test”); cluster.shutdown(); ```

First, the code defines an in-process cluster by creating aLocalClusterobject. Submitting topologies to this virtual cluster is identical to submitting topologies to distributed clusters. It submits a topology to theLocalClusterby callingsubmitTopology, which takes as arguments a name for the running topology, a configuration for the topology, and then the topology itself.

The name is used to identify the topology so that you can kill it later on. A topology will run indefinitely until you kill it.

The configuration is used to tune various aspects of the running topology. The two configurations specified here are very common:

  1. TOPOLOGY_WORKERS(set withsetNumWorkers) specifies how manyprocessesyou want allocated around the cluster to execute the topology. Each component in the topology will execute as manythreads. The number of threads allocated to a given component is configured through thesetBoltandsetSpoutmethods. Thosethreadsexist within workerprocesses. Each workerprocesscontains within it some number ofthreadsfor some number of components. For instance, you may have 300 threads specified across all your components and 50 worker processes specified in your config. Each worker process will execute 6 threads, each of which of could belong to a different component. You tune the performance of Storm topologies by tweaking the parallelism for each component and the number of worker processes those threads should run within.
  2. TOPOLOGY_DEBUG(set withsetDebug), when set to true, tells Storm to log every message every emitted by a component. This is useful in local mode when testing topologies, but you probably want to keep this turned off when running topologies on the cluster.

There’s many other configurations you can set for the topology. The various configurations are detailed onthe Javadoc for Config.

To learn about how to set up your development environment so that you can run topologies in local mode (such as in Eclipse), seeCreating a new Storm project.

Stream groupings

A stream grouping tells a topology how to send tuples between two components. Remember, spouts and bolts execute in parallel as many tasks across the cluster. If you look at how a topology is executing at the task level, it looks something like this:

Storm 的入门介绍和原理(官方文档)

When a task for Bolt A emits a tuple to Bolt B, which task should it send the tuple to?

A “stream grouping” answers this question by telling Storm how to send tuples between sets of tasks. Before we dig into the different kinds of stream groupings, let’s take a look at another topology fromstorm-starter. ThisWordCountTopologyreads sentences off of a spout and streams out ofWordCountBoltthe total number of times it has seen that word before:

```java TopologyBuilder builder = new TopologyBuilder();

builder.setSpout(“sentences”, new RandomSentenceSpout(), 5);

builder.setBolt(“split”, new SplitSentence(), 8) .shuffleGrouping(“sentences”); builder.setBolt(“count”, new WordCount(), 12) .fieldsGrouping(“split”, new Fields(“word”)); ```

SplitSentenceemits a tuple for each word in each sentence it receives, andWordCountkeeps a map in memory from word to count. Each timeWordCountreceives a word, it updates its state and emits the new word count.

There’s a few different kinds of stream groupings.

The simplest kind of grouping is called a “shuffle grouping” which sends the tuple to a random task. A shuffle grouping is used in theWordCountTopologyto send tuples fromRandomSentenceSpoutto theSplitSentencebolt. It has the effect of evenly distributing the work of processing the tuples across all ofSplitSentencebolt’s tasks.

A more interesting kind of grouping is the “fields grouping”. A fields grouping is used between theSplitSentencebolt and theWordCountbolt. It is critical for the functioning of theWordCountbolt that the same word always go to the same task. Otherwise, more than one task will see the same word, and they’ll each emit incorrect values for the count since each has incomplete information. A fields grouping lets you group a stream by a subset of its fields. This causes equal values for that subset of fields to go to the same task. SinceWordCountsubscribes toSplitSentence’s output stream using a fields grouping on the “word” field, the same word always goes to the same task and the bolt produces the correct output.

Fields groupings are the basis of implementing streaming joins and streaming aggregations as well as a plethora of other use cases. Underneath the hood, fields groupings are implemented using mod hashing.

There’s a few other kinds of stream groupings. You can read more about them onConcepts.

Defining Bolts in other languages

Bolts can be defined in any language. Bolts written in another language are executed as subprocesses, and Storm communicates with those subprocesses with JSON messages over stdin/stdout. The communication protocol just requires an ~100 line adapter library, and Storm ships with adapter libraries for Ruby, Python, and Fancy.

Here’s the definition of theSplitSentencebolt fromWordCountTopology:

```java public static class SplitSentence extends ShellBolt implements IRichBolt { public SplitSentence() { super(“python”, “splitsentence.py”); }

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

SplitSentenceoverridesShellBoltand declares it as running usingpythonwith the argumentssplitsentence.py. Here’s the implementation ofsplitsentence.py:

```python import storm

class SplitSentenceBolt(storm.BasicBolt): def process(self, tup): words = tup.values[0].split(“ “) for word in words: storm.emit([word])

SplitSentenceBolt().run() ```

For more information on writing spouts and bolts in other languages, and to learn about how to create topologies in other languages (and avoid the JVM completely), seeUsing non-JVM languages with Storm.

Guaranteeing message processing

Earlier on in this tutorial, we skipped over a few aspects of how tuples are emitted. Those aspects were part of Storm’s reliability API: how Storm guarantees that every message coming off a spout will be fully processed. SeeGuaranteeing message processingfor information on how this works and what you have to do as a user to take advantage of Storm’s reliability capabilities.

Transactional topologies

Storm guarantees that every message will be played through the topology at least once. A common question asked is “how do you do things like counting on top of Storm? Won’t you overcount?” Storm has a feature called transactional topologies that let you achieve exactly-once messaging semantics for most computations. Read more about transactional topologieshere.

Distributed RPC

This tutorial showed how to do basic stream processing on top of Storm. There’s lots more things you can do with Storm’s primitives. One of the most interesting applications of Storm is Distributed RPC, where you parallelize the computation of intense functions on the fly. Read more about Distributed RPChere.

Conclusion

This tutorial gave a broad overview of developing, testing, and deploying Storm topologies. The rest of the documentation dives deeper into all the aspects of using Storm.

在不久的将来,多智时代一定会彻底走入我们的生活,有兴趣入行未来前沿产业的朋友,可以收藏多智时代,及时获取人工智能、大数据、云计算和物联网的前沿资讯和基础知识,让我们一起携手,引领人工智能的未来!

未分类
0
为您推荐
大数据技术改变城市的运作方式,智慧城市呼之欲出

大数据技术改变城市的运作方式,智慧城市呼

纽奥良虽像大多数城市一样有火灾侦测器安装计划,但直到最近还是要由市民主动申装。纽…...

大数据分析面临生死边缘,未来之路怎么走?

大数据分析面临生死边缘,未来之路怎么走?

大数据分析开始朝着营销落地,尤其像数果智能这类服务于企业的大数据分析供应商,不仅…...

什么是工业大数据,要通过3B和3C来理解?

什么是工业大数据,要通过3B和3C来理解?

核心提示:工业视角的转变如果说前三次工业革命分别从机械化、规模化、标准化、和自动…...

大数据普及为什么说肥了芯片厂商?

大数据普及为什么说肥了芯片厂商?

科技界默默无闻的存在,芯片行业年规模增长到了3520亿美元。半导体给无人驾驶汽车带来…...

大数据技术有哪些,为什么说云计算能力是大数据的根本!

大数据技术有哪些,为什么说云计算能力是大

历史规律告诉我们,任何一次大型技术革命,早期人们总是高估它的影响,会有一轮一轮的…...

个人征信牌照推迟落地,大数据 重新定义个人信用!!

个人征信牌照推迟落地,大数据 重新定义个

为金融学的基础正日益坚实。通过互联网大数据精准记录海量个人行为,进而形成分析结论…...