[Storm中文文檔]Trident教程

聲明

本文首發(fā)于個(gè)人技術(shù)博客，轉(zhuǎn)載請(qǐng)注明出處，本文鏈接：http://qifuguang.me/2015/11/27/Storm中文文檔-Trident教程/

前言

最近工作中需要用到Storm，于是買(mǎi)了一本關(guān)于Storm的書(shū)，翻開(kāi)一看基本都是照著官方文檔翻譯的，并且翻譯的質(zhì)量真的不敢恭維。我自認(rèn)為語(yǔ)文水平比英語(yǔ)水平要好，但是我看別人翻譯的書(shū)竟然看不懂，一怒一下，自己研究官方文檔來(lái)了。我知道有很多和我一樣的人，不喜歡看英文的文檔，所以就一邊看文檔，一邊翻譯，分享出來(lái)，希望對(duì)需要的人有所幫助。當(dāng)然，我的英語(yǔ)很渣，翻譯的時(shí)候難免有差錯(cuò)，敬請(qǐng)諒解。

正文

Trident是一個(gè)基于Storm的用于實(shí)時(shí)計(jì)算的高級(jí)抽象原語(yǔ)。它支持高吞吐（每秒百萬(wàn)級(jí)別），有狀態(tài)的流處理，并且還能夠提供低延時(shí)的分布式查詢(xún)功能。如果你熟悉一些比較高級(jí)的批處理工具，比如Pig和Cascading，那么對(duì)于Trident你應(yīng)該有一種似曾相識(shí)的感覺(jué)。Trident具有連接，聚合，分組，自定義行為和過(guò)濾的功能。除此之外，Trident能夠基于內(nèi)存或者數(shù)據(jù)庫(kù)做有狀態(tài)的，增量式的計(jì)算。Trident本身能夠保證每個(gè)Tuple嚴(yán)格只被執(zhí)行一次，所以使用Trident很容易構(gòu)建一個(gè)靠譜的Topology。

Illustrative example

下面通過(guò)一個(gè)例子介紹Trident。這個(gè)例子需要做兩件事：

  1. 從一個(gè)能產(chǎn)生句子的輸入流中實(shí)時(shí)計(jì)算各個(gè)單詞的數(shù)量；
  2. 實(shí)現(xiàn)查詢(xún)功能：輸入一個(gè)句子，句子中每個(gè)單詞用空格分隔，查詢(xún)這個(gè)句子中所有單詞出現(xiàn)的數(shù)量的總和。

出于演示目的，本例將從一個(gè)能夠產(chǎn)生無(wú)限英文句子的輸入流中讀取數(shù)據(jù)：

FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
               new Values("the cow jumped over the moon"),
               new Values("the man went to the store and bought some candy"),
               new Values("four score and seven years ago"),
               new Values("how many apples can you eat"));
spout.setCycle(true);

該Spout能夠循環(huán)產(chǎn)生無(wú)限的英文語(yǔ)句，下面的代碼是計(jì)算單詞出現(xiàn)次數(shù)的部分代碼：

TridentTopology topology = new TridentTopology();        
TridentState wordCounts =
     topology.newStream("spout1", spout)
       .each(new Fields("sentence"), new Split(), new Fields("word"))
       .groupBy(new Fields("word"))
       .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))                
       .parallelismHint(6);

我們一行一行地對(duì)照上面的代碼來(lái)作說(shuō)明。首先創(chuàng)建一個(gè)TridentTopology，它提供了用于構(gòu)建Trident實(shí)時(shí)計(jì)算程序的一些接口。TridentTopology有一個(gè)函數(shù)叫做newStream，它通過(guò)一個(gè)指定的Spout創(chuàng)建一個(gè)新的數(shù)據(jù)輸入流。在本例中，輸入流僅僅是一個(gè)比較簡(jiǎn)單的FixedBatchSpout。輸入流也可以是消息隊(duì)列，比如Kestrel和Kafka。Trident在Zookeeper保存每一個(gè)從輸入流中讀取的Tuple的處理信息，在上面的代碼中，字符串"spout1"表示這些Tuple的處理信息在Zookeeper上的存儲(chǔ)路徑。

