Storm概念學習系列之Tuple元組（數據載體）

 

　　很少說，直接上乾貨！

 

 

Tuple元組

　　Tuple 是 Storm 的主要數據結構，而且是 Storm 中使用的最基本單元、數據模型和元組。

 

　　　　　　

 

 

 

 

 

 

Tuple 描述

　　Tuple 就是一個值列表， Tuple 中的值能夠是任何類型的，動態類型的Tuple的fields能夠不用聲明；默認狀況下，Storm中的Tuple支持私有類型、字符串、字節數組等做爲它的字段值，若是使用其餘類型，就須要序列化該類型。

　　Tuple的字段默認類型有 ：  integer、 float、 double、 long、short、 string、 byte、 binary（byte[]）。

 　　

　　Tuple元組，是消息傳遞的基本單元，是一個命名的值列表，元組中的字段能夠是任何類型的對象。Storm使用元組做爲其數據模型，元組支持全部的基本類型、字符串和字節數組做爲字段值，只要實現類型的序列化接口就可使用該類型的對象。

　　元組原本應該是一個key-value的Map，可是因爲各個組件間傳遞的元組的字段名稱已經事先定義好，因此只要按序把元組填入各個value便可，因此元組是一個vlue的List。

 

 

　　Tuple是Storm採用的數據表示模型，全部的數據都以Tuple的形式在各個組件之間流動。Tuple是一組字段列表，每一個字段由一個字段名和字段值組成，每一個Tuple相似於數據庫中的一行記錄。在默認的狀況下，Tuple的字段類型能夠是integer、long、short、byte、string、double、float、boolean和byte array。固然，你也能夠經過實現序列化器自定義類型。

 

 

　　Tuple 數據結構如圖 1 所示。

　　　　　　　　　　　　　　　　　　　　　　　　

　　　　　　　　　　　　　　　　　　　　　　　　　　   圖 1 Tuple 數據結構

 

 　　Tuple 能夠理解成鍵值對。例如，建立一個Bolt 要發送兩個字段（命名爲 double 和 triple），其中鍵就是定義在declareOutputFields 方法中的 Fields 對象，值就是在 emit 方法中發送的 Values 對象。

 

 

 

　　如下是一個簡單例子

public class DoubleAndTripleBolt extends BaseRichBolt {
OutputCollectorBase _collector;

@Override
public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
_collector = collector;
}

@Override
public void execute(Tuple input) {
int val = input.getInteger(0);
_collector.emit(input, new Values(val*2, val*3));
_collector.ack(input);
}

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

 

 

　　此外，在使用的 Storm Java 包中， backtype.storm.tuple 主要有如下幾個類：

Fileds.class
MessageId.class
Tuple.class
TupleImpl.class
Values.class

　　列出以上內容是爲了更好地理解 Tuple，這樣可以從本質上理解 Tuple，在使用時更加駕輕就熟。

 

 

 

 

Tuple 的生命週期

　　瞭解一個 Tuple 的生命週期就須要查看源碼，以下的 Java 代碼展現了 Spout（消息源）接口發出 Tuple（消息）的整個過程。

public interface ISpout extends Serializable {
void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
void nextTuple();
void ack(Object msgId);
void fail(Object msgId);
void close();
}

　　首 先， Storm 調 用 Spout（消息源）的nextTuple 方法來獲取下一個Tuple， Spout經過Open 方法的參數提供的SpoutOutputCollector將新Tuple發射到其中一個輸出消息流。

　　　　注意：發射Tuple 時， Spout提供一個message-id，經過這個ID 來追蹤該Tuple。

 　　接下來， Storm跟蹤該Tuple的樹形結構是否成功建立，並根據 messageid調用Spout中的ack函數，以確認Tuple是否被徹底處理。若是Tuple超時，則調用 Spout 的 fail 方法。

　　由此看出，同一個Tuple無論是acked，仍是failed都是由建立它的Spout發出並維護的，因此，即便Spout 在集羣環境中同時執行不少的任務，該Tuple 也不會被其餘任務調用或生成 acked或 failed 狀態。總之， Storm會利用內部的 Acker 機制保證每一個Tuple 被可靠地處理。最後，在任務完成後，Spout調用Close方法結束 Tuple 的使命。

 

　　好比