Storm 事務性拓撲

事務性拓撲

正如書中之前所提到的，使用 Storm 編程，可以通過調(diào)用 ack 和 fail 方法來確保一條消息的處理成功或失敗。不過當元組被重發(fā)時，會發(fā)生什么呢？你又該如何砍不會重復計算？

Storm0.7.0 實現(xiàn)了一個新特性——事務性拓撲，這一特性使消息在語義上確保你可以安全的方式重發(fā)消息，并保證它們只會被處理一次。在不支持事務性拓撲的情況下，你無法在準確性，可擴展性，以空錯性上得到保證的前提下完成計算。

NOTE:事務性拓撲是一個構(gòu)建于標準 Storm spout 和 bolt 之上的抽象概念。

設計

在事務性拓撲中，Storm 以并行和順序處理混合的方式處理元組。spout 并行分批創(chuàng)建供 bolt 處理的元組（譯者注：下文將這種分批創(chuàng)建、分批處理的元組稱做批次）。其中一些 bolt 作為提交者以嚴格有序的方式提交處理過的批次。這意味著如果你有每批五個元組的兩個批次，將有兩個元組被 bolt 并行處理，但是直到提交者成功提交了第一個元組之后，才會提交第二個元組。
NOTE： 使用事務性拓撲時，數(shù)據(jù)源要能夠重發(fā)批次，有時候甚至要重復多次。因此確認你的數(shù)據(jù)源——你連接到的那個 spout ——具備這個能力。 這個過程可以被描述為兩個階段： 處理階段 純并行階段，許多批次同時處理。 提交階段 嚴格有序階段，直到批次一成功提交之后，才會提交批次二。 這兩個階段合起來稱為一個 Storm 事務。 NOTE： Storm 使用 zookeeper 儲存事務元數(shù)據(jù)，默認情況下就是拓撲使用的那個 zookeeper。你可以修改以下兩個配置參數(shù)鍵指定其它的 zookeeper——transactional.zookeeper.servers 和transactional.zookeeper.port。

事務實踐

下面我們要創(chuàng)建一個 Twitter 分析工具來了解事務的工作方式。我們從一個 Redis 數(shù)據(jù)庫讀取tweets，通過幾個 bolt 處理它們，最后把結(jié)果保存在另一個 Redis 數(shù)據(jù)庫的列表中。處理結(jié)果就是所有話題和它們的在 tweets 中出現(xiàn)的次數(shù)列表，所有用戶和他們在 tweets 中出現(xiàn)的次數(shù)列表，還有一個包含發(fā)起話題和頻率的用戶列表。 這個工具的拓撲圖。

圖 拓撲概覽

正如你看到的，TweetsTransactionalSpout 會連接你的 tweet 數(shù)據(jù)庫并向拓撲分發(fā)批次。UserSplitterBolt 和 HashTagSplitterBolt 兩個 bolt，從 spout 接收元組。UserSplitterBolt 解析 tweets 并查找用戶——以 @ 開頭的單詞——然后把這些單詞分發(fā)到名為 users 的自定義數(shù)據(jù)流組。HashtagSplitterBolt 從 tweet 查找 # 開頭的單詞，并把它們分發(fā)到名為 hashtags 的自定義數(shù)據(jù)流組。第三個 bolt，UserHashtagJoinBolt，接收前面提到的兩個數(shù)據(jù)流組，并計算具名用戶的一條 tweet 內(nèi)的話題數(shù)量。為了計數(shù)并分發(fā)計算結(jié)果，這是個 BaseBatchBolt（稍后有更多介紹）。

最后一個 bolt——RedisCommitterBolt—— 接收以上三個 bolt 的數(shù)據(jù)流組。它為每樣東西計數(shù)，并在對一個批次完成處理時，把所有結(jié)果保存到 redis。這是一種特殊的 bolt，叫做提交者，在本章后面做更多講解。