Trident是將輸入數(shù)據(jù)分成許多小塊做批量處理的。例如，本例中輸入的數(shù)據(jù)流有可能被分割成如下這樣的小塊：



通常來(lái)說(shuō)，每一個(gè)batch可能包含幾千到上百萬(wàn)的Tuple，這完全取決于輸入的數(shù)據(jù)量。

Trident提供了一整套比較完整的API來(lái)處理這些batch中的數(shù)據(jù)。這些API與Pig和Cascading的API非常類(lèi)似：能夠分組，連接，聚合，執(zhí)行自定義行為，還能進(jìn)行過(guò)濾等等。當(dāng)然，獨(dú)立處理每一個(gè)batch并沒(méi)有多大意義，所以，Trident提供了跨batch的數(shù)據(jù)聚合與存儲(chǔ)功能，比如存儲(chǔ)在Memory，Memcache，Cassandra或者其他存儲(chǔ)設(shè)備。此外，Trident還能提供一流的實(shí)時(shí)查詢(xún)功能。這些狀態(tài)能夠被Trident更新（就像上面的例子），也能作為一個(gè)獨(dú)立的狀態(tài)源存在（筆者注：這句話(huà)不太理解?。?。

回到上面的例子，Spout發(fā)射了一個(gè)包含sentence字段的數(shù)據(jù)流。下一行代碼定義了一個(gè)函數(shù)Split用于處理輸入流中的每一個(gè)Tuple：獲取sentence字段并將其切割成很多單詞。每一個(gè)sentence類(lèi)型的Tuple將會(huì)衍生出很多word類(lèi)型的Tuple：比如本例中'the cow jumped over the moon'這個(gè)sentence類(lèi)型的Tuple將會(huì)衍生出6個(gè)word類(lèi)型的Tuple。下面是Split函數(shù)的定義：

public class Split extends BaseFunction {
   public void execute(TridentTuple tuple, TridentCollector collector) {
       String sentence = tuple.getString(0);
       for(String word: sentence.split(" ")) {
           collector.emit(new Values(word));                
       }
   }
}

正如我們所見(jiàn)到的，Split函數(shù)的定義非常簡(jiǎn)單。僅僅是獲取sentence，然后用空格切割成很多word，然后分別發(fā)射這些word。

剩余的代碼計(jì)算word出現(xiàn)的次數(shù)并將結(jié)果保存起來(lái)。首先數(shù)據(jù)流按照word字段分組，然后每一個(gè)分組都被自定義聚合器Count聚合。persistentAggregate函數(shù)知道怎么存儲(chǔ)和更新計(jì)算結(jié)果的狀態(tài)。在本例中，單詞出現(xiàn)的次數(shù)被保存在內(nèi)存中，但是也可以替換成其他的存儲(chǔ)設(shè)備，比如Memcached，Cassandra等等。比如替換存儲(chǔ)設(shè)備為Memcached很簡(jiǎn)單，只需要將包含persistentAggregate的這一行代碼替換成下面的代碼即可，這兒的serverLocations表示的是Memcached集群的機(jī)器的host/ip列表：

.persistentAggregate(MemcachedState.transactional(serverLocations), new Count(), new Fields("count"))        
MemcachedState.transactional()

persistentAggregate存儲(chǔ)的值就是所有batch聚合之后的值。

Trident一個(gè)比較酷的事情就是它是完全容錯(cuò)的，保證數(shù)據(jù)嚴(yán)格只被執(zhí)行一次的。這使得我們很容易構(gòu)建靠譜的實(shí)時(shí)計(jì)算系統(tǒng)。Trident存儲(chǔ)每個(gè)Tuple的處理狀態(tài)，以便在有錯(cuò)誤發(fā)生時(shí)，可以恢復(fù)數(shù)據(jù)并且防止一個(gè)數(shù)據(jù)被重復(fù)處理。

