Storm系列(五)DRPC實現遠端呼叫
Trident簡介
- Trident擁有一流的抽象，可以讀取和寫入有狀態的來源。狀態可以是拓撲的內部 - 例如，儲存在記憶體中並由HDFS支援 - 或者外部儲存在Memcached或Cassandra等資料庫中。在任何一種情況下，Trident API都沒有區別。
- Trident以容錯的方式管理狀態，以便狀態更新在重試和失敗時是冪等的。這使您可以推理Trident拓撲，就好像每條訊息都是精確處理一次一樣。
- 在進行狀態更新時，可以實現各種級別的容錯
例子說明
假設您正在對流進行計數聚合，並希望將執行計數儲存在資料庫中。現在假設您在資料庫中儲存了一個表示計數的值，並且每次處理新元組時都會增加計數。
發生故障時，將重發送元組。這會在執行狀態更新（或任何帶有副作用的事物）時出現問題 - 您不知道以前是否曾基於此元組成功更新狀態。也許你以前從未處理過元組，在這種情況下你應該增加計數。也許你已經處理了元組併成功遞增了計數，但是元組在另一個步驟中處理失敗。在這種情況下，您不應增加計數。或許您之前看過元組但在更新資料庫時出錯。在這種情況下，您應該更新資料庫。
只需將計數儲存在資料庫中，您就不知道之前是否已經處理過這個元組。因此，您需要更多資訊才能做出正確的決定。Trident提供以下語義，足以實現一次性處理語義：
- 元組作為小批量處理
- 每批元組都有一個稱為“事務ID”（txid）的唯一ID。如果批量重播，則給出完全相同的txid
- 批次之間訂購狀態更新。也就是說，在批處理2的狀態更新成功之前，不會應用批處理3的狀態更新。
使用這些原語，您的State實現可以檢測之前是否已經處理了一批元組，並採取適當的操作以一致的方式更新狀態。您採取的操作取決於輸入splot提供的確切語義，即每批中的內容。在容錯方面有三種可能的splot：“非事務性”，“事務性”和“不透明事務性”。同樣，在容錯方面有三種可能的狀態：“非事務性”，“事務性”和“不透明事務性”。讓我們來看看每個splot型別，看看每種噴口可以達到什麼樣的容錯能力。
Transactional spout(事物性spouts)
請記住，Trident將元組作為小批量處理，每個批次都被賦予唯一的事務ID。spout的屬性根據它們可以提供的關於每批中的含量的保證而變化。事務性spout具有以下屬性：
- 給定txid的批次始終相同。對txid進行批量重放將與第一次為該txid發出批次完全相同的元組集。
- 批處理元組之間沒有重疊（元組是一批或另一批，從不多元組）。
- 每個元組都是一個批處理（沒有跳過元組） 這是一個非常容易理解的事物性spout，將流分為不變的固定批次。Storm 為Kafka 實施了一個事務spout。
為什麼不總是使用事務性spout？ 它們簡單易懂。您可能不使用它的一個原因是因為它們不一定非常容錯。例如，TransactionalTridentKafkaSpout的工作方式是txid的批處理將包含來自主題的所有Kafka分割槽的元組。一旦批次被髮出，那麼在將來重新發出批次的任何時候都必須發出完全相同的元組集合以滿足事務性噴口的語義。現在假設從TransactionalTridentKafkaSpout發出批處理，批處理無法處理，同時其中一個Kafka節點發生故障。您現在無法重播與之前相同的批次（因為節點已關閉且主題的某些分割槽不可用），
這就是存在“不透明事務”spout的原因 - 它們對丟失源節點具有容錯能力，同時仍允許您實現一次性處理語義。
（一方面注意 - 一旦Kafka支援複製，就有可能擁有對節點故障具有容錯能力的事務性spout，但該功能尚不存在。）
假設您的拓撲計算字數，並且您希望將字數儲存在鍵/值資料庫中。鍵將是單詞，值將包含計數。您已經看到只儲存計數，因為該值不足以知道您之前是否處理過一批元組。相反，您可以做的是將事務id與資料庫中的count一起儲存為原子值。然後，在更新計數時，您只需將資料庫中的事務ID與當前批次的事務ID進行比較。如果它們是相同的，則跳過更新 - 由於強大的排序，您確定資料庫中的值包含當前批次。如果它們不同，則增加計數。這個邏輯有效，因為txid的批處理永遠不會改變，
假設您正在處理由以下一批元組組成的txid 3：
["man"]
["man"]
["dog"]
複製程式碼假設資料庫當前包含以下鍵/值對：
man => [count=3, txid=1]
dog => [count=4, txid=3]
apple => [count=10, txid=2]
複製程式碼與“man”關聯的txid為txid 1.由於當前txid為3，因此您確定該批次中未表示此批元組。因此，您可以繼續將計數增加2並更新txid。另一方面，“dog”的txid與當前的txid相同。因此，您確定已知當前批次的增量已在資料庫中表示為“dog”鍵。所以你可以跳過更新。完成更新後，資料庫如下所示：
man => [count=5, txid=3]
dog => [count=4, txid=3]
apple => [count=10, txid=2]
複製程式碼不透明事務性(模糊事務型)
模糊事務型spout不能保證txid的一批元組保持不變。不透明的事務性spout具有以下屬性：
- 每個元組只需一批成功處理。但是，元組可能無法在一個批處理中處理，然後在稍後的批處理中成功處理。
- 對於不透明的事務性spout，如果資料庫中的事務id與當前批處理的事務id相同，則不再可能使用跳過狀態更新的技巧。這是因為批次可能在狀態更新之間發生了變化。
非事務型 spout
非事務型 spout不對每批中的物品提供任何保證。因此它可能最多隻進行一次處理，在這種情況下，在批次失敗後不會重試元組。或者它可能具有至少一次處理，其中元組可以通過多個批次成功處理。對於這種spout，沒有辦法實現完全一次的語義。
不同型別的 Spout 與 State 的總結
模糊事務型 state 具有最好的容錯性特徵，不過這是以在資料庫中儲存更多的內容為代價的（一個 txid 和兩個 value）。事務型 state 要求的儲存空間相對較小，但是它的缺點是隻對事務型 spout 有效。相對的，非事務型要求的儲存空間最少，但是它也不能提供任何的恰好一次的訊息執行語義。
你選擇 state 與 spout 的時候必須在容錯性與儲存空間佔用之間權衡。可以根據你的應用的需求來確定哪種組合最適合你。
Trident程式碼例項
簡單輸出資料
public class TridentTopology1 {
/**
* 接受一組輸入欄位併發出零個或多個元組作為輸出 （類似storm bolt資料流處理元件）
* @author qxw
* @data 2018年9月19日下午6:17:14
*/
public static class MyFunction extends BaseFunction {
private static final long serialVersionUID = 1L;
public void execute(TridentTuple tuple, TridentCollector collector) {
System.out.println("a: "+tuple.getIntegerByField("a"));
System.out.println("b: "+tuple.getIntegerByField("b"));
System.out.println("c: "+tuple.getIntegerByField("c"));
System.out.println("d: "+tuple.getIntegerByField("d"));
}
}
@SuppressWarnings("unchecked")
public static void main(String[] args) {
//固定批處理資料來源（類似storm原生的spout） 宣告2個輸入的欄位
FixedBatchSpout spout =new FixedBatchSpout(new Fields("a","b","c","d"),4,//設定批處理大小
new Values(1,4,7,10),
new Values(2,3,5,7),
new Values(6,9,7,2),
new Values(9,1,6,8) //設定資料內容
);
//是否迴圈傳送
spout.setCycle(false);
//建立topology
TridentTopology topology =new TridentTopology();
//指定資料來源
Stream input=topology.newStream("spout", spout);
//要實現storm原生spolt--bolt的模式在Trident中用each實現
input.each(new Fields("a","b","c","d"),
new MyFunction(),//執行函式 類似bolt
new Fields() //為空不向下發送
);
Config conf = new Config();
conf.setNumWorkers(1);
conf.setMaxSpoutPending(20);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("TridentTopology1", conf, topology.build());
}
}
複製程式碼Trident操作 - flters海量資料過濾
通過要繼承BaseFilter，重寫isKeep方法
public class TridentTopology2 {
/**
* 可以海量資料進行過濾，需要繼承BaseFilter，重寫isKeep方法
* @author qxw
* @data 2018年9月21日上午10:57:00
*/
public static class MyFilter extends BaseFilter {
private static final long serialVersionUID = 1L;
public boolean isKeep(TridentTuple tuple) {
//能夠被2對第1個和第2個值進行相加.然後除2，為0則發射，不為零則不發射射
return tuple.getInteger(1) % 2 == 0;
}
}
/**
* 類似原生storm bolt資料流處理元件
* @author qxw
* @data 2018年9月21日下午3:31:12
*/
public static class MyFunction extends BaseFunction{
private static final long serialVersionUID = 1L;
@Override
public void execute(TridentTuple tuple, TridentCollector collector) {
//獲取tuple輸入內容
Integer a = tuple.getIntegerByField("a");
Integer b = tuple.getIntegerByField("b");
Integer c = tuple.getIntegerByField("c");
Integer d = tuple.getIntegerByField("d");
System.out.println("a: "+ a + ", b: " + b + ", c: " + c + ", d: " + d);
}
}
@SuppressWarnings("unchecked")
public static void main(String[] args) {
//固定批處理資料來源（類似storm原生的spout） 宣告a,b,c,d四個欄位
FixedBatchSpout spout =new FixedBatchSpout(new Fields("a","b","c","d"),4,//設定批處理大小
new Values(1,4,7,10),
new Values(2,3,5,7),
new Values(6,9,7,2),
new Values(9,1,6,8) //設定資料內容
);
//是否迴圈傳送
spout.setCycle(false);
//建立topology
TridentTopology topology =new TridentTopology();
//指定資料來源
Stream input=topology.newStream("spout", spout);
//要實現storm原生spolt--bolt的模式在Trident中用each實現 (隨機分組)
input.shuffle().each(new Fields("a","b","c","d"),new MyFilter()).each(new Fields("a","b","c","d"), new MyFunction(),new Fields());
//本地模式
Config conf = new Config();
conf.setNumWorkers(1);
conf.setMaxSpoutPending(20);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("TridentTopology2", conf, topology.build());
//叢集模式
// StormSubmitter.submitTopology("TridentTopology1", conf, buildTopology());
}
複製程式碼Triden 實現單詞計數統計
public class TridentWordCount {
public static class MyFunction extends BaseFunction {
private static final long serialVersionUID = 1L;
public void execute(TridentTuple tuple, TridentCollector collector) {
String word=tuple.getStringByField("word");
Long count=tuple.getLongByField("count");
System.out.println(word+" : "+count);
}
}
@SuppressWarnings("unchecked")
public static void main(String[] args) {
/* 建立spout */
FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 4,
new Values("java php asd java"),
new Values("php css js html"),
new Values("js php java java"),
new Values("a a b c d"));
//是否迴圈傳送
spout.setCycle(false);
/* 建立topology */
TridentTopology topology = new TridentTopology();
/* 建立Stream spout1, 分詞、統計 */
topology.newStream("spout", spout)
//先切割
.each(new Fields("sentence"), new Split(), new Fields("word"))
//分組
.groupBy(new Fields("word"))
//聚合統計
.aggregate(new Count(), new Fields("count"))
//輸出函式
.each(new Fields("word","count"), new MyFunction(),new Fields())
//設定並行度
.parallelismHint(1);
Config conf = new Config();
conf.setNumWorkers(1);
conf.setMaxSpoutPending(20);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("TridentWordCount", conf, topology.build());
}
}
複製程式碼Trident 實現Drpc
public class TridentDrpc {
private static class MyFunction extends BaseFunction{
public void execute(TridentTuple tridentTuple, TridentCollector tridentCollector) {
String sentence = tridentTuple.getString(0);
for (String word : sentence.split(" ")) {
tridentCollector.emit(new Values(word));
}
}
}
public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
TridentTopology topology=new TridentTopology();
Config conf = new Config();
conf.setMaxSpoutPending(20);
//本地模式
if (args.length==0){
LocalCluster cluster = new LocalCluster();
LocalDRPC drpc = new LocalDRPC();
Stream input=topology.newDRPCStream("data",drpc);
input.each(new Fields("args"),new MyFunction(),new Fields("result")).project(new Fields("result"));
cluster.submitTopology("wordCount", conf, topology.build());
//呼叫
System.err.println("DRPC RESULT: " + drpc.execute("data", "cat the dog jumped"));
drpc.shutdown();
cluster.shutdown();
}else{
//叢集模式
conf.setNumWorkers(2);
StormSubmitter.submitTopology(args[0], conf, topology.build());
}
}
}
複製程式碼