用 TransactionalTopologyBuilder 構(gòu)建拓撲，代碼如下：

01
TransactionalTopologyBuilder builder=
02
    new TransactionalTopologyBuilder("test", "spout", new TweetsTransactionalSpout());
03

04
builder.setBolt("users-splitter", new UserSplitterBolt(), 4).shuffleGrouping("spout");
05
buildeer.setBolt("hashtag-splitter", new HashtagSplitterBolt(), 4).shuffleGrouping("spout");
06

07
builder.setBolt("users-hashtag-manager", new UserHashtagJoinBolt(), r)
08
       .fieldsGrouping("users-splitter", "users", new Fields("tweet_id"))
09
       .fieldsGrouping("hashtag-splitter", "hashtags", new Fields("tweet_id"));
10

11
builder.setBolt("redis-commiter", new RedisCommiterBolt())
12
       .globalGrouping("users-splitter", "users")
13
       .globalGrouping("hashtag-splitter", "hashtags")
14
       .globalGrouping("user-hashtag-merger");  

接下來就看看如何在一個事務性拓撲中實現(xiàn) spout。

Spout

一個事務性拓撲的 spout 與標準 spout 完全不同。

1
public class TweetsTransactionalSpout extends BaseTransactionalSpout<TransactionMetadata>{  

正如你在這個類定義中看到的，TweetsTransactionalSpout 繼承了帶范型的BaseTransactionalSpout。指定的范型類型的對象是事務元數(shù)據(jù)集合。它將在后面的代碼中用于從數(shù)據(jù)源分發(fā)批次。

在這個例子中，TransactionMetadata 定義如下：

01
public class TransactionMetadata implements Serializable {
02
    private static final long serialVersionUID = 1L;
03
    long from;
04
    int quantity;
05

06
    public TransactionMetadata(long from, int quantity) {
07
        this.from = from;
08
        this.quantity = quantity;
09
    }
10
}  

該類的對象維護著兩個屬性 from 和 quantity，它們用來生成批次。

spout 的最后需要實現(xiàn)下面的三個方法：

01
@Override
02
public ITransactionalSpout.Coordinator<TransactionMetadata> getCoordinator(
03
       Map conf, TopologyContext context) {
04
    return new TweetsTransactionalSpoutCoordinator();
05
}
06

07
@Override
08
public backtype.storm.transactional.ITransactionalSpout.Emitter<TransactionMetadata> getEmitter(Map conf, TopologyContext contest) {
09
    return new TweetsTransactionalSpoutEmitter();
10
}
11

12
@Override
13
public void declareOutputFields(OuputFieldsDeclarer declarer) {
14
    declarer.declare(new Fields("txid", "tweet_id", "tweet"));
15
}  

getCoordinator 方法，告訴 Storm 用來協(xié)調(diào)生成批次的類。getEmitter，負責讀取批次并把它們分發(fā)到拓撲中的數(shù)據(jù)流組。最后，就像之前做過的，需要聲明要分發(fā)的域。

RQ 類

為了讓例子簡單點，我們決定用一個類封裝所有對 Redis 的操作。

01
public class RQ {
02
    public static final String NEXT_READ = "NEXT_READ";
03
    public static final String NEXT_WRITE = "NEXT_WRITE";
04

05
    Jedis jedis;
06

07
    public RQ() {
08
        jedis = new Jedis("localhost");
09
    }
10

11
    public long getavailableToRead(long current) {
12
        return getNextWrite() - current;
13
    }
14

15
    public long getNextRead() {
16
        String sNextRead = jedis.get(NEXT_READ);
17
        if(sNextRead == null) {
18
            return 1;
19
        }
20
        return Long.valueOf(sNextRead);
21
    }
22

23
    public long getNextWrite() {
24
        return Long.valueOf(jedis.get(NEXT_WRITE));
25
    }
26

27
    public void close() {
28
        jedis.disconnect();
29
    }
30

31
    public void setNextRead(long nextRead) {
32
        jedis.set(NEXT_READ, ""+nextRead);
33
    }
34

35
    public List<String> getMessages(long from, int quantity) {
36
        String[] keys = new String[quantity];
37
        for (int i = 0; i < quantity; i++) {
38
            keys[i] = ""+(i+from);
39
        }
40
        return jedis.mget(keys);
41
    }
42
}  

仔細閱讀每個方法，確保自己理解了它們的用處。

協(xié)調(diào)者 Coordinator

下面是本例的協(xié)調(diào)者實現(xiàn)。

01
public static class TweetsTransactionalSpoutCoordinator implements ITransactionalSpout.Coordinator<TransactionMetadata> {
02
    TransactionMetadata lastTransactionMetadata;
03
    RQ rq = new RQ();
04
    long nextRead = 0;
05

06
    public TweetsTransactionalSpoutCoordinator() {
07
        nextRead = rq.getNextRead();
08
    }
09

10
    @Override
11
    public TransactionMetadata initializeTransaction(BigInteger txid, TransactionMetadata prevMetadata) {
12
        long quantity = rq.getAvailableToRead(nextRead);
13
        quantity = quantity > MAX_TRANSACTION_SIZE ? MAX_TRANSACTION_SIZE : quantity;
14
        TransactionMetadata ret = new TransactionMetadata(nextRead, (int)quantity);
15
        nextRead += quantity;
16
        return ret;
17
    }
18

19
    @Override
20
    public boolean isReady() {
21
        return rq.getAvailableToRead(nextRead) > 0;
22
    }
23

24
    @Override
25
    public void close() {
26
        rq.close();
27
    }
28
}  

值得一提的是，在整個拓撲中只會有一個提交者實例。創(chuàng)建提交者實例時，它會從 redis 讀取一個從1開始的序列號，這個序列號標識要讀取的 tweet 下一條。

第一個方法是 isReady。在 initializeTransaction 之前調(diào)用它確認數(shù)據(jù)源已就緒并可讀取。此方法應當相應的返回 true 或 false。在此例中，讀取 tweets 數(shù)量并與已讀數(shù)量比較。它們之間的不同就在于可讀 tweets 數(shù)。如果它大于0，就意味著還有 tweets 未讀。

最后，執(zhí)行 initializeTransaction。正如你看到的，它接收 txid 和 prevMetadata作為參數(shù)。第一個參數(shù)是 Storm 生成的事務 ID，作為批次的惟一性標識。prevMetadata 是協(xié)調(diào)器生成的前一個事務元數(shù)據(jù)對象。

在這個例子中，首先確認有多少 tweets 可讀。只要確認了這一點，就創(chuàng)建一個TransactionMetadata 對象，標識讀取的第一個 tweet（譯者注：對象屬性 from ），以及讀取的 tweets 數(shù)量（譯者注：對象屬性 quantity ）。

元數(shù)據(jù)對象一經(jīng)返回，Storm 把它跟 txid 一起保存在 zookeeper。這樣就確保了一旦發(fā)生故障，Storm 可以利用分發(fā)器(譯者注：Emitter，見下文)重新發(fā)送批次。

Emitter

創(chuàng)建事務性 spout 的最后一步是實現(xiàn)分發(fā)器（Emitter）。實現(xiàn)如下：

01
public static class TweetsTransactionalSpoutEmitter implements ITransactionalSpout.Emitter<TransactionMetadata> {
02

03
</pre>
04
<pre>    RQ rq = new RQ();</pre>
05
<pre>    public TweetsTransactionalSpoutEmitter() {}</pre>
06
<pre>    @Override
07
    public void emitBatch(TransactionAttempt tx, TransactionMetadata coordinatorMeta, BatchOutputCollector collector) {
08
        rq.setNextRead(coordinatorMeta.from+coordinatorMeta.quantity);
09
        List<String> messages = rq.getMessages(coordinatorMeta.from, <span style="font-family: Georgia, 'Times New Roman', 'Bitstream Charter', Times, serif; font-size: 13px; line-height: 19px;">coordinatorMeta.quantity);
10
</span>        long tweetId = coordinatorMeta.from;
11
        for (String message : messages) {
12
            collector.emit(new Values(tx, ""+tweetId, message));
13
            tweetId++;
14
        }
15
    }
16

17
    @Override
18
    public void cleanupBefore(BigInteger txid) {}
19

20
    @Override
21
    public void close() {
22
        rq.close();
23
    }</pre>
24
<pre>
25
}  

分發(fā)器從數(shù)據(jù)源讀取數(shù)據(jù)并從數(shù)據(jù)流組發(fā)送數(shù)據(jù)。分發(fā)器應當問題能夠為相同的事務 id 和事務元數(shù)據(jù)發(fā)送相同的批次。這樣，如果在處理批次的過程中發(fā)生了故障，Storm 就能夠利用分發(fā)器重復相同的事務 id 和事務元數(shù)據(jù)，并確保批次已經(jīng)重復過了。Storm 會在TransactionAttempt 對象里為嘗試次數(shù)增加計數(shù)（譯者注：attempt id ）。這樣就能知道批次已經(jīng)重復過了。

在這里 emitBatch 是個重要方法。在這個方法中，使用傳入的元數(shù)據(jù)對象從 redis 得到tweets，同時增加 redis 維持的已讀 tweets 數(shù)。當然它還會把讀到的 tweets 分發(fā)到拓撲。

Bolts

首先看一下這個拓撲中的標準 bolt：

01
public class UserSplitterBolt implements IBasicBolt{
02
    private static final long serialVersionUID = 1L;
03

04
    @Override
05
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
06
        declarer.declareStream("users", new Fields("txid","tweet_id","user"));
07
    }
08

09
    @Override
10
    public Map<String, Object> getComponentConfiguration() {
11
        return null;
12
    }
13

14
    @Override
15
    public void prepare(Map stormConf, TopologyContext context) {}
16

17
    @Override
18
    public void execute(Tuple input, BasicOutputCollector collector) {
19
        String tweet = input.getStringByField("tweet");
20
        String tweetId = input.getStringByField("tweet_id");
21
        StringTokenizer strTok = new StringTokenizer(tweet, " ");
22
        HashSet<String> users = new HashSet<String>();
23

24
        while(strTok.hasMoreTokens()) {
25
            String user = strTok.nextToken();
26

27
            //確保這是個真實的用戶，并且在這個tweet中沒有重復
28
            if(user.startsWith("@") && !users.contains(user)) {
29
                collector.emit("users", new Values(tx, tweetId, user));
30
                users.add(user);
31
            }
32
        }
33
    }
34

35
    @Override
36
    public void cleanup(){}
37
}  

