2015-12-02 68 views
0

我是新來的風暴,所以請忍受我。(卡夫卡)噴口風暴返回的東西清單,不能傳播此列表螺栓

我有一項服務,每當我們與結果列表打電話時,都會向卡夫卡發送一條消息。

有一個KafkaSpout讀取每一條消息,消息中包含beforementioned列表。

這只是一個JSON,我可以解開它。現在,這裏的問題:

我在傑克遜計劃做這個轉換操作,但是這個計劃可以基本上返回一個Values對象,它不是一個對象列表,但基本上是一個字段值對列表。

另一件事可能是一個博爾特這個值(這只是一個擴展的ArrayList)對象獲取並在單一元素解開它,並給他們的每一個到下一個博爾特。這是一個解決方案嗎?我可以從一次呼叫發射多個物體到一個螺栓嗎?

是否有更智能的解決方案?

回答

2

是的,你可以發出從單一博爾特多個元組。在位於here

public void execute(Tuple tuple) { 
    String sentence = tuple.getString(0); 
    for(String word: sentence.split(" ")) { 
    _collector.emit(tuple, new Values(word)); //emits multiple tuples 
    } 
    _collector.ack(tuple); 
} 

正如可以看到for循環可以同時發射的多個元組的示例螺栓通過執行方法尋找。這樣做會創建一個更大的消息樹。這可能會導致問題,這取決於您的可靠性保證和數據大小。

根據我的經驗,很難/不可能操縱KafkaSpout裏面的數據。所以有幾個關於你的設置的筆記。

  • 我想要做的第一件事就是改變服務發送給卡夫卡的內容。你可以發送它在單個項目而不是一個大項目。如果你無法改變這一點。
  • 像你剛纔提到的那樣設置多個Bolts,Spout => UnwrapBolt => ProcessBolt,其中UnwrapBolt將你的一個數據源作爲單獨的元組發送出去,然後ProcessBolt獲取每個單獨的元組,然後處理它們,但是你需要。
+0

謝謝morganw09dev。我實際上使用了第一種方法,我只用一條消息減少了消息傳遞。一個問題:如果我發出每個元組而不是最後一個元組,會發生什麼? – dierre

+1

我不知道這是可能的/可取的嗎???我的解釋一直是,通常使用aple元組來表示你已經完成了爲特定的Bolt處理這個元組。所以,當你在同一個Bolt中使用它時,確定一個元組沒有什麼意義。我不知道這個觀點是如何官方的。 – morganw09dev

+0

哦,好的。我會嘗試,我會讓你知道發生了什麼:) – dierre