聊聊storm TridentTopology的構建
序
本文主要研究一下storm TridentTopology的構建html
實例
@Test
public void testDebugTopologyBuild(){
FixedBatchSpout spout = new FixedBatchSpout(new Fields("user", "score"), 3,
new Values("nickt1", 4),
new Values("nickt2", 7),
new Values("nickt3", 8),
new Values("nickt4", 9),
new Values("nickt5", 7),
new Values("nickt6", 11),
new Values("nickt7", 5)
);
spout.setCycle(false);
TridentTopology topology = new TridentTopology();
Stream stream1 = topology.newStream("spout1",spout)
.each(new Fields("user", "score"), new BaseFunction() {
@Override
public void execute(TridentTuple tuple, TridentCollector collector) {
System.out.println("tuple:"+tuple);
}
},new Fields());
topology.build();
}
- 後面的分析爲了簡單起見，不少是依據這個實例來
TridentTopology.newStream
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/TridentTopology.javajava
public Stream newStream(String txId, IRichSpout spout) {
return newStream(txId, new RichSpoutBatchExecutor(spout));
}
public Stream newStream(String txId, IPartitionedTridentSpout spout) {
return newStream(txId, new PartitionedTridentSpoutExecutor(spout));
}
public Stream newStream(String txId, IOpaquePartitionedTridentSpout spout) {
return newStream(txId, new OpaquePartitionedTridentSpoutExecutor(spout));
}
public Stream newStream(String txId, ITridentDataSource dataSource) {
if (dataSource instanceof IBatchSpout) {
return newStream(txId, (IBatchSpout) dataSource);
} else if (dataSource instanceof ITridentSpout) {
return newStream(txId, (ITridentSpout) dataSource);
} else if (dataSource instanceof IPartitionedTridentSpout) {
return newStream(txId, (IPartitionedTridentSpout) dataSource);
} else if (dataSource instanceof IOpaquePartitionedTridentSpout) {
return newStream(txId, (IOpaquePartitionedTridentSpout) dataSource);
} else {
throw new UnsupportedOperationException("Unsupported stream");
}
}
public Stream newStream(String txId, IBatchSpout spout) {
Node n = new SpoutNode(getUniqueStreamId(), spout.getOutputFields(), txId, spout, SpoutNode.SpoutType.BATCH);
return addNode(n);
}
public Stream newStream(String txId, ITridentSpout spout) {
Node n = new SpoutNode(getUniqueStreamId(), spout.getOutputFields(), txId, spout, SpoutNode.SpoutType.BATCH);
return addNode(n);
}
protected Stream addNode(Node n) {
registerNode(n);
return new Stream(this, n.name, n);
}
protected void registerNode(Node n) {
_graph.addVertex(n);
if(n.stateInfo!=null) {
String id = n.stateInfo.id;
if(!_colocate.containsKey(id)) {
_colocate.put(id, new ArrayList());
}
_colocate.get(id).add(n);
}
}
- newStream的第一個參數是txId，第二個參數是ITridentDataSource
- ITridentDataSource分爲好幾個類型，分別有IBatchSpout、ITridentSpout、IPartitionedTridentSpout、IOpaquePartitionedTridentSpout
- 最後都是建立SpoutNode，而後registerNode添加到_graph(
若是node的stateInfo不爲null，還會添加到_colocate，不過SpoutNode該值爲null)，注意SpoutNode的SpoutType爲SpoutNode.SpoutType.BATCH
Node
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/planner/Node.javanode
public class Node extends DefaultResourceDeclarer<Node> implements Serializable {
private static final AtomicInteger INDEX = new AtomicInteger(0);
private String nodeId;
public String name = null;
public Fields allOutputFields;
public String streamId;
public Integer parallelismHint = null;
public NodeStateInfo stateInfo = null;
public int creationIndex;
public Node(String streamId, String name, Fields allOutputFields) {
this.nodeId = UUID.randomUUID().toString();
this.allOutputFields = allOutputFields;
this.streamId = streamId;
this.name = name;
this.creationIndex = INDEX.incrementAndGet();
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
return nodeId.equals(((Node) o).nodeId);
}
@Override
public int hashCode() {
return nodeId.hashCode();
}
@Override
public String toString() {
return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
}
public String shortString() {
return "nodeId: " + nodeId + ", allOutputFields: " + allOutputFields;
}
}
- Node繼承了DefaultResourceDeclarer，而它實現了resources相關的接口：ResourceDeclarer以及ITridentResource
- Node有幾個子類，分別是SpoutNode、ProcessorNode、PartitionNode
- SpoutNode就是spout信息的節點描述，ProcessorNode通常是trident的each、map、aggregrate、reduce、project等操做的節點描述，PartitionNode就是partition相關的節點描述
TridentTopology.build
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/TridentTopology.javaapache
public StormTopology build() {
DefaultDirectedGraph<Node, IndexedEdge> graph = (DefaultDirectedGraph) _graph.clone();
//......
List<SpoutNode> spoutNodes = new ArrayList<>();
// can be regular nodes (static state) or processor nodes
Set<Node> boltNodes = new LinkedHashSet<>();
for(Node n: graph.vertexSet()) {
if(n instanceof SpoutNode) {
spoutNodes.add((SpoutNode) n);
} else if(!(n instanceof PartitionNode)) {
boltNodes.add(n);
}
}
Set<Group> initialGroups = new LinkedHashSet<>();
//......
for(Node n: boltNodes) {
initialGroups.add(new Group(graph, n));
}
GraphGrouper grouper = new GraphGrouper(graph, initialGroups);
grouper.mergeFully();
Collection<Group> mergedGroups = grouper.getAllGroups();
// add identity partitions between groups
for(IndexedEdge<Node> e: new HashSet<>(graph.edgeSet())) {
if(!(e.source instanceof PartitionNode) && !(e.target instanceof PartitionNode)) {
Group g1 = grouper.nodeGroup(e.source);
Group g2 = grouper.nodeGroup(e.target);
// g1 being null means the source is a spout node
if(g1==null && !(e.source instanceof SpoutNode))
throw new RuntimeException("Planner exception: Null source group must indicate a spout node at this phase of planning");
if(g1==null || !g1.equals(g2)) {
graph.removeEdge(e);
PartitionNode pNode = makeIdentityPartition(e.source);
graph.addVertex(pNode);
graph.addEdge(e.source, pNode, new IndexedEdge(e.source, pNode, 0));
graph.addEdge(pNode, e.target, new IndexedEdge(pNode, e.target, e.index));
}
}
}
//......
// add in spouts as groups so we can get parallelisms
for(Node n: spoutNodes) {
grouper.addGroup(new Group(graph, n));
}
grouper.reindex();
mergedGroups = grouper.getAllGroups();
Map<Node, String> batchGroupMap = new HashMap<>();
List<Set<Node>> connectedComponents = new ConnectivityInspector<>(graph).connectedSets();
for(int i=0; i<connectedComponents.size(); i++) {
String groupId = "bg" + i;
for(Node n: connectedComponents.get(i)) {
batchGroupMap.put(n, groupId);
}
}
// System.out.println("GRAPH:");
// System.out.println(graph);
Map<Group, Integer> parallelisms = getGroupParallelisms(graph, grouper, mergedGroups);
TridentTopologyBuilder builder = new TridentTopologyBuilder();
Map<Node, String> spoutIds = genSpoutIds(spoutNodes);
Map<Group, String> boltIds = genBoltIds(mergedGroups);
for(SpoutNode sn: spoutNodes) {
Integer parallelism = parallelisms.get(grouper.nodeGroup(sn));
Map<String, Number> spoutRes = new HashMap<>(_resourceDefaults);
spoutRes.putAll(sn.getResources());
Number onHeap = spoutRes.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB);
Number offHeap = spoutRes.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB);
Number cpuLoad = spoutRes.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT);
SpoutDeclarer spoutDeclarer = null;
if(sn.type == SpoutNode.SpoutType.DRPC) {
spoutDeclarer = builder.setBatchPerTupleSpout(spoutIds.get(sn), sn.streamId,
(IRichSpout) sn.spout, parallelism, batchGroupMap.get(sn));
} else {
ITridentSpout s;
if(sn.spout instanceof IBatchSpout) {
s = new BatchSpoutExecutor((IBatchSpout)sn.spout);
} else if(sn.spout instanceof ITridentSpout) {
s = (ITridentSpout) sn.spout;
} else {
throw new RuntimeException("Regular rich spouts not supported yet... try wrapping in a RichSpoutBatchExecutor");
// TODO: handle regular rich spout without batches (need lots of updates to support this throughout)
}
spoutDeclarer = builder.setSpout(spoutIds.get(sn), sn.streamId, sn.txId, s, parallelism, batchGroupMap.get(sn));
}
if(onHeap != null) {
if(offHeap != null) {
spoutDeclarer.setMemoryLoad(onHeap, offHeap);
}
else {
spoutDeclarer.setMemoryLoad(onHeap);
}
}
if(cpuLoad != null) {
spoutDeclarer.setCPULoad(cpuLoad);
}
}
for(Group g: mergedGroups) {
if(!isSpoutGroup(g)) {
Integer p = parallelisms.get(g);
Map<String, String> streamToGroup = getOutputStreamBatchGroups(g, batchGroupMap);
Map<String, Number> groupRes = g.getResources(_resourceDefaults);
Number onHeap = groupRes.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB);
Number offHeap = groupRes.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB);
Number cpuLoad = groupRes.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT);
BoltDeclarer d = builder.setBolt(boltIds.get(g), new SubtopologyBolt(graph, g.nodes, batchGroupMap), p,
committerBatches(g, batchGroupMap), streamToGroup);
if(onHeap != null) {
if(offHeap != null) {
d.setMemoryLoad(onHeap, offHeap);
}
else {
d.setMemoryLoad(onHeap);
}
}
if(cpuLoad != null) {
d.setCPULoad(cpuLoad);
}
Collection<PartitionNode> inputs = uniquedSubscriptions(externalGroupInputs(g));
for(PartitionNode n: inputs) {
Node parent = TridentUtils.getParent(graph, n);
String componentId = parent instanceof SpoutNode ?
spoutIds.get(parent) : boltIds.get(grouper.nodeGroup(parent));
d.grouping(new GlobalStreamId(componentId, n.streamId), n.thriftGrouping);
}
}
}
HashMap<String, Number> combinedMasterCoordResources = new HashMap<String, Number>(_resourceDefaults);
combinedMasterCoordResources.putAll(_masterCoordResources);
return builder.buildTopology(combinedMasterCoordResources);
}
- 這裏建立了TridentTopologyBuilder，而後對於spoutNodes，調用TridentTopologyBuilder.setSpout(String id, String streamName, String txStateId, ITridentSpout spout, Integer parallelism, String batchGroup)方法，添加spout
- 對於IBatchSpout類型的spout，經過BatchSpoutExecutor包裝爲ITridentSpout
- 這裏的streamName爲streamId，經過UniqueIdGen.getUniqueStreamId生成，以s開頭，以後是_streamCounter的計數，好比1，合起來就是s1；txStateId爲用戶傳入的txId；batchGroup以bg開頭，以後是connectedComponents的元素的index，好比0，合起來就是bg0；parallelism參數就是用戶構建topology時設置的
- 設置完spout以後，就是設置spout的相關資源配置，好比memoryLoad、cpuLoad；以後設置bolt，這裏使用的是SubtopologyBolt，而後設置bolt相關的資源配置
- 最後調用TridentTopologyBuilder.buildTopology
TridentTopologyBuilder.setSpout
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentTopologyBuilder.javasegmentfault
Map<GlobalStreamId, String> _batchIds = new HashMap();
Map<String, TransactionalSpoutComponent> _spouts = new HashMap();
public SpoutDeclarer setSpout(String id, String streamName, String txStateId, ITridentSpout spout, Integer parallelism, String batchGroup) {
Map<String, String> batchGroups = new HashMap();
batchGroups.put(streamName, batchGroup);
markBatchGroups(id, batchGroups);
TransactionalSpoutComponent c = new TransactionalSpoutComponent(spout, streamName, parallelism, txStateId, batchGroup);
_spouts.put(id, c);
return new SpoutDeclarerImpl(c);
}
private void markBatchGroups(String component, Map<String, String> batchGroups) {
for(Map.Entry<String, String> entry: batchGroups.entrySet()) {
_batchIds.put(new GlobalStreamId(component, entry.getKey()), entry.getValue());
}
}
- 這裏調用了markBatchGroups，將新的component添加到_batchIds中，同時也添加到_spouts中
TridentTopologyBuilder.setBolt
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentTopologyBuilder.javaapp
Map<GlobalStreamId, String> _batchIds = new HashMap();
Map<String, Component> _bolts = new HashMap();
// map from stream name to batch id
public BoltDeclarer setBolt(String id, ITridentBatchBolt bolt, Integer parallelism, Set<String> committerBatches, Map<String, String> batchGroups) {
markBatchGroups(id, batchGroups);
Component c = new Component(bolt, parallelism, committerBatches);
_bolts.put(id, c);
return new BoltDeclarerImpl(c);
}
private void markBatchGroups(String component, Map<String, String> batchGroups) {
for(Map.Entry<String, String> entry: batchGroups.entrySet()) {
_batchIds.put(new GlobalStreamId(component, entry.getKey()), entry.getValue());
}
}
- 這裏調用了markBatchGroups將新的component添加到_batchIds中，同時也添加到_bolts中；對於trident來講，就是一系列的ProcessorNode(
可能也會有PartitionNode)
TridentTopologyBuilder.buildTopology
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentTopologyBuilder.javadom
public StormTopology buildTopology(Map<String, Number> masterCoordResources) {
TopologyBuilder builder = new TopologyBuilder();
Map<GlobalStreamId, String> batchIdsForSpouts = fleshOutStreamBatchIds(false);
Map<GlobalStreamId, String> batchIdsForBolts = fleshOutStreamBatchIds(true);
Map<String, List<String>> batchesToCommitIds = new HashMap<>();
Map<String, List<ITridentSpout>> batchesToSpouts = new HashMap<>();
for(String id: _spouts.keySet()) {
TransactionalSpoutComponent c = _spouts.get(id);
if(c.spout instanceof IRichSpout) {
//TODO: wrap this to set the stream name
builder.setSpout(id, (IRichSpout) c.spout, c.parallelism);
} else {
String batchGroup = c.batchGroupId;
if(!batchesToCommitIds.containsKey(batchGroup)) {
batchesToCommitIds.put(batchGroup, new ArrayList<String>());
}
batchesToCommitIds.get(batchGroup).add(c.commitStateId);
if(!batchesToSpouts.containsKey(batchGroup)) {
batchesToSpouts.put(batchGroup, new ArrayList<ITridentSpout>());
}
batchesToSpouts.get(batchGroup).add((ITridentSpout) c.spout);
BoltDeclarer scd =
builder.setBolt(spoutCoordinator(id), new TridentSpoutCoordinator(c.commitStateId, (ITridentSpout) c.spout))
.globalGrouping(masterCoordinator(c.batchGroupId), MasterBatchCoordinator.BATCH_STREAM_ID)
.globalGrouping(masterCoordinator(c.batchGroupId), MasterBatchCoordinator.SUCCESS_STREAM_ID);
for(Map<String, Object> m: c.componentConfs) {
scd.addConfigurations(m);
}
Map<String, TridentBoltExecutor.CoordSpec> specs = new HashMap();
specs.put(c.batchGroupId, new CoordSpec());
BoltDeclarer bd = builder.setBolt(id,
new TridentBoltExecutor(
new TridentSpoutExecutor(
c.commitStateId,
c.streamName,
((ITridentSpout) c.spout)),
batchIdsForSpouts,
specs),
c.parallelism);
bd.allGrouping(spoutCoordinator(id), MasterBatchCoordinator.BATCH_STREAM_ID);
bd.allGrouping(masterCoordinator(batchGroup), MasterBatchCoordinator.SUCCESS_STREAM_ID);
if(c.spout instanceof ICommitterTridentSpout) {
bd.allGrouping(masterCoordinator(batchGroup), MasterBatchCoordinator.COMMIT_STREAM_ID);
}
for(Map<String, Object> m: c.componentConfs) {
bd.addConfigurations(m);
}
}
}
//......
Number onHeap = masterCoordResources.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB);
Number offHeap = masterCoordResources.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB);
Number cpuLoad = masterCoordResources.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT);
for(String batch: batchesToCommitIds.keySet()) {
List<String> commitIds = batchesToCommitIds.get(batch);
SpoutDeclarer masterCoord = builder.setSpout(masterCoordinator(batch), new MasterBatchCoordinator(commitIds, batchesToSpouts.get(batch)));
if(onHeap != null) {
if(offHeap != null) {
masterCoord.setMemoryLoad(onHeap, offHeap);
}
else {
masterCoord.setMemoryLoad(onHeap);
}
}
if(cpuLoad != null) {
masterCoord.setCPULoad(cpuLoad);
}
}
for(String id: _bolts.keySet()) {
Component c = _bolts.get(id);
Map<String, CoordSpec> specs = new HashMap<>();
for(GlobalStreamId s: getBoltSubscriptionStreams(id)) {
String batch = batchIdsForBolts.get(s);
if(!specs.containsKey(batch)) specs.put(batch, new CoordSpec());
CoordSpec spec = specs.get(batch);
CoordType ct;
if(_batchPerTupleSpouts.containsKey(s.get_componentId())) {
ct = CoordType.single();
} else {
ct = CoordType.all();
}
spec.coords.put(s.get_componentId(), ct);
}
for(String b: c.committerBatches) {
specs.get(b).commitStream = new GlobalStreamId(masterCoordinator(b), MasterBatchCoordinator.COMMIT_STREAM_ID);
}
BoltDeclarer d = builder.setBolt(id, new TridentBoltExecutor(c.bolt, batchIdsForBolts, specs), c.parallelism);
for(Map<String, Object> conf: c.componentConfs) {
d.addConfigurations(conf);
}
for(InputDeclaration inputDecl: c.declarations) {
inputDecl.declare(d);
}
Map<String, Set<String>> batchToComponents = getBoltBatchToComponentSubscriptions(id);
for(Map.Entry<String, Set<String>> entry: batchToComponents.entrySet()) {
for(String comp: entry.getValue()) {
d.directGrouping(comp, TridentBoltExecutor.COORD_STREAM(entry.getKey()));
}
}
for(String b: c.committerBatches) {
d.allGrouping(masterCoordinator(b), MasterBatchCoordinator.COMMIT_STREAM_ID);
}
}
return builder.createTopology();
}
- buildTopology對於非IRichSpout的的spout會在topology中建立TridentSpoutCoordinator這個bolt，它globalGrouping了MasterBatchCoordinator.BATCH_STREAM_ID(
$batch)、MasterBatchCoordinator.SUCCESS_STREAM_ID($success)這兩個stream；同時還建立了TridentBoltExecutor這個bolt，它allGrouping了MasterBatchCoordinator.BATCH_STREAM_ID($batch)、MasterBatchCoordinator.SUCCESS_STREAM_ID($success)，對於spout是ICommitterTridentSpout類型的，還allGrouping了MasterBatchCoordinator.COMMIT_STREAM_ID($commit)；注意這裏將非IRichSpout的spout轉換爲bolt - 以後對batchesToCommitIds中的每一個batch建立MasterBatchCoordinator這個spout，正好前前面的TridentSpoutCoordinator以及TridentBoltExecutor銜接起來
- 對於bolt來講(
包裝了ProcessorNode的SubtopologyBolt)，這裏設置了TridentBoltExecutor這個bolt，它directGrouping了TridentBoltExecutor.COORD_STREAM($coord-)，同時還allGrouping了MasterBatchCoordinator.COMMIT_STREAM_ID($commit)
TridentTopologyBuilder.createTopology
storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentTopologyBuilder.javaide
public StormTopology createTopology() {
Map<String, Bolt> boltSpecs = new HashMap<>();
Map<String, SpoutSpec> spoutSpecs = new HashMap<>();
maybeAddCheckpointSpout();
for(String boltId: _bolts.keySet()) {
IRichBolt bolt = _bolts.get(boltId);
bolt = maybeAddCheckpointTupleForwarder(bolt);
ComponentCommon common = getComponentCommon(boltId, bolt);
try{
maybeAddCheckpointInputs(common);
boltSpecs.put(boltId, new Bolt(ComponentObject.serialized_java(Utils.javaSerialize(bolt)), common));
}catch(RuntimeException wrapperCause){
if (wrapperCause.getCause() != null && NotSerializableException.class.equals(wrapperCause.getCause().getClass())){
throw new IllegalStateException(
"Bolt '" + boltId + "' contains a non-serializable field of type " + wrapperCause.getCause().getMessage() + ", " +
"which was instantiated prior to topology creation. " + wrapperCause.getCause().getMessage() + " " +
"should be instantiated within the prepare method of '" + boltId + " at the earliest.", wrapperCause);
}
throw wrapperCause;
}
}
for(String spoutId: _spouts.keySet()) {
IRichSpout spout = _spouts.get(spoutId);
ComponentCommon common = getComponentCommon(spoutId, spout);
try{
spoutSpecs.put(spoutId, new SpoutSpec(ComponentObject.serialized_java(Utils.javaSerialize(spout)), common));
}catch(RuntimeException wrapperCause){
if (wrapperCause.getCause() != null && NotSerializableException.class.equals(wrapperCause.getCause().getClass())){
throw new IllegalStateException(
"Spout '" + spoutId + "' contains a non-serializable field of type " + wrapperCause.getCause().getMessage() + ", " +
"which was instantiated prior to topology creation. " + wrapperCause.getCause().getMessage() + " " +
"should be instantiated within the prepare method of '" + spoutId + " at the earliest.", wrapperCause);
}
throw wrapperCause;
}
}
StormTopology stormTopology = new StormTopology(spoutSpecs,
boltSpecs,
new HashMap<String, StateSpoutSpec>());
stormTopology.set_worker_hooks(_workerHooks);
return Utils.addVersions(stormTopology);
}
/**
* If the topology has at least one stateful bolt
* add a {@link CheckpointSpout} component to the topology.
*/
private void maybeAddCheckpointSpout() {
if (hasStatefulBolt) {
setSpout(CHECKPOINT_COMPONENT_ID, new CheckpointSpout(), 1);
}
}
- createTopology的時候，判斷若是有stateful的bolt，則會添加CheckpointSpout這個spout；同時對每一個bolt判斷若是是statefulBolt且不是StatefulBoltExecutor，那麼會添加CheckpointTupleForwarder
- 通過buildTopology的一系列設置，到了createTopology這裏，已經有了3個bolt，一個是包裝了ProcessNode的TridentBoltExecutor，一個是TridentSpoutCoordinator，還有一個是包裝了原始spout的TridentBoltExecutor
- spout這裏只有一個就是MasterBatchCoordinator，在buildTopology的時候，對於非IRichSpout的的spout，會被轉化爲TridentSpoutCoordinator這個bolt
拓撲結構
- 之前面的實例來說，通過TridentTopologyBuilder的createTopology，最後的拓撲結構爲一個spout爲MasterBatchCoordinator(
$mastercoord-bg0)，3個bolt分別爲TridentSpoutCoordinator($spoutcoord-spout-spout1)、包裝了非IRichSpout的的spout的TridentBoltExecutor(spout-spout1)、包裝了ProcessorNode的TridentBoltExecutor(b-0)；一共涉及到了幾個stream，分別爲MasterBatchCoordinator.SUCCESS_STREAM_ID($success)、MasterBatchCoordinator.COMMIT_STREAM_ID($commit)、MasterBatchCoordinator.BATCH_STREAM_ID($batch)、TridentBoltExecutor.COORD_STREAM($coord-bg0)、s一、s2 -
$mastercoord-bg0它declare了$success、$commit、$batch這三個stream，outputFields均爲tx這個字段 -
$spoutcoord-spout-spout1它接收了$mastercoord-bg0的$success、$batch這兩個stream，同時declare了$batch這個stream，outputFields爲[tx,metadata] -
spout-spout1，它allGrouping接收$mastercoord-bg0的$success，以及$spoutcoord-spout-spout1的$batch這兩個stream的數據；同時會往$coord-bg0發送[id,count]數據，以及stream(s1)發送數據tuple -
b-0它接收了spout-spout1的$coord-bg0以及s1這兩個stream的數據，以後往stream(s2)發送數據(output_fields:[$batchId, user, score])，同時也會往stream($coord-bg0)發送[id, count]數據
小結
- TridentTopologyBuilder在buildTopology的時候，對於非IRichSpout的的spout，會被轉化爲TridentBoltExecutor這個bolt，同時會新增一個TridentSpoutCoordinator這個bolt；ProcessorNode則會被包裝爲TridentBoltExecutor這個bolt；TridentTopology爲了方便管理將用戶設定的spout包裝爲bolt，而後建立MasterBatchCoordinator做爲真正的spout
- TridentBoltExecutor.COORD_STREAM(
$coord-)這個stream用來在component之間傳遞[id, count]數據，用於保障tuple在每一個component可以完整傳輸，即spout和bolt都會往該stream發送[id, count]數據 - MasterBatchCoordinator、TridentSpoutCoordinator、包裝原始spout的TridentBoltExecutor(
spout-spout1)它們之間的關係以下：master會給spout-spout1發送suceess數據(tuple\指令)，給coordinator發送suceess、batch數據(tuple\指令)；coordinator會給spout-spout1發送batch數據(tuple\指令)
doc
相關文章
- 1. 聊聊storm TridentTopology的構建
- 2. 聊聊storm trident的coordinator
- 3. 聊聊storm WindowTridentProcessor的FreshCollector
- 4. 聊聊storm的messageTimeout
- 5. 聊聊storm的IEventLogger
- 6. 聊聊storm的PartialKeyGrouping
- 7. 聊聊storm的LoggingMetricsConsumer
- 8. 聊聊storm的WindowedBoltExecutor
- 9. 聊聊storm的maxSpoutPending
- 10. 聊聊storm的tickTuple
- 更多相關文章...
- • Maven 構建生命週期 - Maven教程
- • Maven 構建配置文件 - Maven教程
- • 適用於PHP初學者的學習線路和建議
- • 漫談MySQL的鎖機制
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 聊聊storm TridentTopology的構建
- 2. 聊聊storm trident的coordinator
- 3. 聊聊storm WindowTridentProcessor的FreshCollector
- 4. 聊聊storm的messageTimeout
- 5. 聊聊storm的IEventLogger
- 6. 聊聊storm的PartialKeyGrouping
- 7. 聊聊storm的LoggingMetricsConsumer
- 8. 聊聊storm的WindowedBoltExecutor
- 9. 聊聊storm的maxSpoutPending
- 10. 聊聊storm的tickTuple