正如本章前面提到的，UserSplitterBolt 接收元組，解析 tweet 文本，分發(fā) @ 開頭的單詞————tweeter 用戶。HashtagSplitterBolt 的實現(xiàn)也非常相似。

01
public class HashtagSplitterBolt implements IBasicBolt{
02
    private static final long serialVersionUID = 1L;
03

04
    @Override
05
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
06
        declarer.declareStream("hashtags", new Fields("txid","tweet_id","hashtag"));
07
    }
08

09
    @Override
10
    public Map<String, Object> getComponentConfiguration() {
11
        return null;
12
    }
13

14
    @Override
15
    public void prepare(Map stormConf, TopologyContext context) {}
16

17
    @Oerride
18
    public void execute(Tuple input, BasicOutputCollector collector) {
19
        String tweet = input.getStringByField("tweet");
20
        String tweetId = input.getStringByField("tweet_id");
21
        StringTokenizer strTok = new StringTokenizer(tweet, " ");
22
        TransactionAttempt tx = (TransactionAttempt)input.getValueByField("txid");
23
        HashSet<String> words = new HashSet<String>();
24

25
        while(strTok.hasMoreTokens()) {
26
            String word = strTok.nextToken();
27

28
            if(word.startsWith("#") && !words.contains(word)){
29
                collector.emit("hashtags", new Values(tx, tweetId, word));
30
                words.add(word);
31
            }
32
        }
33
    }