persistentAggregate方法會(huì)將數(shù)據(jù)流轉(zhuǎn)換成TridentState對(duì)象。在本例中TridentState對(duì)象就是所有單詞的統(tǒng)計(jì)數(shù)據(jù)。我們將會(huì)運(yùn)用這個(gè)TridentState對(duì)象來(lái)實(shí)現(xiàn)一個(gè)分布式的實(shí)時(shí)查詢(xún)系統(tǒng)。

Topology的另一個(gè)部分就是實(shí)現(xiàn)一個(gè)低延時(shí)的分布式查詢(xún)系統(tǒng)：輸入一個(gè)句子，句子中的單詞用空格分隔，查詢(xún)系統(tǒng)返回這些單詞的統(tǒng)計(jì)數(shù)據(jù)的和。除了在后臺(tái)是并行執(zhí)行的，這種查詢(xún)和普通的RPC調(diào)用沒(méi)有啥區(qū)別。下面是一個(gè)查詢(xún)的例子：

DRPCClient client = new DRPCClient("drpc.server.location", 3772);
System.out.println(client.execute("words", "cat dog the man");
// prints the JSON-encoded result, e.g.: "[[5078]]"

正如你所見(jiàn)到的，它和普通的RPC調(diào)用完全沒(méi)有區(qū)別，除了它是在Storm集群間并行執(zhí)行的。對(duì)于一般的小型的查詢(xún)，耗時(shí)大概在10ms，當(dāng)然越重的查詢(xún)花費(fèi)的時(shí)間會(huì)更長(zhǎng)，盡管延時(shí)還受到所分配的資源多少的影響。

實(shí)現(xiàn)分布式查詢(xún)系統(tǒng)的代碼如下：

topology.newDRPCStream("words")
       .each(new Fields("args"), new Split(), new Fields("word"))
       .groupBy(new Fields("word"))
       .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
       .each(new Fields("count"), new FilterNull())
       .aggregate(new Fields("count"), new Sum(), new Fields("sum"));

相同的TridentTopology被用來(lái)創(chuàng)建一個(gè)新的DRPC流，這個(gè)函數(shù)命名為words。當(dāng)執(zhí)行查詢(xún)操作的時(shí)候，這個(gè)名字會(huì)作為DRPCClient的第一個(gè)參數(shù)。

每一個(gè)DRPC請(qǐng)求會(huì)被當(dāng)做一個(gè)只有一個(gè)Tuple的batch處理。Tuple包含一個(gè)叫做args的字段，該字段表示DRPC客戶(hù)端提供的參數(shù)，在本例中，該參數(shù)就是一個(gè)sentence。

首先，Split函數(shù)用來(lái)將輸入的sentence切割成很多個(gè)word。然后數(shù)據(jù)流會(huì)按照word分組，然后stateQuery函數(shù)將會(huì)在第一部分創(chuàng)建的TridentState（wordCounts）上執(zhí)行查詢(xún)操作。StateQuery接收一個(gè)數(shù)據(jù)源（在本例中，就是已經(jīng)計(jì)算好的單詞統(tǒng)計(jì)數(shù)據(jù)）和一個(gè)用于查詢(xún)的函數(shù)作為輸入。在本例中，我們使用MapGet函數(shù)來(lái)獲取單詞的統(tǒng)計(jì)值。由于單詞是按照和構(gòu)建TridentState時(shí)一樣的方法（按照word分組）分組的，所以每一個(gè)單詞的查詢(xún)請(qǐng)求都會(huì)被路由到管理和更新該單詞的統(tǒng)計(jì)數(shù)據(jù)的分區(qū)中取執(zhí)行。

下一步，沒(méi)有出現(xiàn)過(guò)的單詞（count值為0）將會(huì)被Storm內(nèi)建的FilterNull過(guò)濾器過(guò)濾，然后Sum函數(shù)將會(huì)把所有單詞的統(tǒng)計(jì)數(shù)據(jù)做一次求和操作，保存到sum字段中，并返回給DRPC客戶(hù)端。

Trident致力于提高Topology結(jié)構(gòu)的性能。在上面的示例中，Trident自動(dòng)完成了兩件事：

  1. 讀取和寫(xiě)入操作會(huì)自動(dòng)使用batch的方式執(zhí)行，如果有20次更新需要被同步到數(shù)據(jù)庫(kù)中，Trident會(huì)自動(dòng)將這些操作匯總到一起，只做一次讀寫(xiě)操作，而不是20次。因此Trident可以在方便你計(jì)算的同時(shí)提高極高的性能。
  2. Trident對(duì)聚合操作做了極大的優(yōu)化。Trident并不是簡(jiǎn)單地把一個(gè)Group中所有的Tuple都發(fā)送到同一個(gè)機(jī)器上進(jìn)行聚合，如果條件允許，Trident在將數(shù)據(jù)通過(guò)網(wǎng)路發(fā)送之前已經(jīng)做了部分聚合操作，Count聚合器在每一個(gè)分區(qū)分別計(jì)算統(tǒng)計(jì)值，然后通過(guò)網(wǎng)絡(luò)發(fā)送分區(qū)的計(jì)算結(jié)果，然后將所有分區(qū)的計(jì)算結(jié)果進(jìn)行匯總得到總的結(jié)果。這與MapReduce的計(jì)算模型非常相似。

再看一個(gè)Trident的例子：

Reach

這個(gè)例子是一個(gè)用于計(jì)算一個(gè)給定的URL的Reach的純粹的DRPC Topology。什么是URL的Reach？Reach是指在Twitter上看到過(guò)一個(gè)URL的不同的人的數(shù)量，要計(jì)算Reach值，首先你要獲取曾經(jīng)發(fā)布過(guò)該URL的所有人，然后獲取所有這些人的所有粉絲，然后將這些粉絲做唯一化處理，得到的數(shù)字就是該URL的Reach值。如果使用單機(jī)計(jì)算URL的Reach值，這將會(huì)是一個(gè)非常繁重的任務(wù)，因?yàn)檫@將會(huì)產(chǎn)生成千上萬(wàn)的數(shù)據(jù)庫(kù)請(qǐng)求，千萬(wàn)級(jí)別的Tuple數(shù)量。使用Storm和Trident，你能夠?qū)⑸厦嬲f(shuō)到的這些步驟在集群機(jī)器間作并行計(jì)算。

