storm Transactional spouts有哪些特性

1、有著同樣txid的batch一定是一樣的。當(dāng)重播一個txid對應(yīng)的batch時，一定會重播和之前對應(yīng)txid的batch中同樣的tuples。

2、各個batch之間是沒有交集的。每個tuple只能屬于一個batch

3、每一個tuple都屬于一個batch，無一例外

這是一類非常容易理解的spout， tuple 流被劃分為固定的batch并且永不改變。（trident-kafka 有一個 transactional spout 的實現(xiàn)。）

你也許會問：為什么我們不總是使用transactional spout？這很容易理解。一個原因是并不是所有的地方都需要容錯的。舉例來說，TransactionalTridentKafkaSpout 工作的方式是一個batch包含的tuple來自某個kafka topic中的所有partition。一旦這個batch被發(fā)出，在任何時候如果這個batch被重新發(fā)出時，它必須包含原來所有的tuple以滿足 transactional spout的語義?，F(xiàn)在我們假定一個batch被TransactionalTridentKafkaSpout所發(fā)出，這個batch沒有被成功處理，并且同時kafka的一個節(jié)點也down掉了。你就無法像之前一樣重播一個完全一樣的batch（因為kakfa的節(jié)點down掉，該topic的一部分partition可能會無法使用），整個處理會被中斷。

這也就是"opaque transactional" spouts（不透明事務(wù)spout）存在的原因 - 他們對于丟失源節(jié)點這種情況是容錯的，仍然能夠幫你達(dá)到有且只有一次處理的語義。后面會對這種spout有所介紹。

在討論"opaque transactional" spout之前，我們先來看看怎樣為transactional spout設(shè)計一個具有exactly-once語義的State實現(xiàn)。這個State的類型是"transactionalstate" 并且它利用了任何一個txid總是對應(yīng)同樣的tuple序列這個語義。

假如說你有一個用來計算單詞出現(xiàn)次數(shù)的topology，你想要將單詞的出現(xiàn)次數(shù)以key/value對的形式存儲到數(shù)據(jù)庫中。key就是單詞，value就是這個這個單詞出現(xiàn)的次數(shù)。你已經(jīng)看到只是存儲一個數(shù)量是不足以知道你是否已經(jīng)處理過一個batch的。你可以通過將value和txid一起存儲到數(shù)據(jù)庫中。這樣的話，當(dāng)更新這個count之前，你可以先去比較數(shù)據(jù)庫中存儲的txid和現(xiàn)在要存儲的txid。如果一樣，就跳過什么都不做，因為這個value之前已經(jīng)被處理過了。如果不一樣，就執(zhí)行存儲。這個邏輯可以工作的前提就是txid永不改變，并且Trident保證狀態(tài)的更新是在batch之間嚴(yán)格順序進(jìn)行的。

考慮下面這個例子的運行邏輯，假定你在處理一個txid為3的包含下面tuple的batch：

["man"]

["man"]

["dog"]

假定數(shù)據(jù)庫中當(dāng)前保存了下面這樣的key/value 對：

man => [count=3, txid=1]

dog => [count=4, txid=3]

apple => [count=10, txid=2]

單詞“man”對應(yīng)的txid是1. 因為當(dāng)前的txid是3，你可以確定你還沒有為這個batch中的tuple更新過這個單詞的數(shù)量。所以你可以放心的給count加2并更新txid為3. 與此同時，單詞“dog”的txid和當(dāng)前的txid是相同的，因此你可以跳過這次更新。此時數(shù)據(jù)庫中的數(shù)據(jù)如下：

man => [count=5, txid=3]

dog => [count=4, txid=3]

apple => [count=10, txid=2]