34

35
    @Override
36
    public void cleanup(){}
37
}  

現(xiàn)在看看 UserHashTagJoinBolt 的實現(xiàn)。首先要注意的是它是一個 BaseBatchBolt。這意味著，execute 方法會操作接收到的元組，但是不會分發(fā)新的元組。批次完成時，Storm 會調(diào)用 finishBatch 方法。

01
public void execute(Tuple tuple) {
02
    String source = tuple.getSourceStreamId();
03
    String tweetId = tuple.getStringByField("tweet_id");
04

05
    if("hashtags".equals(source)) {
06
        String hashtag = tuple.getStringByField("hashtag");
07
        add(tweetHashtags, tweetId, hashtag);
08
    } else if("users".equals(source)) {
09
        String user = tuple.getStringByField("user");
10
        add(userTweets, user, tweetId);
11
    }
12
}  

既然要結(jié)合 tweet 中提到的用戶為出現(xiàn)的所有話題計數(shù)，就需要加入前面的 bolts 創(chuàng)建的兩個數(shù)據(jù)流組。這件事要以批次為單位進程，在批次處理完成時，調(diào)用 finishBatch 方法。

01
@Override
02
public void finishBatch() {
03
    for(String user:userTweets.keySet()){
04
        Set<String> tweets = getUserTweets(user);
05
        HashMap<String, Integer> hashtagsCounter = new HashMap<String, Integer>();
06
        for(String tweet:tweets){
07
            Set<String> hashtags=getTweetHashtags(tweet);
08
            if(hashtags!=null){
09
                for(String hashtag:hashtags){
10
                    Integer count=hashtagsCounter.get(hashtag);
11
                    if(count==null){count=0;}
12
                    count++;
13
                    hashtagsCounter.put(hashtag,count);
14
                }
15
            }
16
        }
17
        for(String hashtag:hashtagsCounter.keySet()){
18
            int count=hashtagsCounter.get(hashtag);
19
            collector.emit(new Values(id,user,hashtag,count));
20
        }
21
    }
22
}  