該Topoloy將會(huì)從兩個(gè)地方讀取數(shù)據(jù)。一個(gè)用于讀取曾經(jīng)發(fā)布過(guò)該URL的人，另一個(gè)用于讀取這些人的粉絲。定義如下：

TridentState urlToTweeters =
       topology.newStaticState(getUrlToTweetersState());
TridentState tweetersToFollowers =
       topology.newStaticState(getTweeterToFollowersState());

topology.newDRPCStream("reach")
       .stateQuery(urlToTweeters, new Fields("args"), new MapGet(), new Fields("tweeters"))
       .each(new Fields("tweeters"), new ExpandList(), new Fields("tweeter"))
       .shuffle()
       .stateQuery(tweetersToFollowers, new Fields("tweeter"), new MapGet(), new Fields("followers"))
       .parallelismHint(200)
       .each(new Fields("followers"), new ExpandList(), new Fields("follower"))
       .groupBy(new Fields("follower"))
       .aggregate(new One(), new Fields("one"))
       .parallelismHint(20)
       .aggregate(new Count(), new Fields("reach"));

上面的的topology使用newStaticState方法創(chuàng)建一個(gè)TridentState對(duì)象，表示一種外部存儲(chǔ)，使用這個(gè)TridentState對(duì)象，我們就能在該topology上面進(jìn)行查詢(xún)了，和其他所有狀態(tài)源一樣，查詢(xún)請(qǐng)求會(huì)自動(dòng)轉(zhuǎn)換成batch類(lèi)型的請(qǐng)求，以提高性能。

