聊聊storm TridentBoltExecutor的finishBatch方法
序
本文主要研究一下storm TridentBoltExecutor的finishBatch方法html
MasterBatchCoordinator.nextTuple
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/MasterBatchCoordinator.javajava
public void nextTuple() {
sync();
}
private void sync() {
// note that sometimes the tuples active may be less than max_spout_pending, e.g.
// max_spout_pending = 3
// tx 1, 2, 3 active, tx 2 is acked. there won't be a commit for tx 2 (because tx 1 isn't committed yet),
// and there won't be a batch for tx 4 because there's max_spout_pending tx active
TransactionStatus maybeCommit = _activeTx.get(_currTransaction);
if(maybeCommit!=null && maybeCommit.status == AttemptStatus.PROCESSED) {
maybeCommit.status = AttemptStatus.COMMITTING;
_collector.emit(COMMIT_STREAM_ID, new Values(maybeCommit.attempt), maybeCommit.attempt);
LOG.debug("Emitted on [stream = {}], [tx_status = {}], [{}]", COMMIT_STREAM_ID, maybeCommit, this);
}
if(_active) {
if(_activeTx.size() < _maxTransactionActive) {
Long curr = _currTransaction;
for(int i=0; i<_maxTransactionActive; i++) {
if(!_activeTx.containsKey(curr) && isReady(curr)) {
// by using a monotonically increasing attempt id, downstream tasks
// can be memory efficient by clearing out state for old attempts
// as soon as they see a higher attempt id for a transaction
Integer attemptId = _attemptIds.get(curr);
if(attemptId==null) {
attemptId = 0;
} else {
attemptId++;
}
_attemptIds.put(curr, attemptId);
for(TransactionalState state: _states) {
state.setData(CURRENT_ATTEMPTS, _attemptIds);
}
TransactionAttempt attempt = new TransactionAttempt(curr, attemptId);
final TransactionStatus newTransactionStatus = new TransactionStatus(attempt);
_activeTx.put(curr, newTransactionStatus);
_collector.emit(BATCH_STREAM_ID, new Values(attempt), attempt);
LOG.debug("Emitted on [stream = {}], [tx_attempt = {}], [tx_status = {}], [{}]", BATCH_STREAM_ID, attempt, newTransactionStatus, this);
_throttler.markEvent();
}
curr = nextTransactionId(curr);
}
}
}
}
- MasterBatchCoordinator是整個trident的真正的spout，它的nextTuple方法會向TridentSpoutCoordinator向MasterBatchCoordinator.BATCH_STREAM_ID(
$batch)發射tuple
TridentSpoutCoordinator.execute
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/spout/TridentSpoutCoordinator.javaapache
public void execute(Tuple tuple, BasicOutputCollector collector) {
TransactionAttempt attempt = (TransactionAttempt) tuple.getValue(0);
if(tuple.getSourceStreamId().equals(MasterBatchCoordinator.SUCCESS_STREAM_ID)) {
_state.cleanupBefore(attempt.getTransactionId());
_coord.success(attempt.getTransactionId());
} else {
long txid = attempt.getTransactionId();
Object prevMeta = _state.getPreviousState(txid);
Object meta = _coord.initializeTransaction(txid, prevMeta, _state.getState(txid));
_state.overrideState(txid, meta);
collector.emit(MasterBatchCoordinator.BATCH_STREAM_ID, new Values(attempt, meta));
}
}
- TridentSpoutCoordinator接收MasterBatchCoordinator在MasterBatchCoordinator.BATCH_STREAM_ID(
$batch)發過來的tuple，而後向包裝用戶spout的TridentBoltExecutor發送batch指令
TridentBoltExecutor(TridentSpoutExecutor)
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentBoltExecutor.javasegmentfault
public void execute(Tuple tuple) {
if(TupleUtils.isTick(tuple)) {
long now = System.currentTimeMillis();
if(now - _lastRotate > _messageTimeoutMs) {
_batches.rotate();
_lastRotate = now;
}
return;
}
String batchGroup = _batchGroupIds.get(tuple.getSourceGlobalStreamId());
if(batchGroup==null) {
// this is so we can do things like have simple DRPC that doesn't need to use batch processing
_coordCollector.setCurrBatch(null);
_bolt.execute(null, tuple);
_collector.ack(tuple);
return;
}
IBatchID id = (IBatchID) tuple.getValue(0);
//get transaction id
//if it already exists and attempt id is greater than the attempt there
TrackedBatch tracked = (TrackedBatch) _batches.get(id.getId());
// if(_batches.size() > 10 && _context.getThisTaskIndex() == 0) {
// System.out.println("Received in " + _context.getThisComponentId() + " " + _context.getThisTaskIndex()
// + " (" + _batches.size() + ")" +
// "\ntuple: " + tuple +
// "\nwith tracked " + tracked +
// "\nwith id " + id +
// "\nwith group " + batchGroup
// + "\n");
//
// }
//System.out.println("Num tracked: " + _batches.size() + " " + _context.getThisComponentId() + " " + _context.getThisTaskIndex());
// this code here ensures that only one attempt is ever tracked for a batch, so when
// failures happen you don't get an explosion in memory usage in the tasks
if(tracked!=null) {
if(id.getAttemptId() > tracked.attemptId) {
_batches.remove(id.getId());
tracked = null;
} else if(id.getAttemptId() < tracked.attemptId) {
// no reason to try to execute a previous attempt than we've already seen
return;
}
}
if(tracked==null) {
tracked = new TrackedBatch(new BatchInfo(batchGroup, id, _bolt.initBatchState(batchGroup, id)), _coordConditions.get(batchGroup), id.getAttemptId());
_batches.put(id.getId(), tracked);
}
_coordCollector.setCurrBatch(tracked);
//System.out.println("TRACKED: " + tracked + " " + tuple);
TupleType t = getTupleType(tuple, tracked);
if(t==TupleType.COMMIT) {
tracked.receivedCommit = true;
checkFinish(tracked, tuple, t);
} else if(t==TupleType.COORD) {
int count = tuple.getInteger(1);
tracked.reportedTasks++;
tracked.expectedTupleCount+=count;
checkFinish(tracked, tuple, t);
} else {
tracked.receivedTuples++;
boolean success = true;
try {
_bolt.execute(tracked.info, tuple);
if(tracked.condition.expectedTaskReports==0) {
success = finishBatch(tracked, tuple);
}
} catch(FailedException e) {
failBatch(tracked, e);
}
if(success) {
_collector.ack(tuple);
} else {
_collector.fail(tuple);
}
}
_coordCollector.setCurrBatch(null);
}
private boolean finishBatch(TrackedBatch tracked, Tuple finishTuple) {
boolean success = true;
try {
_bolt.finishBatch(tracked.info);
String stream = COORD_STREAM(tracked.info.batchGroup);
for(Integer task: tracked.condition.targetTasks) {
_collector.emitDirect(task, stream, finishTuple, new Values(tracked.info.batchId, Utils.get(tracked.taskEmittedTuples, task, 0)));
}
if(tracked.delayedAck!=null) {
_collector.ack(tracked.delayedAck);
tracked.delayedAck = null;
}
} catch(FailedException e) {
failBatch(tracked, e);
success = false;
}
_batches.remove(tracked.info.batchId.getId());
return success;
}
- TridentBoltExecutor.execute方法，首先會建立並初始化TrackedBatch(
若是TrackedBatch不存在的話)，以後接收到batch指令的時候，對tracked.receivedTuple累加，而後調用_bolt.execute(tracked.info, tuple) - 對於spout來講，這裏的_bolt是TridentSpoutExecutor，它的execute方法會往下游的TridentBoltExecutor發射一個batch的tuples；因爲spout的expectedTaskReports==0，因此這裏在調用完TridentSpoutExecutor發射batch的tuples時，它就立馬調用finishBatch
- finishBatch操做，這裏會經過COORD_STREAM往下游的TridentBoltExecutor發射[id,count]數據，告知下游TridentBoltExecutor說它一共發射了多少tuples
TridentBoltExecutor(SubtopologyBolt)
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentBoltExecutor.javaapp
@Override
public void execute(Tuple tuple) {
if(TupleUtils.isTick(tuple)) {
long now = System.currentTimeMillis();
if(now - _lastRotate > _messageTimeoutMs) {
_batches.rotate();
_lastRotate = now;
}
return;
}
String batchGroup = _batchGroupIds.get(tuple.getSourceGlobalStreamId());
if(batchGroup==null) {
// this is so we can do things like have simple DRPC that doesn't need to use batch processing
_coordCollector.setCurrBatch(null);
_bolt.execute(null, tuple);
_collector.ack(tuple);
return;
}
IBatchID id = (IBatchID) tuple.getValue(0);
//get transaction id
//if it already exists and attempt id is greater than the attempt there
TrackedBatch tracked = (TrackedBatch) _batches.get(id.getId());
// if(_batches.size() > 10 && _context.getThisTaskIndex() == 0) {
// System.out.println("Received in " + _context.getThisComponentId() + " " + _context.getThisTaskIndex()
// + " (" + _batches.size() + ")" +
// "\ntuple: " + tuple +
// "\nwith tracked " + tracked +
// "\nwith id " + id +
// "\nwith group " + batchGroup
// + "\n");
//
// }
//System.out.println("Num tracked: " + _batches.size() + " " + _context.getThisComponentId() + " " + _context.getThisTaskIndex());
// this code here ensures that only one attempt is ever tracked for a batch, so when
// failures happen you don't get an explosion in memory usage in the tasks
if(tracked!=null) {
if(id.getAttemptId() > tracked.attemptId) {
_batches.remove(id.getId());
tracked = null;
} else if(id.getAttemptId() < tracked.attemptId) {
// no reason to try to execute a previous attempt than we've already seen
return;
}
}
if(tracked==null) {
tracked = new TrackedBatch(new BatchInfo(batchGroup, id, _bolt.initBatchState(batchGroup, id)), _coordConditions.get(batchGroup), id.getAttemptId());
_batches.put(id.getId(), tracked);
}
_coordCollector.setCurrBatch(tracked);
//System.out.println("TRACKED: " + tracked + " " + tuple);
TupleType t = getTupleType(tuple, tracked);
if(t==TupleType.COMMIT) {
tracked.receivedCommit = true;
checkFinish(tracked, tuple, t);
} else if(t==TupleType.COORD) {
int count = tuple.getInteger(1);
tracked.reportedTasks++;
tracked.expectedTupleCount+=count;
checkFinish(tracked, tuple, t);
} else {
tracked.receivedTuples++;
boolean success = true;
try {
_bolt.execute(tracked.info, tuple);
if(tracked.condition.expectedTaskReports==0) {
success = finishBatch(tracked, tuple);
}
} catch(FailedException e) {
failBatch(tracked, e);
}
if(success) {
_collector.ack(tuple);
} else {
_collector.fail(tuple);
}
}
_coordCollector.setCurrBatch(null);
}
private void checkFinish(TrackedBatch tracked, Tuple tuple, TupleType type) {
if(tracked.failed) {
failBatch(tracked);
_collector.fail(tuple);
return;
}
CoordCondition cond = tracked.condition;
boolean delayed = tracked.delayedAck==null &&
(cond.commitStream!=null && type==TupleType.COMMIT
|| cond.commitStream==null);
if(delayed) {
tracked.delayedAck = tuple;
}
boolean failed = false;
if(tracked.receivedCommit && tracked.reportedTasks == cond.expectedTaskReports) {
if(tracked.receivedTuples == tracked.expectedTupleCount) {
finishBatch(tracked, tuple);
} else {
//TODO: add logging that not all tuples were received
failBatch(tracked);
_collector.fail(tuple);
failed = true;
}
}
if(!delayed && !failed) {
_collector.ack(tuple);
}
}
private boolean finishBatch(TrackedBatch tracked, Tuple finishTuple) {
boolean success = true;
try {
_bolt.finishBatch(tracked.info);
String stream = COORD_STREAM(tracked.info.batchGroup);
for(Integer task: tracked.condition.targetTasks) {
_collector.emitDirect(task, stream, finishTuple, new Values(tracked.info.batchId, Utils.get(tracked.taskEmittedTuples, task, 0)));
}
if(tracked.delayedAck!=null) {
_collector.ack(tracked.delayedAck);
tracked.delayedAck = null;
}
} catch(FailedException e) {
failBatch(tracked, e);
success = false;
}
_batches.remove(tracked.info.batchId.getId());
return success;
}
- TridentBoltExecutor(
SubtopologyBolt)是spout下游的bolt，它的_bolt是SubtopologyBolt，並且它的tracked.condition.expectedTaskReports不爲0，於是它是在接收到TupleType.COORD的tuple的時候，才進行checkFinish操做(這裏先忽略TupleType.COMMIT類型) - 因爲BoltExecutor是使用Utils.asyncLoop來挨個消費receiveQueue的數據的，並且emitBatch的時候也是挨個接收batch的tuples，最後再接收到TridentBoltExecutor(
TridentSpoutExecutor)在finishBatch的時候經過COORD_STREAM發過來的[id,count]的tuple(注意這裏的COORD_STREAM是分發給每一個task的，若是TridentBoltExecutor有多個parallel，則他們是按各自的task來接收的) - 因此TridentBoltExecutor(
SubtopologyBolt)先挨個處理每一個tuple，處理完以後才輪到TupleType.COORD這個tuple，而後觸發checkFinish操做；在沒有commitStream的狀況下，tracked.receivedCommit默認爲true，於是這裏只要檢測收到的tuples與應收的tuples數一致，就執行_bolt.finishBatch操做完成一個batch，而後再往它的下游TridentBoltExecutor發射它應收的[id,count]的tuple
小結
- 對於trident來講，真正的spout是MasterBatchCoordinator，它的nextTuple會觸發batch的發送，它將batch指令發送給TridentSpoutCoordinator，而TridentSpoutCoordinator將觸發TridentBoltExecutor(
TridentSpoutExecutor)的execute方法，進而觸發ITridentSpout的emitter的emitBatch，從而發送一個batch的數據 - TridentBoltExecutor(
TridentSpoutExecutor)的expectedTaskReports==0，它在調用完TridentSpoutExecutor發射batch的tuples時，就立馬調用finishBatch操做，經過COORD_STREAM往下游的TridentBoltExecutor發射[id,count]數據，告知下游TridentBoltExecutor說它一共發射了多少tuples - spout的下游bolt爲TridentBoltExecutor(
SubtopologyBolt)，它的tracked.condition.expectedTaskReports不爲0，於是它是在接收到TupleType.COORD的tuple的時候，才進行checkFinish操做(這裏先忽略TupleType.COMMIT類型)，因爲spout是先執行emitBatch操做再最後finishBatch發送[id,count]數據，正常狀況下按順序進入到TridentBoltExecutor(SubtopologyBolt)的receiveQueue隊列，而後TridentBoltExecutor(SubtopologyBolt)挨個消費tuple，調用SubtopologyBolt.execute，最後再處理[id,count]數據，觸發checkFinish操做，只要檢測收到的tuples與應收的tuples數一致，就執行SubtopologyBolt.finishBatch操做完成這個batch，而後再往它的下游TridentBoltExecutor發射它應收的[id,count]的tuple
doc
相關文章
- 1. 聊聊storm TridentBoltExecutor的finishBatch方法
- 2. 聊聊storm trident spout的_maxTransactionActive
- 3. 聊聊storm的AggregateProcessor的execute及finishBatch方法
- 4. 聊聊storm TridentWindowManager的pendingTriggers
- 5. [case46]聊聊storm trident spout的_maxTransactionActive
- 6. 聊聊storm trident的coordinator
- 7. 聊聊storm的ICommitterTridentSpout
- 8. 聊聊storm的OpaquePartitionedTridentSpoutExecutor
- 9. 聊聊storm TridentTopology的構建
- 10. 聊聊storm WindowTridentProcessor的FreshCollector
- 更多相關文章...
- • Spring實例化Bean的三種方法 - Spring教程
- • ASP Contents.Remove 方法 - ASP 教程
- • 常用的分佈式事務解決方案
- • SpringBoot中properties文件不能自動提示解決方法
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. vs2019運行opencv圖片顯示代碼時，窗口亂碼
- 2. app自動化 - 元素定位不到？別慌，看完你就能解決
- 3. 在Win8下用cisco ××× Client連接時報Reason 422錯誤的解決方法
- 4. eclipse快速補全代碼
- 5. Eclipse中Java/Html/Css/Jsp/JavaScript等代碼的格式化
- 6. idea+spring boot +mabitys(wanglezapin)+mysql (1)
- 7. 勒索病毒發生變種 新文件名將帶有「.UIWIX」後綴
- 8. 【原創】Python 源文件編碼解讀
- 9. iOS9企業部署分發問題深入瞭解與解決
- 10. 安裝pytorch報錯CondaHTTPError：******
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 聊聊storm TridentBoltExecutor的finishBatch方法
- 2. 聊聊storm trident spout的_maxTransactionActive
- 3. 聊聊storm的AggregateProcessor的execute及finishBatch方法
- 4. 聊聊storm TridentWindowManager的pendingTriggers
- 5. [case46]聊聊storm trident spout的_maxTransactionActive
- 6. 聊聊storm trident的coordinator
- 7. 聊聊storm的ICommitterTridentSpout
- 8. 聊聊storm的OpaquePartitionedTridentSpoutExecutor
- 9. 聊聊storm TridentTopology的構建
- 10. 聊聊storm WindowTridentProcessor的FreshCollector