這個方法計算每對用戶-話題出現(xiàn)的次數(shù)，并為之生成和分發(fā)元組。

你可以在 GitHub 上找到并下載完整代碼。（譯者注：https://github.com/storm-book/examples-ch08-transactional-topologies 這個倉庫里沒有代碼，誰知道哪里有代碼麻煩說一聲。）

提交者 bolts

我們已經(jīng)學習了，批次通過協(xié)調(diào)器和分發(fā)器怎樣在拓撲中傳遞。在拓撲中，這些批次中的元組以并行的，沒有特定次序的方式處理。

協(xié)調(diào)者 bolts 是一類特殊的批處理 bolts，它們實現(xiàn)了 IComh mitter 或者通過TransactionalTopologyBuilder 調(diào)用 setCommiterBolt 設置了提交者 bolt。它們與其它的批處理 bolts 最大的不同在于，提交者 bolts的finishBatch 方法在提交就緒時執(zhí)行。這一點發(fā)生在之前所有事務都已成功提交之后。另外，finishBatch 方法是順序執(zhí)行的。因此如果同時有事務 ID1 和事務 ID2 兩個事務同時執(zhí)行，只有在 ID1 沒有任何差錯的執(zhí)行了 finishBatch 方法之后，ID2 才會執(zhí)行該方法。

下面是這個類的實現(xiàn)