這個(gè)拓?fù)涞亩x非常簡(jiǎn)單，它就是一個(gè)簡(jiǎn)單的批處理任務(wù)。首先，urlToTweeters用于查詢(xún)?cè)?jīng)發(fā)布過(guò)該URL的人，返回一個(gè)列表，然后使用ExpandList函數(shù)將該列表轉(zhuǎn)換成一個(gè)個(gè)的Tuple分別發(fā)射出去。

接下來(lái)，我們獲取每一個(gè)tweeter的followers。我們使用suffle函數(shù)將需要處理的tweeter分配到集群的每一個(gè)分區(qū)，然后集群的每一個(gè)分區(qū)會(huì)分別獲取他們收到的tweeter的follower，可以為該步驟設(shè)置很大的并行度，提高查詢(xún)性能。

接下來(lái)，tweeter的所有follower需要去重，這個(gè)可以分兩個(gè)步驟完成：首先按照f(shuō)ollower分組，然后使用One聚合器對(duì)每一個(gè)分組進(jìn)行聚合，One聚合器僅僅是簡(jiǎn)單地為每一個(gè)分組發(fā)射一個(gè)Tuple（重復(fù)的Tuple被舍棄），保存在one字段中，然后使用Count聚合器統(tǒng)計(jì)Tuple的數(shù)量，存入reach字段中，這個(gè)值就是URL的Reach值。One聚合器的定義如下：

public class One implements CombinerAggregator<Integer> {
   public Integer init(TridentTuple tuple) {
       return 1;
   }

   public Integer combine(Integer val1, Integer val2) {
       return 1;
   }

   public Integer zero() {
       return 1;
   }        
}

這是一個(gè)"combiner aggregator"，為了提高性能，它會(huì)在將Tuple通過(guò)網(wǎng)絡(luò)傳輸之前做部分聚合操作。Sum集合器也是一個(gè)"combiner aggregator"，因此在最后計(jì)算總值是非常高效的。

再看看Trident更細(xì)節(jié)的東西.

Fields and tuples

Trident的數(shù)據(jù)模型就是一個(gè)TridentTuple，它是一個(gè)命名的值列表。在topology中，TridentTuple是在一系列的計(jì)算中增量產(chǎn)生的。這些操作一般以一組字段作為輸入，然后產(chǎn)生一組輸出字段，輸入一般是輸入Tuple的一組子字段。

參照如下的例子，假設(shè)你有一個(gè)輸入流，命名為stream，它包含三個(gè)字段: x, y, z， 為了運(yùn)行一個(gè)過(guò)濾器，并且這個(gè)過(guò)濾器只接受輸入流中的y字段，我們可以這樣寫(xiě)：

stream.each(new Fields("y"), new MyFilter())

假設(shè)MyFilter的定義如下：

public class MyFilter extends BaseFilter {
   public boolean isKeep(TridentTuple tuple) {
       return tuple.getInteger(0) < 10;
   }
}

只保留y字段的值小于10的Tuple。TridentTuple傳給MyFilter過(guò)濾器的輸入Tuple包含一個(gè)字段y。需要注意的是，選擇一個(gè)Tuple的某些字段的這個(gè)操作是非常高效的。

接下來(lái)看看"function fields"是如何工作的。假設(shè)你定義了一個(gè)函數(shù)：

public class AddAndMultiply extends BaseFunction {
   public void execute(TridentTuple tuple, TridentCollector collector) {
       int i1 = tuple.getInteger(0);
       int i2 = tuple.getInteger(1);
       collector.emit(new Values(i1 + i2, i1 * i2));
   }
}

