flink-storm_2.11-1.6.2-sources.jar!/org/apache/flink/storm/wrappers/BoltWrapper.java

flink-storm_2.11-1.6.2-sources.jar!/org/apache/flink/storm/wrappers/SpoutWrapper.java

这里使用FlinkLocalCluster.getLocalCluster()来创建或获取FlinkLocalCluster，之后调用FlinkLocalCluster.submitTopology来提交topology，结束时通过FlinkLocalCluster.shutdown来关闭cluster

序 本文主要研究一下flink LocalEnvironment的execute方法 实例 {代码...} 这里使用DataSet从csv读取数据，然后进行flatMap、groupBy、sum操作，最后调用print输出 DataSet.print flink-java-1.6.2-sources.jar!/org/apache/flink/api/java/DataSet.java {代码...} print方法这里主要是调用collect方法，获取结果，然后...

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/TridentTopology.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/spout/ICommitterTridentSpout.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/MasterBatchCoordinator.java

b-1主要是处理CombinerAggregator的init，它的parallelismHint为1，由于它的上游bolt有3个task，因而它的TridentBoltExecutor的tracked.condition.expectedTaskReports为3，它要等到这三个task的聚合数据都到了之后，才能finishBatch

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/MasterBatchCoordinator.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentBoltExecutor.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/windowing/WindowTridentProcessor.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentBoltExecutor.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/windowing/WindowTridentProcessor.java

storm-kafka-client-1.2.2-sources.jar!/org/apache/storm/kafka/spout/KafkaSpoutConfig.java

storm-1.2.2/storm-core/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/TridentTopology.java

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java

storm-2.0.0/storm-server/src/main/java/org/apache/storm/metric/LoggingClusterMetricsConsumer.java

storm-2.0.0/storm-client/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java

storm-2.0.0/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/operation/Function.java

storm-2.0.0/storm-client/src/jvm/org/apache/storm/trident/state/StateType.java

storm-2.0.0/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java

这里有两个相关的参数，分别是storm.messaging.netty.buffer.high.watermark以及storm.messaging.netty.buffer.low.watermark

序 本文主要研究一下storm的messageTimeout TOPOLOGY_MESSAGE_TIMEOUT_SECS storm-2.0.0/storm-client/src/jvm/org/apache/storm/Config.java {代码...} defaults.yaml中topology.enable.message.timeouts默认为true defaults.yaml中topology.message.timeout.secs默认为30 defaults.yaml中topology.tick.tuple.freq.sec...

TOPOLOGY_MAX_SPOUT_PENDING设置的是一个spout task已经emit等待ack的tuple的最大数量，该配置仅仅对于发射可靠tuple(设置msgId)的spout起作用

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/IWaitStrategy.java

DRPCSpout的outputFields为Fields("args", "return-info")，ReturnResults接收的fields为Fields("result", "return-info")

这里在SplitStreamBolt里头将stream分为两个，之后有两个CountStreamBolt分别处理两个stream的数据，最后归到同一个stream由ReportBolt消费tuple

JoinBolt继承了BaseWindowedBolt，定义了Selector selectorType、LinkedHashMap<String, JoinInfo> joinCriteria、FieldSelector[] outputFields等属性，用于记录关联类型及关联关系