01
public class RedisCommiterCommiterBolt extends BaseTransactionalBolt implements ICommitter {
02
    public static final String LAST_COMMITED_TRANSACTION_FIELD = "LAST_COMMIT";
03
    TransactionAttempt id;
04
    BatchOutputCollector collector;
05
    Jedis jedis;
06

07
    @Override
08
    public void prepare(Map conf, TopologyContext context,
09
                        BatchOutputCollector collector, TransactionAttempt id) {
10
        this.id = id;
11
        this.collector = collector;
12
        this.jedis = new Jedis("localhost");
13
    }
14

15
    HashMap<String, Long> hashtags = new HashMap<String,Long>();
16
    HashMap<String, Long> users = new HashMap<String, Long>();
17
    HashMap<String, Long> usersHashtags = new HashMap<String, Long>();
18

19
    private void count(HashMap<String, Long> map, String key, int count) {
20
        Long value = map.get(key);
21
        if(value == null){value = (long)0;}
22
        value += count;
23
        map.put(key,value);
24
    }
25

26
    @Override
27
    public void execute(Tuple tuple) {
28
        String origin = tuple. getSourceComponent();
29
        if("sers-splitter".equals(origin)) {
30
            String user = tuple.getStringByField("user");
31
            count(users, user, 1);
32
        } else if("hashtag-splitter".equals(origin)) {
33
            String hashtag = tuple.getStringByField("hashtag");
34
            count(hashtags, hashtag, 1);
35
        } else if("user-hashtag-merger".quals(origin)) {
36
            String hashtag = tuple.getStringByField("hashtag");
37
            String user = tuple.getStringByField("user");
38
            String key = user + ":" + hashtag;
39
            Integer count = tuple.getIntegerByField("count");
40
            count(usersHashtags, key, count);
41
        }
42
    }
43

44
    @Override
45
    public void finishBatch() {
46
        String lastCommitedTransaction = jedis.get(LAST_COMMITED_TRANSACTION_FIELD);
47
        String currentTransaction = ""+id.getTransactionId();
48

49
        if(currentTransaction.equals(lastCommitedTransaction)) {return;}
50

51
        Transaction multi = jedis.multi();
52

53
        multi.set(LAST_COMMITED_TRANSACTION_FIELD, currentTransaction);
54

55
        Set<String> keys = hashtags.keySet();
56
        for (String hashtag : keys) {
57
            Long count = hashtags.get(hashtag);
58
            multi.hincrBy("hashtags", hashtag, count);
59
        }
60

61
        keys = users.keySet();
62
        for (String user : keys) {
63
            Long count =users.get(user);
64
            multi.hincrBy("users",user,count);
65
        }
66

67
        keys = usersHashtags.keySet();
68
        for (String key : keys) {
69
            Long count = usersHashtags.get(key);
70
            multi.hincrBy("users_hashtags", key, count);
71
        }
72

73
        multi.exec();
74
    }
75

76
    @Override
77
    public void declareOutputFields(OutputFieldsDeclarer declarer) {}
78
}  

這個實現(xiàn)很簡單，但是在 finishBatch 有一個細節(jié)。

1
...
2
multi.set(LAST_COMMITED_TRANSACTION_FIELD, currentTransaction);
3
...  

在這里向數(shù)據(jù)庫保存提交的最后一個事務 ID。為什么要這樣做？記住，如果事務失敗了，Storm將會盡可能多的重復必要的次數(shù)。如果你不確定已經(jīng)處理了這個事務，你就會多算，事務拓撲也就沒有用了。所以請記?。罕４孀詈筇峤坏氖聞?ID，并在提交前檢查。

分區(qū)的事務 Spouts

對一個 spout 來說，從一個分區(qū)集合中讀取批次是很普通的。接著這個例子，你可能有很多redis 數(shù)據(jù)庫，而 tweets 可能會分別保存在這些 redis 數(shù)據(jù)庫里。通過實現(xiàn)IPartitionedTransactionalSpout，Storm 提供了一些工具用來管理每個分區(qū)的狀態(tài)并保證重播的能力。