該函數(shù)接收兩個(gè)數(shù)字同時(shí)發(fā)射兩個(gè)數(shù)字：輸入的兩個(gè)數(shù)字的和和輸入的兩個(gè)數(shù)字的積。假設(shè)你有一個(gè)輸入流，包含x,y,z三個(gè)字段，你可以像下面這樣使用上面的函數(shù)：

stream.each(new Fields("x", "y"), new AddAndMultiply(), new Fields("added", "multiplied"));

函數(shù)的輸入字段是追加到輸入的字段后面的，所以該函數(shù)執(zhí)行后的輸出會(huì)包含5個(gè)字段：x,y,z, added, multiplied，added字段是AddAndMultiply發(fā)射的第一個(gè)字段，multiplied是第二個(gè)字段。

但是對(duì)于聚合操作，新產(chǎn)生的字段將會(huì)替換輸入的字段。所以，如果你有一個(gè)輸入流，包含val1,val2兩個(gè)字段，當(dāng)你做如下的操作：

stream.aggregate(new Fields("val2"), new Sum(), new Fields("sum"))

之后，輸出的Tuple會(huì)只包含sum一個(gè)字段，表示該batch種所有val2字段的值的總和。

對(duì)于分組之后再聚合的操作，輸出字段將會(huì)包含分組的字段和聚合操作新產(chǎn)生的字段。比如：

stream.groupBy(new Fields("val1"))
     .aggregate(new Fields("val2"), new Sum(), new Fields("sum"))

上面的例子的輸出結(jié)果中將會(huì)包含val1和sum兩個(gè)字段。

State

實(shí)時(shí)計(jì)算系統(tǒng)一個(gè)比較關(guān)鍵的問(wèn)題就是如何確保在有錯(cuò)誤發(fā)生或者重試時(shí)還能保證計(jì)算的正確性。出問(wèn)題是無(wú)法避免的，所以當(dāng)某個(gè)節(jié)點(diǎn)宕機(jī)或者其他錯(cuò)誤出現(xiàn)時(shí)，我們需要重試?，F(xiàn)在的問(wèn)題是：如何保證在重試過(guò)程中，一條數(shù)據(jù)值被處理一次？

這是一個(gè)比較困難的問(wèn)題，我們通過(guò)一個(gè)例子進(jìn)行說(shuō)明。假設(shè)你現(xiàn)在正在統(tǒng)計(jì)輸入流中處理過(guò)的Tuple的數(shù)量，并且需要將統(tǒng)計(jì)結(jié)果保存到數(shù)據(jù)庫(kù)中。如果你僅僅在數(shù)據(jù)庫(kù)中存儲(chǔ)一個(gè)統(tǒng)計(jì)值，現(xiàn)在如果你想進(jìn)行一次狀態(tài)更新，那么你將無(wú)法知道當(dāng)前的這個(gè)Tuple是否之前已經(jīng)被處理過(guò)，或者之前嘗試處理過(guò)，數(shù)據(jù)庫(kù)也更新成功，但是之后的某些步驟失敗了，亦或者其他步驟都處理成功了，但是更新數(shù)據(jù)庫(kù)失敗了。

要解決這個(gè)問(wèn)題需要做如下兩件事情：

  1. 為每一個(gè)batch分配一個(gè)唯一的transaction id（txid），當(dāng)batch數(shù)據(jù)重試時(shí)，該batch會(huì)具有和之前一樣的txid；
  2. 數(shù)據(jù)的更新操作在batch之間是強(qiáng)有序的。也就是說(shuō)，如果batch 2更新完成之前，batch 3不允許更新。