下面我們修改 TweetsTransactionalSpout，使它可以處理數(shù)據(jù)分區(qū)。

首先，繼承 BasePartitionedTransactionalSpout，它實現(xiàn)了IPartitionedTransactionalSpout。

1
public class TweetsPartitionedTransactionalSpout extends
2
       BasePartitionedTransactionalSpout<TransactionMetadata> {
3
...
4
}  

然后告訴 Storm 誰是你的協(xié)調(diào)器。

01
public static class TweetsPartitionedTransactionalCoordinator implements Coordinator {
02
    @Override
03
    public int numPartitions() {
04
        return 4;
05
    }
06

07
    @Override
08
    public boolean isReady() {
09
        return true;
10
    }
11

12
    @Override
13
    public void close() {}
14
}  

在這個例子里，協(xié)調(diào)器很簡單。numPartitions 方法，告訴 Storm 一共有多少分區(qū)。而且你要注意，不要返回任何元數(shù)據(jù)。對于 IPartitionedTransactionalSpout，元數(shù)據(jù)由分發(fā)器直接管理。

下面是分發(fā)器的實現(xiàn)：

01
public static class TweetsPartitionedTransactionalEmitter
02
       implements Emitter<TransactionMetadata> {
03
    PartitionedRQ rq = new ParttionedRQ();
04

05
    @Override
06
    public TransactionMetadata emitPartitionBatchNew(TransactionAttempt tx,
07
            BatchOutputCollector collector, int partition,
08
            TransactionMetadata lastPartitioonMeta) {
09
        long nextRead;
10

11
        if(lastPartitionMeta == null) {
12
            nextRead = rq.getNextRead(partition);
13
        }else{
14
            nextRead = lastPartitionMeta.from + lastPartitionMeta.quantity;
15
            rq.setNextRead(partition, nextRead); //移動游標
16
        }
17

18
        long quantity = rq.getAvailableToRead(partition, nextRead);
19
        quantity = quantity > MAX_TRANSACTION_SIZE ? MAX_TRANSACTION_SIZE : quantity;
20
        TransactionMetadata metadata = new TransactionMetadata(nextRead, (int)quantity);
21

22
        emitPartitionBatch(tx, collector, partition, metadata);
23
        return metadata;
24
    }
25

26
    @Override
27
    public void emitPartitionBatch(TransactionAttempt tx, BatchOutputCollector collector,
28
            int partition, TransactionMetadata partitionMeta) {
29
        if(partitionMeta.quantity <= 0){
30
            return;
31
        }
32

33
        List<String> messages = rq.getMessages(partition, partitionMeta.from,
34
               partitionMeta.quantity);
35

36
        long tweetId = partitionMeta.from;
37
        for (String msg : messages) {
38
            collector.emit(new Values(tx, ""+tweetId, msg));
39
            tweetId++;
40
        }
41
    }
42

43
    @Override
44
    public void close() {}
45
}  

這里有兩個重要的方法，emitPartitionBatchNew，和 emitPartitionBatch。對于 emitPartitionBatchNew，從 Storm 接收分區(qū)參數(shù)，該參數(shù)決定應該從哪個分區(qū)讀取批次。在這個方法中，決定獲取哪些 tweets，生成相應的元數(shù)據(jù)對象，調(diào)用 emitPartitionBatch，返回元數(shù)據(jù)對象，并且元數(shù)據(jù)對象會在方法返回時立即保存到 zookeeper。

Storm 會為每一個分區(qū)發(fā)送相同的事務 ID，表示一個事務貫穿了所有數(shù)據(jù)分區(qū)。通過emitPartitionBatch 讀取分區(qū)中的 tweets，并向拓撲分發(fā)批次。如果批次處理失敗了，Storm 將會調(diào)用 emitPartitionBatch 利用保存下來的元數(shù)據(jù)重復這個批次。

NOTE: 完整的源碼請見：https://github.com/storm-book/examples-ch08-transactional-topologies（譯者注：原文如此，實際上這個倉庫里什么也沒有）