具備以上兩個(gè)條件之后，你就能實(shí)現(xiàn)有且僅有一次更新的目的。除了保存統(tǒng)計(jì)值之外，你還需要將txid也保存在數(shù)據(jù)庫(kù)中。當(dāng)更新數(shù)據(jù)時(shí)，你就可以將數(shù)據(jù)庫(kù)中保存的txid和當(dāng)前處理的batch的txid做比較。如果兩者相等，你應(yīng)該忽略該batch，因?yàn)閎atch之間的處理是強(qiáng)有序的，你能夠跟確定當(dāng)前batch在之前已經(jīng)被處理過(guò)了；如果兩者不一致，則表示當(dāng)前batch之前沒(méi)有處理過(guò)，你需要在數(shù)據(jù)庫(kù)中更新數(shù)據(jù)。

當(dāng)然，上面說(shuō)到的這些操作都不需要你自己實(shí)現(xiàn)，他們都已經(jīng)被Trident封裝在內(nèi)部的實(shí)現(xiàn)里，并且會(huì)自動(dòng)實(shí)現(xiàn)。如果你不想在數(shù)據(jù)庫(kù)中花費(fèi)外的空間去存儲(chǔ)txid，你可以不做。但是在這種情況下，Trident只能保證一個(gè)Tuple至少被處理一次，無(wú)法保證只被處理一次?？梢詤⒖?a target="_blank" rel="nofollow">這篇文檔解更多關(guān)于Trident State的知識(shí)。

State允許你使用任何策略來(lái)保存狀態(tài)。所以它可以將狀態(tài)保存在外部的數(shù)據(jù)庫(kù)，也可以保存在內(nèi)存中并備份到HDFS中（類(lèi)似于Hbase的工作模式）。State并不需要永久保存狀態(tài)，例如，你可以實(shí)現(xiàn)一個(gè)內(nèi)存版的State，僅僅保存最近的X個(gè)小時(shí)的數(shù)據(jù)，老數(shù)據(jù)直接丟棄?？梢詤⒄?a target="_blank" rel="nofollow">這個(gè)項(xiàng)目看看Memcached integration的實(shí)現(xiàn)。

Execution of Trident topologies

Trident的Topology會(huì)被編譯成效率最高的Storm Topology。只有在需要對(duì)數(shù)據(jù)進(jìn)行repartition的時(shí)候（如groupby或者shuffle）才會(huì)把tuple通過(guò)network發(fā)送出去，比如你有一個(gè)Trident Topology如下：



它將被編譯成如下的Storm Topology：


英文文檔原文地址：http://storm.apache.org/documentation/Trident-tutorial.html

如果你喜歡我的文章，請(qǐng)關(guān)注我的微信訂閱號(hào)：“機(jī)智的程序猿”，更多精彩，盡在其中：


最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 原文鏈接 譯者：魏勇 Trident 中含有對(duì)狀態(tài)化（stateful）的數(shù)據(jù)源進(jìn)行讀取和寫(xiě)入操作的一級(jí)抽象...
    Albert陳凱閱讀 576評(píng)論 0贊 1
  • 本文內(nèi)容部分來(lái)自Trident Tutorial。 Trident是基于Storm的實(shí)時(shí)計(jì)算模型的高級(jí)抽象。它可以...
    看山遠(yuǎn)兮閱讀 521評(píng)論 0贊 5
  • 聲明 本文首發(fā)于個(gè)人技術(shù)博客，轉(zhuǎn)載請(qǐng)注明出處，本文鏈接：http://qifuguang.me/2015/11/2...
    winwill2012閱讀 2,181評(píng)論 2贊 4
  • Storm入門(mén)系列之一：storm核心概念及特性 本文的將介紹一些 storm 入門(mén)的基礎(chǔ)知識(shí)，包括 storm ...
    zhaif閱讀 3,403評(píng)論 0贊 17
  • 本文參加#感悟三下鄉(xiāng)，青春筑夢(mèng)行#活動(dòng)，本人承諾，文章內(nèi)容為原創(chuàng)，且未在其他平臺(tái)發(fā)表過(guò)。古宇湖的水，云頂寨的風(fēng)，隆...
    Caimineral閱讀 617評(píng)論 0贊 0

友情鏈接更多精彩內(nèi)容