模糊的事務性拓撲

到目前為止，你可能已經(jīng)學會了如何讓擁有相同事務 ID 的批次在出錯時重播。但是在有些場景下這樣做可能就不太合適了。然后會發(fā)生什么呢？

事實證明，你仍然可以實現(xiàn)在語義上精確的事務，不過這需要更多的開發(fā)工作，你要記錄由 Storm 重復的事務之前的狀態(tài)。既然能在不同時刻為相同的事務 ID 得到不同的元組，你就需要把事務重置到之前的狀態(tài)，并從那里繼續(xù)。

比如說，如果你為收到的所有 tweets 計數(shù)，你已數(shù)到5，而最后的事務 ID 是321，這時你多數(shù)了8個。你要維護以下三個值 ——previousCount=5,currentCount=13，以及l(fā)astTransactionId=321。假設事物 ID321 又發(fā)分了一次，而你又得到了4個元組，而不是之前的8個，提交器會探測到這是相同的事務 ID，它將會把結(jié)果重置到 previousCount 的值5，并在此基礎上加4，然后更新 currentCount 為9。

另外，在之前的一個事務被取消時，每個并行處理的事務都要被取消。這是為了確保你沒有丟失任何數(shù)據(jù)。

你的 spout 可以實現(xiàn) IOpaquePartitionedTransactionalSpout，而且正如你看到的，協(xié)調(diào)器和分發(fā)器也很簡單。

01
public static class TweetsOpaquePartitionedTransactionalSpoutCoordinator implements IOpaquePartitionedTransactionalSpout.Coordinator {
02
    @Override
03
    public boolean isReady() {
04
        return true;
05
    }
06
}
07

08
public static class TweetsOpaquePartitionedTransactionalSpoutEmitter
09
       implements IOpaquePartitionedTransactionalSpout.Emitter<TransactionMetadata> {
10
    PartitionedRQ rq  = new PartitionedRQ();
11

12
    @Override
13
    public TransactionMetadata emitPartitionBatch(TransactionAttempt tx,
14
           BatchOutputCollector collector, int partion,
15
           TransactionMetadata lastPartitonMeta) {
16
        long nextRead;
17

18
        if(lastPartitionMeta == null) {
19
            nextRead = rq.getNextRead(partition);
20
        }else{
21
            nextRead = lastPartitionMeta.from + lastPartitionMeta.quantity;
22
            rq.setNextRead(partition, nextRead);//移動游標
23
        }
24

25
        long quantity = rq.getAvailabletoRead(partition, nextRead);
26
        quantity = quantity > MAX_TRANSACTION_SIZE ? MAX_TRANSACTION_SIZE : quantity;
27
        TransactionMetadata metadata = new TransactionMetadata(nextRead, (int)quantity);
28
        emitMessages(tx, collector, partition, metadata);
29
        return metadata;
30
    }
31

32
    private void emitMessage(TransactionAttempt tx, BatchOutputCollector collector,
33
                 int partition, TransactionMetadata partitionMeta) {
34
        if(partitionMeta.quantity <= 0){return;}
35

36
        List<String> messages = rq.getMessages(partition, partitionMeta.from, partitionMeta.quantity);
37
        long tweetId = partitionMeta.from;
38
        for(String msg : messages) {
39
            collector.emit(new Values(tx, ""+tweetId, msg));
40
            tweetId++;
41
        }
42
    }
43

44
    @Override
45
    public int numPartitions() {
46
        return 4;
47
    }
48

49
    @Override
50
    public void close() {}
51
}  

最有趣的方法是 emitPartitionBatch，它獲取之前提交的元數(shù)據(jù)。你要用它生成批次。這個批次不需要與之前的那個一致，你可能根本無法創(chuàng)建完全一樣的批次。剩余的工作由提交器 bolts借助之前的狀態(tài)完成。