You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2016/06/01 06:21:00 UTC
svn commit: r1746399 [31/31] - in /storm/site: _posts/ publish/
publish/2014/04/17/ publish/about/ publish/releases/0.10.0/
publish/releases/0.10.0/about/ publish/releases/0.10.1/
publish/releases/0.10.1/about/ publish/releases/1.0.0/ publish/releases/...
Modified: storm/site/publish/releases/2.0.0-SNAPSHOT/storm-kafka.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/2.0.0-SNAPSHOT/storm-kafka.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/2.0.0-SNAPSHOT/storm-kafka.html (original)
+++ storm/site/publish/releases/2.0.0-SNAPSHOT/storm-kafka.html Wed Jun 1 06:20:57 2016
@@ -153,81 +153,77 @@ Currently, we support the following two
<h4 id="zkhosts">ZkHosts</h4>
<p>ZkHosts is what you should use if you want to dynamically track Kafka broker to partition mapping. This class uses
-Kafka's ZooKeeper entries to track brokerHost -> partition mapping. You can instantiate an object by calling
-<code>java
- public ZkHosts(String brokerZkStr, String brokerZkPath)
- public ZkHosts(String brokerZkStr)
-</code>
-Where brokerZkStr is just ip:port (e.g. localhost:2181). brokerZkPath is the root directory under which all the topics and
-partition information is stored. By default this is /brokers which is what the default Kafka implementation uses.</p>
-
-<p>By default, the broker-partition mapping is refreshed every 60 seconds from ZooKeeper. If you want to change it, you
+Kafka's ZooKeeper entries to track brokerHost -> partition mapping. You can instantiate an object by calling</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="nf">ZkHosts</span><span class="p">(</span><span class="n">String</span> <span class="n">brokerZkStr</span><span class="o">,</span> <span class="n">String</span> <span class="n">brokerZkPath</span><span class="o">)</span>
+<span class="kd">public</span> <span class="nf">ZkHosts</span><span class="p">(</span><span class="n">String</span> <span class="n">brokerZkStr</span><span class="o">)</span>
+</code></pre></div>
+<p>Where brokerZkStr is just ip:port (e.g. localhost:2181). brokerZkPath is the root directory under which all the topics and
+partition information is stored. By default this is /brokers which is what the default Kafka implementation uses.
+By default, the broker-partition mapping is refreshed every 60 seconds from ZooKeeper. If you want to change it, you
should set host.refreshFreqSecs to your chosen value.</p>
<h4 id="statichosts">StaticHosts</h4>
<p>This is an alternative implementation where broker -> partition information is static. In order to construct an instance
of this class, you need to first construct an instance of GlobalPartitionInformation.</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">Broker</span> <span class="n">brokerForPartition0</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Broker</span><span class="o">(</span><span class="s">"localhost"</span><span class="o">);</span><span class="c1">//localhost:9092</span>
- <span class="n">Broker</span> <span class="n">brokerForPartition1</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Broker</span><span class="o">(</span><span class="s">"localhost"</span><span class="o">,</span> <span class="mi">9092</span><span class="o">);</span><span class="c1">//localhost:9092 but we specified the port explicitly</span>
- <span class="n">Broker</span> <span class="n">brokerForPartition2</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Broker</span><span class="o">(</span><span class="s">"localhost:9092"</span><span class="o">);</span><span class="c1">//localhost:9092 specified as one string.</span>
- <span class="n">GlobalPartitionInformation</span> <span class="n">partitionInfo</span> <span class="o">=</span> <span class="k">new</span> <span class="n">GlobalPartitionInformation</span><span class="o">();</span>
- <span class="n">partitionInfo</span><span class="o">.</span><span class="na">addPartition</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">brokerForPartition0</span><span class="o">);</span><span class="c1">//mapping from partition 0 to brokerForPartition0</span>
- <span class="n">partitionInfo</span><span class="o">.</span><span class="na">addPartition</span><span class="o">(</span><span class="mi">1</span><span class="o">,</span> <span class="n">brokerForPartition1</span><span class="o">);</span><span class="c1">//mapping from partition 1 to brokerForPartition1</span>
- <span class="n">partitionInfo</span><span class="o">.</span><span class="na">addPartition</span><span class="o">(</span><span class="mi">2</span><span class="o">,</span> <span class="n">brokerForPartition2</span><span class="o">);</span><span class="c1">//mapping from partition 2 to brokerForPartition2</span>
- <span class="n">StaticHosts</span> <span class="n">hosts</span> <span class="o">=</span> <span class="k">new</span> <span class="n">StaticHosts</span><span class="o">(</span><span class="n">partitionInfo</span><span class="o">);</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">Broker</span> <span class="n">brokerForPartition0</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Broker</span><span class="o">(</span><span class="s">"localhost"</span><span class="o">);</span><span class="c1">//localhost:9092</span>
+<span class="n">Broker</span> <span class="n">brokerForPartition1</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Broker</span><span class="o">(</span><span class="s">"localhost"</span><span class="o">,</span> <span class="mi">9092</span><span class="o">);</span><span class="c1">//localhost:9092 but we specified the port explicitly</span>
+<span class="n">Broker</span> <span class="n">brokerForPartition2</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Broker</span><span class="o">(</span><span class="s">"localhost:9092"</span><span class="o">);</span><span class="c1">//localhost:9092 specified as one string.</span>
+<span class="n">GlobalPartitionInformation</span> <span class="n">partitionInfo</span> <span class="o">=</span> <span class="k">new</span> <span class="n">GlobalPartitionInformation</span><span class="o">();</span>
+<span class="n">partitionInfo</span><span class="o">.</span><span class="na">addPartition</span><span class="o">(</span><span class="mi">0</span><span class="o">,</span> <span class="n">brokerForPartition0</span><span class="o">);</span><span class="c1">//mapping from partition 0 to brokerForPartition0</span>
+<span class="n">partitionInfo</span><span class="o">.</span><span class="na">addPartition</span><span class="o">(</span><span class="mi">1</span><span class="o">,</span> <span class="n">brokerForPartition1</span><span class="o">);</span><span class="c1">//mapping from partition 1 to brokerForPartition1</span>
+<span class="n">partitionInfo</span><span class="o">.</span><span class="na">addPartition</span><span class="o">(</span><span class="mi">2</span><span class="o">,</span> <span class="n">brokerForPartition2</span><span class="o">);</span><span class="c1">//mapping from partition 2 to brokerForPartition2</span>
+<span class="n">StaticHosts</span> <span class="n">hosts</span> <span class="o">=</span> <span class="k">new</span> <span class="n">StaticHosts</span><span class="o">(</span><span class="n">partitionInfo</span><span class="o">);</span>
</code></pre></div>
<h3 id="kafkaconfig">KafkaConfig</h3>
-<p>The second thing needed for constructing a kafkaSpout is an instance of KafkaConfig.
-<code>java
- public KafkaConfig(BrokerHosts hosts, String topic)
- public KafkaConfig(BrokerHosts hosts, String topic, String clientId)
-</code></p>
-
+<p>The second thing needed for constructing a KafkaSpout is an instance of KafkaConfig.</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="nf">KafkaConfig</span><span class="p">(</span><span class="n">BrokerHosts</span> <span class="n">hosts</span><span class="o">,</span> <span class="n">String</span> <span class="n">topic</span><span class="o">)</span>
+<span class="kd">public</span> <span class="nf">KafkaConfig</span><span class="p">(</span><span class="n">BrokerHosts</span> <span class="n">hosts</span><span class="o">,</span> <span class="n">String</span> <span class="n">topic</span><span class="o">,</span> <span class="n">String</span> <span class="n">clientId</span><span class="o">)</span>
+</code></pre></div>
<p>The BrokerHosts can be any implementation of BrokerHosts interface as described above. The topic is name of Kafka topic.
-The optional ClientId is used as a part of the ZooKeeper path where the spout's current consumption offset is stored.</p>
+The optional ClientId is used as a part of the ZooKeeper path where the spout's current consumption offset is stored.
+There are 2 extensions of KafkaConfig currently in use.
+SpoutConfig is an extension of KafkaConfig that supports additional fields with ZooKeeper connection info and for controlling
+behavior specific to KafkaSpout. The Zkroot will be used as root to store your consumer's offset. The id should uniquely
+identify your spout.</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="nf">SpoutConfig</span><span class="p">(</span><span class="n">BrokerHosts</span> <span class="n">hosts</span><span class="o">,</span> <span class="n">String</span> <span class="n">topic</span><span class="o">,</span> <span class="n">String</span> <span class="n">zkRoot</span><span class="o">,</span> <span class="n">String</span> <span class="n">id</span><span class="o">);</span>
+<span class="kd">public</span> <span class="nf">SpoutConfig</span><span class="p">(</span><span class="n">BrokerHosts</span> <span class="n">hosts</span><span class="o">,</span> <span class="n">String</span> <span class="n">topic</span><span class="o">,</span> <span class="n">String</span> <span class="n">id</span><span class="o">);</span>
+</code></pre></div>
+<p>You need to use the correct SpoutConifg for different use cases:</p>
-<p>There are 2 extensions of KafkaConfig currently in use.</p>
+<ul>
+<li>Core KafkaSpout only accepts an instance of SpoutConfig.</li>
+<li>TridentKafkaConfig is another extension of KafkaConfig.</li>
+<li>TridentKafkaEmitter only accepts TridentKafkaConfig.</li>
+</ul>
+
+<p>In addition to these parameters, SpoutConfig contains the following fields that control how KafkaSpout behaves:</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// setting for how often to save the current Kafka offset to ZooKeeper</span>
+<span class="kd">public</span> <span class="kt">long</span> <span class="n">stateUpdateIntervalMs</span> <span class="o">=</span> <span class="mi">2000</span><span class="o">;</span>
+
+<span class="c1">// Exponential back-off retry settings. These are used when retrying messages after a bolt</span>
+<span class="c1">// calls OutputCollector.fail().</span>
+<span class="c1">// Note: be sure to set org.apache.storm.Config.MESSAGE_TIMEOUT_SECS appropriately to prevent</span>
+<span class="c1">// resubmitting the message while still retrying.</span>
+<span class="kd">public</span> <span class="kt">long</span> <span class="n">retryInitialDelayMs</span> <span class="o">=</span> <span class="mi">0</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">double</span> <span class="n">retryDelayMultiplier</span> <span class="o">=</span> <span class="mf">1.0</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">long</span> <span class="n">retryDelayMaxMs</span> <span class="o">=</span> <span class="mi">60</span> <span class="o">*</span> <span class="mi">1000</span><span class="o">;</span>
-<p>Spoutconfig is an extension of KafkaConfig that supports additional fields with ZooKeeper connection info and for controlling
-behavior specific to KafkaSpout. The Zkroot will be used as root to store your consumer's offset. The id should uniquely
-identify your spout.
-<code>java
-public SpoutConfig(BrokerHosts hosts, String topic, String zkRoot, String id);
-public SpoutConfig(BrokerHosts hosts, String topic, String id);
-</code>
-In addition to these parameters, SpoutConfig contains the following fields that control how KafkaSpout behaves:
-```java
- // setting for how often to save the current Kafka offset to ZooKeeper
- public long stateUpdateIntervalMs = 2000;</p>
-<div class="highlight"><pre><code class="language-" data-lang="">// Exponential back-off retry settings. These are used when retrying messages after a bolt
-// calls OutputCollector.fail().
-// Note: be sure to set org.apache.storm.Config.MESSAGE_TIMEOUT_SECS appropriately to prevent
-// resubmitting the message while still retrying.
-public long retryInitialDelayMs = 0;
-public double retryDelayMultiplier = 1.0;
-public long retryDelayMaxMs = 60 * 1000;
-
-// if set to true, spout will set Kafka topic as the emitted Stream ID
-public boolean topicAsStreamId = false;
-</code></pre></div><div class="highlight"><pre><code class="language-" data-lang="">Core KafkaSpout only accepts an instance of SpoutConfig.
-
-TridentKafkaConfig is another extension of KafkaConfig.
-TridentKafkaEmitter only accepts TridentKafkaConfig.
-
-The KafkaConfig class also has bunch of public variables that controls your application's behavior. Here are defaults:
-```java
- public int fetchSizeBytes = 1024 * 1024;
- public int socketTimeoutMs = 10000;
- public int fetchMaxWait = 10000;
- public int bufferSizeBytes = 1024 * 1024;
- public MultiScheme scheme = new RawMultiScheme();
- public boolean ignoreZkOffsets = false;
- public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
- public long maxOffsetBehind = Long.MAX_VALUE;
- public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
- public int metricsTimeBucketSizeInSecs = 60;
+<span class="c1">// if set to true, spout will set Kafka topic as the emitted Stream ID</span>
+<span class="kd">public</span> <span class="kt">boolean</span> <span class="n">topicAsStreamId</span> <span class="o">=</span> <span class="kc">false</span><span class="o">;</span>
+</code></pre></div>
+<p>The KafkaConfig class also has bunch of public variables that controls your application's behavior. Here are defaults:</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kt">int</span> <span class="n">fetchSizeBytes</span> <span class="o">=</span> <span class="mi">1024</span> <span class="o">*</span> <span class="mi">1024</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">int</span> <span class="n">socketTimeoutMs</span> <span class="o">=</span> <span class="mi">10000</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">int</span> <span class="n">fetchMaxWait</span> <span class="o">=</span> <span class="mi">10000</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">int</span> <span class="n">bufferSizeBytes</span> <span class="o">=</span> <span class="mi">1024</span> <span class="o">*</span> <span class="mi">1024</span><span class="o">;</span>
+<span class="kd">public</span> <span class="n">MultiScheme</span> <span class="n">scheme</span> <span class="o">=</span> <span class="k">new</span> <span class="n">RawMultiScheme</span><span class="o">();</span>
+<span class="kd">public</span> <span class="kt">boolean</span> <span class="n">ignoreZkOffsets</span> <span class="o">=</span> <span class="kc">false</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">long</span> <span class="n">startOffsetTime</span> <span class="o">=</span> <span class="n">kafka</span><span class="o">.</span><span class="na">api</span><span class="o">.</span><span class="na">OffsetRequest</span><span class="o">.</span><span class="na">EarliestTime</span><span class="o">();</span>
+<span class="kd">public</span> <span class="kt">long</span> <span class="n">maxOffsetBehind</span> <span class="o">=</span> <span class="n">Long</span><span class="o">.</span><span class="na">MAX_VALUE</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">boolean</span> <span class="n">useStartOffsetTimeIfOffsetOutOfRange</span> <span class="o">=</span> <span class="kc">true</span><span class="o">;</span>
+<span class="kd">public</span> <span class="kt">int</span> <span class="n">metricsTimeBucketSizeInSecs</span> <span class="o">=</span> <span class="mi">60</span><span class="o">;</span>
</code></pre></div>
<p>Most of them are self explanatory except MultiScheme.</p>
@@ -235,8 +231,8 @@ The KafkaConfig class also has bunch of
<p>MultiScheme is an interface that dictates how the byte[] consumed from Kafka gets transformed into a storm tuple. It
also controls the naming of your output field.</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="kd">public</span> <span class="n">Iterable</span><span class="o"><</span><span class="n">List</span><span class="o"><</span><span class="n">Object</span><span class="o">>></span> <span class="n">deserialize</span><span class="o">(</span><span class="kt">byte</span><span class="o">[]</span> <span class="n">ser</span><span class="o">);</span>
- <span class="kd">public</span> <span class="n">Fields</span> <span class="n">getOutputFields</span><span class="o">();</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="n">Iterable</span><span class="o"><</span><span class="n">List</span><span class="o"><</span><span class="n">Object</span><span class="o">>></span> <span class="n">deserialize</span><span class="o">(</span><span class="kt">byte</span><span class="o">[]</span> <span class="n">ser</span><span class="o">);</span>
+<span class="kd">public</span> <span class="n">Fields</span> <span class="nf">getOutputFields</span><span class="p">(</span><span class="o">);</span>
</code></pre></div>
<p>The default <code>RawMultiScheme</code> just takes the <code>byte[]</code> and returns a tuple with <code>byte[]</code> as is. The name of the
outputField is "bytes". There are alternative implementation like <code>SchemeAsMultiScheme</code> and
@@ -293,21 +289,21 @@ as a transitive dependency. This allows
<p>When building a project with storm-kafka, you must explicitly add the Kafka dependency. For example, to
use Kafka 0.8.1.1 built against Scala 2.10, you would use the following dependency in your <code>pom.xml</code>:</p>
-<div class="highlight"><pre><code class="language-xml" data-lang="xml"> <span class="nt"><dependency></span>
- <span class="nt"><groupId></span>org.apache.kafka<span class="nt"></groupId></span>
- <span class="nt"><artifactId></span>kafka_2.10<span class="nt"></artifactId></span>
- <span class="nt"><version></span>0.8.1.1<span class="nt"></version></span>
- <span class="nt"><exclusions></span>
- <span class="nt"><exclusion></span>
- <span class="nt"><groupId></span>org.apache.zookeeper<span class="nt"></groupId></span>
- <span class="nt"><artifactId></span>zookeeper<span class="nt"></artifactId></span>
- <span class="nt"></exclusion></span>
- <span class="nt"><exclusion></span>
- <span class="nt"><groupId></span>log4j<span class="nt"></groupId></span>
- <span class="nt"><artifactId></span>log4j<span class="nt"></artifactId></span>
- <span class="nt"></exclusion></span>
- <span class="nt"></exclusions></span>
- <span class="nt"></dependency></span>
+<div class="highlight"><pre><code class="language-xml" data-lang="xml"><span class="nt"><dependency></span>
+ <span class="nt"><groupId></span>org.apache.kafka<span class="nt"></groupId></span>
+ <span class="nt"><artifactId></span>kafka_2.10<span class="nt"></artifactId></span>
+ <span class="nt"><version></span>0.8.1.1<span class="nt"></version></span>
+ <span class="nt"><exclusions></span>
+ <span class="nt"><exclusion></span>
+ <span class="nt"><groupId></span>org.apache.zookeeper<span class="nt"></groupId></span>
+ <span class="nt"><artifactId></span>zookeeper<span class="nt"></artifactId></span>
+ <span class="nt"></exclusion></span>
+ <span class="nt"><exclusion></span>
+ <span class="nt"><groupId></span>log4j<span class="nt"></groupId></span>
+ <span class="nt"><artifactId></span>log4j<span class="nt"></artifactId></span>
+ <span class="nt"></exclusion></span>
+ <span class="nt"></exclusions></span>
+<span class="nt"></dependency></span>
</code></pre></div>
<p>Note that the ZooKeeper and log4j dependencies are excluded to prevent version conflicts with Storm's dependencies.</p>
@@ -322,8 +318,8 @@ org.apache.storm.kafka.trident.TridentKa
<h3 id="tupletokafkamapper-and-tridenttupletokafkamapper">TupleToKafkaMapper and TridentTupleToKafkaMapper</h3>
<p>These interfaces have 2 methods defined:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">K</span> <span class="nf">getKeyFromTuple</span><span class="p">(</span><span class="n">Tuple</span><span class="o">/</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">);</span>
- <span class="n">V</span> <span class="n">getMessageFromTuple</span><span class="o">(</span><span class="n">Tuple</span><span class="o">/</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">);</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">K</span> <span class="nf">getKeyFromTuple</span><span class="p">(</span><span class="n">Tuple</span><span class="o">/</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">);</span>
+<span class="n">V</span> <span class="nf">getMessageFromTuple</span><span class="p">(</span><span class="n">Tuple</span><span class="o">/</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">);</span>
</code></pre></div>
<p>As the name suggests, these methods are called to map a tuple to Kafka key and Kafka message. If you just want one field
as key and one field as value, then you can use the provided FieldNameBasedTupleToKafkaMapper.java
@@ -353,61 +349,61 @@ map with key kafka.broker.properties.</p
<h3 id="putting-it-all-together">Putting it all together</h3>
-<p>For the bolt :
-```java
- TopologyBuilder builder = new TopologyBuilder();</p>
-<div class="highlight"><pre><code class="language-" data-lang=""> Fields fields = new Fields("key", "message");
- FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
- new Values("storm", "1"),
- new Values("trident", "1"),
- new Values("needs", "1"),
- new Values("javadoc", "1")
- );
- spout.setCycle(true);
- builder.setSpout("spout", spout, 5);
- KafkaBolt bolt = new KafkaBolt()
- .withTopicSelector(new DefaultTopicSelector("test"))
- .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
- builder.setBolt("forwardToKafka", bolt, 8).shuffleGrouping("spout");
-
- Config conf = new Config();
- //set producer properties.
- Properties props = new Properties();
- props.put("metadata.broker.list", "localhost:9092");
- props.put("request.required.acks", "1");
- props.put("serializer.class", "kafka.serializer.StringEncoder");
- conf.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);
-
- StormSubmitter.submitTopology("kafkaboltTest", conf, builder.createTopology());
-</code></pre></div><div class="highlight"><pre><code class="language-" data-lang="">
-For Trident:
-
-```java
- Fields fields = new Fields("word", "count");
- FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
- new Values("storm", "1"),
- new Values("trident", "1"),
- new Values("needs", "1"),
- new Values("javadoc", "1")
- );
- spout.setCycle(true);
-
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("spout1", spout);
-
- TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
- .withKafkaTopicSelector(new DefaultTopicSelector("test"))
- .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count"));
- stream.partitionPersist(stateFactory, fields, new TridentKafkaUpdater(), new Fields());
-
- Config conf = new Config();
- //set producer properties.
- Properties props = new Properties();
- props.put("metadata.broker.list", "localhost:9092");
- props.put("request.required.acks", "1");
- props.put("serializer.class", "kafka.serializer.StringEncoder");
- conf.put(TridentKafkaState.KAFKA_BROKER_PROPERTIES, props);
- StormSubmitter.submitTopology("kafkaTridentTest", conf, topology.build());
+<p>For the bolt :</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">TopologyBuilder</span> <span class="n">builder</span> <span class="o">=</span> <span class="k">new</span> <span class="n">TopologyBuilder</span><span class="o">();</span>
+
+<span class="n">Fields</span> <span class="n">fields</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"key"</span><span class="o">,</span> <span class="s">"message"</span><span class="o">);</span>
+<span class="n">FixedBatchSpout</span> <span class="n">spout</span> <span class="o">=</span> <span class="k">new</span> <span class="n">FixedBatchSpout</span><span class="o">(</span><span class="n">fields</span><span class="o">,</span>
+ <span class="mi">4</span><span class="o">,</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"storm"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"trident"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"needs"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"javadoc"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">)</span>
+<span class="o">);</span>
+<span class="n">spout</span><span class="o">.</span><span class="na">setCycle</span><span class="o">(</span><span class="kc">true</span><span class="o">);</span>
+<span class="n">builder</span><span class="o">.</span><span class="na">setSpout</span><span class="o">(</span><span class="s">"spout"</span><span class="o">,</span> <span class="n">spout</span><span class="o">,</span> <span class="mi">5</span><span class="o">);</span>
+<span class="n">KafkaBolt</span> <span class="n">bolt</span> <span class="o">=</span> <span class="k">new</span> <span class="n">KafkaBolt</span><span class="o">()</span>
+ <span class="o">.</span><span class="na">withTopicSelector</span><span class="o">(</span><span class="k">new</span> <span class="n">DefaultTopicSelector</span><span class="o">(</span><span class="s">"test"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">withTupleToKafkaMapper</span><span class="o">(</span><span class="k">new</span> <span class="n">FieldNameBasedTupleToKafkaMapper</span><span class="o">());</span>
+<span class="n">builder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"forwardToKafka"</span><span class="o">,</span> <span class="n">bolt</span><span class="o">,</span> <span class="mi">8</span><span class="o">).</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"spout"</span><span class="o">);</span>
+
+<span class="n">Config</span> <span class="n">conf</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Config</span><span class="o">();</span>
+<span class="c1">//set producer properties.</span>
+<span class="n">Properties</span> <span class="n">props</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+<span class="n">props</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"metadata.broker.list"</span><span class="o">,</span> <span class="s">"localhost:9092"</span><span class="o">);</span>
+<span class="n">props</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"request.required.acks"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">);</span>
+<span class="n">props</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"serializer.class"</span><span class="o">,</span> <span class="s">"kafka.serializer.StringEncoder"</span><span class="o">);</span>
+<span class="n">conf</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">KafkaBolt</span><span class="o">.</span><span class="na">KAFKA_BROKER_PROPERTIES</span><span class="o">,</span> <span class="n">props</span><span class="o">);</span>
+
+<span class="n">StormSubmitter</span><span class="o">.</span><span class="na">submitTopology</span><span class="o">(</span><span class="s">"kafkaboltTest"</span><span class="o">,</span> <span class="n">conf</span><span class="o">,</span> <span class="n">builder</span><span class="o">.</span><span class="na">createTopology</span><span class="o">());</span>
+</code></pre></div>
+<p>For Trident:</p>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">Fields</span> <span class="n">fields</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">,</span> <span class="s">"count"</span><span class="o">);</span>
+<span class="n">FixedBatchSpout</span> <span class="n">spout</span> <span class="o">=</span> <span class="k">new</span> <span class="n">FixedBatchSpout</span><span class="o">(</span><span class="n">fields</span><span class="o">,</span>
+ <span class="mi">4</span><span class="o">,</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"storm"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"trident"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"needs"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"javadoc"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">)</span>
+<span class="o">);</span>
+<span class="n">spout</span><span class="o">.</span><span class="na">setCycle</span><span class="o">(</span><span class="kc">true</span><span class="o">);</span>
+
+<span class="n">TridentTopology</span> <span class="n">topology</span> <span class="o">=</span> <span class="k">new</span> <span class="n">TridentTopology</span><span class="o">();</span>
+<span class="n">Stream</span> <span class="n">stream</span> <span class="o">=</span> <span class="n">topology</span><span class="o">.</span><span class="na">newStream</span><span class="o">(</span><span class="s">"spout1"</span><span class="o">,</span> <span class="n">spout</span><span class="o">);</span>
+
+<span class="n">TridentKafkaStateFactory</span> <span class="n">stateFactory</span> <span class="o">=</span> <span class="k">new</span> <span class="n">TridentKafkaStateFactory</span><span class="o">()</span>
+ <span class="o">.</span><span class="na">withKafkaTopicSelector</span><span class="o">(</span><span class="k">new</span> <span class="n">DefaultTopicSelector</span><span class="o">(</span><span class="s">"test"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">withTridentTupleToKafkaMapper</span><span class="o">(</span><span class="k">new</span> <span class="n">FieldNameBasedTupleToKafkaMapper</span><span class="o">(</span><span class="s">"word"</span><span class="o">,</span> <span class="s">"count"</span><span class="o">));</span>
+<span class="n">stream</span><span class="o">.</span><span class="na">partitionPersist</span><span class="o">(</span><span class="n">stateFactory</span><span class="o">,</span> <span class="n">fields</span><span class="o">,</span> <span class="k">new</span> <span class="n">TridentKafkaUpdater</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">());</span>
+
+<span class="n">Config</span> <span class="n">conf</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Config</span><span class="o">();</span>
+<span class="c1">//set producer properties.</span>
+<span class="n">Properties</span> <span class="n">props</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+<span class="n">props</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"metadata.broker.list"</span><span class="o">,</span> <span class="s">"localhost:9092"</span><span class="o">);</span>
+<span class="n">props</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"request.required.acks"</span><span class="o">,</span> <span class="s">"1"</span><span class="o">);</span>
+<span class="n">props</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"serializer.class"</span><span class="o">,</span> <span class="s">"kafka.serializer.StringEncoder"</span><span class="o">);</span>
+<span class="n">conf</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">TridentKafkaState</span><span class="o">.</span><span class="na">KAFKA_BROKER_PROPERTIES</span><span class="o">,</span> <span class="n">props</span><span class="o">);</span>
+<span class="n">StormSubmitter</span><span class="o">.</span><span class="na">submitTopology</span><span class="o">(</span><span class="s">"kafkaTridentTest"</span><span class="o">,</span> <span class="n">conf</span><span class="o">,</span> <span class="n">topology</span><span class="o">.</span><span class="na">build</span><span class="o">());</span>
</code></pre></div>
Added: storm/site/publish/releases/2.0.0-SNAPSHOT/storm-metrics-profiling-internal-actions.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/2.0.0-SNAPSHOT/storm-metrics-profiling-internal-actions.html?rev=1746399&view=auto
==============================================================================
--- storm/site/publish/releases/2.0.0-SNAPSHOT/storm-metrics-profiling-internal-actions.html (added)
+++ storm/site/publish/releases/2.0.0-SNAPSHOT/storm-metrics-profiling-internal-actions.html Wed Jun 1 06:20:57 2016
@@ -0,0 +1,285 @@
+<!DOCTYPE html>
+<html>
+ <head>
+ <meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
+ <meta name="viewport" content="width=device-width, initial-scale=1">
+
+ <link rel="shortcut icon" href="/favicon.ico" type="image/x-icon">
+ <link rel="icon" href="/favicon.ico" type="image/x-icon">
+
+ <title>Storm Metrics for Profiling Various Storm Internal Actions</title>
+
+ <!-- Bootstrap core CSS -->
+ <link href="/assets/css/bootstrap.min.css" rel="stylesheet">
+ <!-- Bootstrap theme -->
+ <link href="/assets/css/bootstrap-theme.min.css" rel="stylesheet">
+
+ <!-- Custom styles for this template -->
+ <link rel="stylesheet" href="http://fortawesome.github.io/Font-Awesome/assets/font-awesome/css/font-awesome.css">
+ <link href="/css/style.css" rel="stylesheet">
+ <link href="/assets/css/owl.theme.css" rel="stylesheet">
+ <link href="/assets/css/owl.carousel.css" rel="stylesheet">
+ <script type="text/javascript" src="/assets/js/jquery.min.js"></script>
+ <script type="text/javascript" src="/assets/js/bootstrap.min.js"></script>
+ <script type="text/javascript" src="/assets/js/owl.carousel.min.js"></script>
+ <script type="text/javascript" src="/assets/js/storm.js"></script>
+ <!-- Just for debugging purposes. Don't actually copy these 2 lines! -->
+ <!--[if lt IE 9]><script src="../../assets/js/ie8-responsive-file-warning.js"></script><![endif]-->
+
+ <!-- HTML5 shim and Respond.js for IE8 support of HTML5 elements and media queries -->
+ <!--[if lt IE 9]>
+ <script src="https://oss.maxcdn.com/html5shiv/3.7.2/html5shiv.min.js"></script>
+ <script src="https://oss.maxcdn.com/respond/1.4.2/respond.min.js"></script>
+ <![endif]-->
+ </head>
+
+
+ <body>
+ <header>
+ <div class="container-fluid">
+ <div class="row">
+ <div class="col-md-5">
+ <a href="/index.html"><img src="/images/logo.png" class="logo" /></a>
+ </div>
+ <div class="col-md-5">
+
+ <h1>Version: 2.0.0-SNAPSHOT</h1>
+
+ </div>
+ <div class="col-md-2">
+ <a href="/downloads.html" class="btn-std btn-block btn-download">Download</a>
+ </div>
+ </div>
+ </div>
+</header>
+<!--Header End-->
+<!--Navigation Begin-->
+<div class="navbar" role="banner">
+ <div class="container-fluid">
+ <div class="navbar-header">
+ <button class="navbar-toggle" type="button" data-toggle="collapse" data-target=".bs-navbar-collapse">
+ <span class="icon-bar"></span>
+ <span class="icon-bar"></span>
+ <span class="icon-bar"></span>
+ </button>
+ </div>
+ <nav class="collapse navbar-collapse bs-navbar-collapse" role="navigation">
+ <ul class="nav navbar-nav">
+ <li><a href="/index.html" id="home">Home</a></li>
+ <li><a href="/getting-help.html" id="getting-help">Getting Help</a></li>
+ <li><a href="/about/integrates.html" id="project-info">Project Information</a></li>
+ <li class="dropdown">
+ <a href="#" class="dropdown-toggle" data-toggle="dropdown" id="documentation">Documentation <b class="caret"></b></a>
+ <ul class="dropdown-menu">
+
+
+ <li><a href="/releases/2.0.0-SNAPSHOT/index.html">2.0.0-SNAPSHOT</a></li>
+
+
+
+
+
+ <li><a href="/releases/1.0.1/index.html">1.0.1</a></li>
+
+
+
+ <li><a href="/releases/1.0.0/index.html">1.0.0</a></li>
+
+
+
+ <li><a href="/releases/0.10.1/index.html">0.10.1</a></li>
+
+
+
+
+
+ <li><a href="/releases/0.10.0/index.html">0.10.0</a></li>
+
+
+
+ <li><a href="/releases/0.9.6/index.html">0.9.6</a></li>
+
+
+
+
+
+
+
+
+
+
+
+
+ </ul>
+ </li>
+ <li><a href="/talksAndVideos.html">Talks and Slideshows</a></li>
+ <li class="dropdown">
+ <a href="#" class="dropdown-toggle" data-toggle="dropdown" id="contribute">Community <b class="caret"></b></a>
+ <ul class="dropdown-menu">
+ <li><a href="/contribute/Contributing-to-Storm.html">Contributing</a></li>
+ <li><a href="/contribute/People.html">People</a></li>
+ <li><a href="/contribute/BYLAWS.html">ByLaws</a></li>
+ </ul>
+ </li>
+ <li><a href="/2016/05/06/storm101-released.html" id="news">News</a></li>
+ </ul>
+ </nav>
+ </div>
+</div>
+
+
+
+ <div class="container-fluid">
+ <h1 class="page-title">Storm Metrics for Profiling Various Storm Internal Actions</h1>
+ <div class="row">
+ <div class="col-md-12">
+ <!-- Documentation -->
+
+<p class="post-meta"></p>
+
+<h1 id="storm-metrics-for-profiling-various-storm-internal-actions">Storm Metrics for Profiling Various Storm Internal Actions</h1>
+
+<p>With the addition of these metrics, Storm users can collect, view, and analyze the performance of various internal actions. The actions that are profiled include thrift rpc calls and http quests within the storm daemons. For instance, in the Storm Nimbus daemon, the following thrift calls defined in the Nimbus$Iface are profiled:</p>
+
+<ul>
+<li>submitTopology</li>
+<li>submitTopologyWithOpts</li>
+<li>killTopology</li>
+<li>killTopologyWithOpts</li>
+<li>activate</li>
+<li>deactivate</li>
+<li>rebalance</li>
+<li>setLogConfig</li>
+<li>getLogConfig</li>
+</ul>
+
+<p>Various HTTP GET and POST requests are marked for profiling as well such as the GET and POST requests for the Storm UI daemon (ui/core.cj)
+To implement these metrics the following packages are used:
+- io.dropwizard.metrics
+- metrics-clojure</p>
+
+<h2 id="how-it-works">How it works</h2>
+
+<p>By using packages io.dropwizard.metrics and metrics-clojure (clojure wrapper for the metrics Java API), we can mark functions to profile by declaring (defmeter num-some-func-calls) and then adding the (mark! num-some-func-calls) to where the function is invoked. For example:</p>
+<div class="highlight"><pre><code class="language-" data-lang="">(defmeter num-some-func-calls)
+(defn some-func [args]
+ (mark! num-some-func-calls)
+ (body))
+</code></pre></div>
+<p>What essentially the mark! API call does is increment a counter that represents how many times a certain action occured. For instantanous measurements user can use gauges. For example: </p>
+<div class="highlight"><pre><code class="language-" data-lang="">(defgauge nimbus:num-supervisors
+ (fn [] (.size (.supervisors (:storm-cluster-state nimbus) nil))))
+</code></pre></div>
+<p>The above example will get the number of supervisors in the cluster. This metric is not accumlative like one previously discussed.</p>
+
+<p>A metrics reporting server needs to also be activated to collect the metrics. You can do this by calling the following function:</p>
+<div class="highlight"><pre><code class="language-" data-lang="">(defn start-metrics-reporters []
+ (jmx/start (jmx/reporter {})))
+</code></pre></div>
+<h2 id="how-to-collect-the-metrics">How to collect the metrics</h2>
+
+<p>Metrics can be reported via JMX or HTTP. A user can use JConsole or VisualVM to connect to the jvm process and view the stats.</p>
+
+<p>To view the metrics in a GUI use VisualVM or JConsole. Screenshot of using VisualVm for metrics: </p>
+
+<p><img src="images/viewing_metrics_with_VisualVM.png" alt="Viewing metrics with VisualVM"></p>
+
+<p>For detailed information regarding how to collect the metrics please reference: </p>
+
+<p><a href="https://dropwizard.github.io/metrics/3.1.0/getting-started/">https://dropwizard.github.io/metrics/3.1.0/getting-started/</a></p>
+
+<p>If you want use JMX and view metrics through JConsole or VisualVM, remember launch JVM processes your want to profile with the correct JMX configurations. For example in Storm you would add the following to conf/storm.yaml</p>
+<div class="highlight"><pre><code class="language-" data-lang="">nimbus.childopts: "-Xmx1024m -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=3333 -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
+
+ui.childopts: "-Xmx768m -Dcom.sun.management.jmxremote.port=3334 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
+
+logviewer.childopts: "-Xmx128m -Dcom.sun.management.jmxremote.port=3335 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
+
+drpc.childopts: "-Xmx768m -Dcom.sun.management.jmxremote.port=3336 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
+
+supervisor.childopts: "-Xmx256m -Dcom.sun.management.jmxremote.port=3337 -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
+</code></pre></div>
+<h3 id="please-note">Please Note:</h3>
+
+<p>Since we shade all of the packages we use, additional plugins for collecting metrics might not work at this time. Currently collecting the metrics via JMX is supported.</p>
+
+<p>For more information about io.dropwizard.metrics and metrics-clojure packages please reference their original documentation:
+- <a href="https://dropwizard.github.io/metrics/3.1.0/">https://dropwizard.github.io/metrics/3.1.0/</a>
+- <a href="http://metrics-clojure.readthedocs.org/en/latest/">http://metrics-clojure.readthedocs.org/en/latest/</a></p>
+
+
+
+ </div>
+ </div>
+ </div>
+<footer>
+ <div class="container-fluid">
+ <div class="row">
+ <div class="col-md-3">
+ <div class="footer-widget">
+ <h5>Meetups</h5>
+ <ul class="latest-news">
+
+ <li><a href="http://www.meetup.com/Apache-Storm-Apache-Kafka/">Apache Storm & Apache Kafka</a> <span class="small">(Sunnyvale, CA)</span></li>
+
+ <li><a href="http://www.meetup.com/Apache-Storm-Kafka-Users/">Apache Storm & Kafka Users</a> <span class="small">(Seattle, WA)</span></li>
+
+ <li><a href="http://www.meetup.com/New-York-City-Storm-User-Group/">NYC Storm User Group</a> <span class="small">(New York, NY)</span></li>
+
+ <li><a href="http://www.meetup.com/Bay-Area-Stream-Processing">Bay Area Stream Processing</a> <span class="small">(Emeryville, CA)</span></li>
+
+ <li><a href="http://www.meetup.com/Boston-Storm-Users/">Boston Realtime Data</a> <span class="small">(Boston, MA)</span></li>
+
+ <li><a href="http://www.meetup.com/storm-london">London Storm User Group</a> <span class="small">(London, UK)</span></li>
+
+ <!-- <li><a href="http://www.meetup.com/Apache-Storm-Kafka-Users/">Seatle, WA</a> <span class="small">(27 Jun 2015)</span></li> -->
+ </ul>
+ </div>
+ </div>
+ <div class="col-md-3">
+ <div class="footer-widget">
+ <h5>About Storm</h5>
+ <p>Storm integrates with any queueing system and any database system. Storm's spout abstraction makes it easy to integrate a new queuing system. Likewise, integrating Storm with database systems is easy.</p>
+ </div>
+ </div>
+ <div class="col-md-3">
+ <div class="footer-widget">
+ <h5>First Look</h5>
+ <ul class="footer-list">
+ <li><a href="/releases/current/Rationale.html">Rationale</a></li>
+ <li><a href="/releases/current/Tutorial.html">Tutorial</a></li>
+ <li><a href="/releases/current/Setting-up-development-environment.html">Setting up development environment</a></li>
+ <li><a href="/releases/current/Creating-a-new-Storm-project.html">Creating a new Storm project</a></li>
+ </ul>
+ </div>
+ </div>
+ <div class="col-md-3">
+ <div class="footer-widget">
+ <h5>Documentation</h5>
+ <ul class="footer-list">
+ <li><a href="/releases/current/index.html">Index</a></li>
+ <li><a href="/releases/current/javadocs/index.html">Javadoc</a></li>
+ <li><a href="/releases/current/FAQ.html">FAQ</a></li>
+ </ul>
+ </div>
+ </div>
+ </div>
+ <hr/>
+ <div class="row">
+ <div class="col-md-12">
+ <p align="center">Copyright © 2015 <a href="http://www.apache.org">Apache Software Foundation</a>. All Rights Reserved.
+ <br>Apache Storm, Apache, the Apache feather logo, and the Apache Storm project logos are trademarks of The Apache Software Foundation.
+ <br>All other marks mentioned may be trademarks or registered trademarks of their respective owners.</p>
+ </div>
+ </div>
+ </div>
+</footer>
+<!--Footer End-->
+<!-- Scroll to top -->
+<span class="totop"><a href="#"><i class="fa fa-angle-up"></i></a></span>
+
+</body>
+
+</html>
+
Modified: storm/site/publish/releases/current/Clojure-DSL.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Clojure-DSL.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Clojure-DSL.html (original)
+++ storm/site/publish/releases/current/Clojure-DSL.html Wed Jun 1 06:20:57 2016
@@ -154,7 +154,7 @@
<p>To define a topology, use the <code>topology</code> function. <code>topology</code> takes in two arguments: a map of "spout specs" and a map of "bolt specs". Each spout and bolt spec wires the code for the component into the topology by specifying things like inputs and parallelism.</p>
-<p>Let's take a look at an example topology definition <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/clj/storm/starter/clj/word_count.clj">from the storm-starter project</a>:</p>
+<p>Let's take a look at an example topology definition <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj">from the storm-starter project</a>:</p>
<div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">topology</span><span class="w">
</span><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="p">(</span><span class="nf">spout-spec</span><span class="w"> </span><span class="n">sentence-spout</span><span class="p">)</span><span class="w">
</span><span class="s">"2"</span><span class="w"> </span><span class="p">(</span><span class="nf">spout-spec</span><span class="w"> </span><span class="p">(</span><span class="nf">sentence-spout-parameterized</span><span class="w">
@@ -316,7 +316,7 @@
<p>If you leave out the option map, it defaults to {:prepare true}. The output declaration for <code>defspout</code> has the same syntax as <code>defbolt</code>.</p>
-<p>Here's an example <code>defspout</code> implementation from <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/clj/storm/starter/clj/word_count.clj">storm-starter</a>:</p>
+<p>Here's an example <code>defspout</code> implementation from <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj">storm-starter</a>:</p>
<div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">defspout</span><span class="w"> </span><span class="n">sentence-spout</span><span class="w"> </span><span class="p">[</span><span class="s">"sentence"</span><span class="p">]</span><span class="w">
</span><span class="p">[</span><span class="n">conf</span><span class="w"> </span><span class="n">context</span><span class="w"> </span><span class="n">collector</span><span class="p">]</span><span class="w">
</span><span class="p">(</span><span class="k">let</span><span class="w"> </span><span class="p">[</span><span class="n">sentences</span><span class="w"> </span><span class="p">[</span><span class="s">"a little brown dog"</span><span class="w">
Modified: storm/site/publish/releases/current/Common-patterns.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Common-patterns.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Common-patterns.html (original)
+++ storm/site/publish/releases/current/Common-patterns.html Wed Jun 1 06:20:57 2016
@@ -194,7 +194,7 @@
<span class="n">builder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"merge"</span><span class="o">,</span> <span class="k">new</span> <span class="n">MergeObjects</span><span class="o">())</span>
<span class="o">.</span><span class="na">globalGrouping</span><span class="o">(</span><span class="s">"rank"</span><span class="o">);</span>
</code></pre></div>
-<p>This pattern works because of the fields grouping done by the first bolt which gives the partitioning you need for this to be semantically correct. You can see an example of this pattern in storm-starter <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/storm/starter/RollingTopWords.java">here</a>.</p>
+<p>This pattern works because of the fields grouping done by the first bolt which gives the partitioning you need for this to be semantically correct. You can see an example of this pattern in storm-starter <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/org/apache/storm/starter/RollingTopWords.java">here</a>.</p>
<p>If however you have a known skew in the data being processed it can be advantageous to use partialKeyGrouping instead of fieldsGrouping. This will distribute the load for each key between two downstream bolts instead of a single one.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">builder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"count"</span><span class="o">,</span> <span class="k">new</span> <span class="n">CountObjects</span><span class="o">(),</span> <span class="n">parallelism</span><span class="o">)</span>
@@ -204,7 +204,7 @@
<span class="n">builder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"merge"</span><span class="o">,</span> <span class="k">new</span> <span class="n">MergeRanksObjects</span><span class="o">())</span>
<span class="o">.</span><span class="na">globalGrouping</span><span class="o">(</span><span class="s">"rank"</span><span class="o">);</span>
</code></pre></div>
-<p>The topology needs an extra layer of processing to aggregate the partial counts from the upstream bolts but this only processes aggregated values now so the bolt it is not subject to the load caused by the skewed data. You can see an example of this pattern in storm-starter <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/storm/starter/SkewedRollingTopWords.java">here</a>.</p>
+<p>The topology needs an extra layer of processing to aggregate the partial counts from the upstream bolts but this only processes aggregated values now so the bolt it is not subject to the load caused by the skewed data. You can see an example of this pattern in storm-starter <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/org/apache/storm/starter/SkewedRollingTopWords.java">here</a>.</p>
<h3 id="timecachemap-for-efficiently-keeping-a-cache-of-things-that-have-been-recently-updated">TimeCacheMap for efficiently keeping a cache of things that have been recently updated</h3>
Modified: storm/site/publish/releases/current/Distributed-RPC.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Distributed-RPC.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Distributed-RPC.html (original)
+++ storm/site/publish/releases/current/Distributed-RPC.html Wed Jun 1 06:20:57 2016
@@ -241,7 +241,7 @@
<p>A single reach computation can involve thousands of database calls and tens of millions of follower records during the computation. It's a really, really intense computation. As you're about to see, implementing this function on top of Storm is dead simple. On a single machine, reach can take minutes to compute; on a Storm cluster, you can compute reach for even the hardest URLs in a couple seconds.</p>
-<p>A sample reach topology is defined in storm-starter <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/storm/starter/ReachTopology.java">here</a>. Here's how you define the reach topology:</p>
+<p>A sample reach topology is defined in storm-starter <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/org/apache/storm/starter/ReachTopology.java">here</a>. Here's how you define the reach topology:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">LinearDRPCTopologyBuilder</span> <span class="n">builder</span> <span class="o">=</span> <span class="k">new</span> <span class="n">LinearDRPCTopologyBuilder</span><span class="o">(</span><span class="s">"reach"</span><span class="o">);</span>
<span class="n">builder</span><span class="o">.</span><span class="na">addBolt</span><span class="o">(</span><span class="k">new</span> <span class="n">GetTweeters</span><span class="o">(),</span> <span class="mi">3</span><span class="o">);</span>
<span class="n">builder</span><span class="o">.</span><span class="na">addBolt</span><span class="o">(</span><span class="k">new</span> <span class="n">GetFollowers</span><span class="o">(),</span> <span class="mi">12</span><span class="o">)</span>
Modified: storm/site/publish/releases/current/SECURITY.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/SECURITY.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/SECURITY.html (original)
+++ storm/site/publish/releases/current/SECURITY.html Wed Jun 1 06:20:57 2016
@@ -537,7 +537,7 @@ This config file also needs to be owned
it can do so by leveraging the impersonation feature.In order to submit topology as some other user , you can use <code>StormSubmitter.submitTopologyAs</code> API. Alternatively you can use <code>NimbusClient.getConfiguredClientAs</code>
to get a nimbus client as some other user and perform any nimbus action(i.e. kill/rebalance/activate/deactivate) using this client. </p>
-<p>To ensure only authorized users can perform impersonation you should start nimbus with <code>nimbus.impersonation.authorizer</code> set to <code>org.apache.storm.security.auth.authorizer.ImpersonationAuthorizer</code>.
+<p>Impersonation authorization is disabled by default which means any user can perform impersonation. To ensure only authorized users can perform impersonation you should start nimbus with <code>nimbus.impersonation.authorizer</code> set to <code>org.apache.storm.security.auth.authorizer.ImpersonationAuthorizer</code>.
The <code>ImpersonationAuthorizer</code> uses <code>nimbus.impersonation.acl</code> as the acl to authorize users. Following is a sample nimbus config for supporting impersonation:</p>
<div class="highlight"><pre><code class="language-yaml" data-lang="yaml"><span class="s">nimbus.impersonation.authorizer</span><span class="pi">:</span> <span class="s">org.apache.storm.security.auth.authorizer.ImpersonationAuthorizer</span>
<span class="s">nimbus.impersonation.acl</span><span class="pi">:</span>
Modified: storm/site/publish/releases/current/Setting-up-a-Storm-cluster.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Setting-up-a-Storm-cluster.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Setting-up-a-Storm-cluster.html (original)
+++ storm/site/publish/releases/current/Setting-up-a-Storm-cluster.html Wed Jun 1 06:20:57 2016
@@ -168,7 +168,7 @@
<p>Next you need to install Storm's dependencies on Nimbus and the worker machines. These are:</p>
<ol>
-<li>Java 6</li>
+<li>Java 7</li>
<li>Python 2.6.6</li>
</ol>
Modified: storm/site/publish/releases/current/State-checkpointing.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/State-checkpointing.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/State-checkpointing.html (original)
+++ storm/site/publish/releases/current/State-checkpointing.html Wed Jun 1 06:20:57 2016
@@ -228,6 +228,9 @@ is saved and then the checkpoint tuple i
streams before it saves its state so that the state represents a consistent state across the topology. Once the checkpoint spout receives
ACK from all the bolts, the state commit is complete and the transaction is recorded as committed by the checkpoint spout.</p>
+<p>The state checkpointing does not currently checkpoint the state of the spout. Yet, once the state of all bolts are checkpointed, and once the checkpoint tuples are acked, the tuples emitted by the spout are also acked.
+It also implies that <code>topology.state.checkpoint.interval.ms</code> is lower than <code>topology.message.timeout.secs</code>. </p>
+
<p>The state commit works like a three phase commit protocol with a prepare and commit phase so that the state across the topology is saved
in a consistent and atomic manner.</p>
Modified: storm/site/publish/releases/current/Transactional-topologies.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Transactional-topologies.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Transactional-topologies.html (original)
+++ storm/site/publish/releases/current/Transactional-topologies.html Wed Jun 1 06:20:57 2016
@@ -224,7 +224,7 @@
<h2 id="the-basics-through-example">The basics through example</h2>
-<p>You build transactional topologies by using <a href="javadocs/org/apache/storm/transactional/TransactionalTopologyBuilder.html">TransactionalTopologyBuilder</a>. Here's the transactional topology definition for a topology that computes the global count of tuples from the input stream. This code comes from <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/storm/starter/TransactionalGlobalCount.java">TransactionalGlobalCount</a> in storm-starter.</p>
+<p>You build transactional topologies by using <a href="javadocs/org/apache/storm/transactional/TransactionalTopologyBuilder.html">TransactionalTopologyBuilder</a>. Here's the transactional topology definition for a topology that computes the global count of tuples from the input stream. This code comes from <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/org/apache/storm/starter/TransactionalGlobalCount.java">TransactionalGlobalCount</a> in storm-starter.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">MemoryTransactionalSpout</span> <span class="n">spout</span> <span class="o">=</span> <span class="k">new</span> <span class="n">MemoryTransactionalSpout</span><span class="o">(</span><span class="n">DATA</span><span class="o">,</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">),</span> <span class="n">PARTITION_TAKE_PER_BATCH</span><span class="o">);</span>
<span class="n">TransactionalTopologyBuilder</span> <span class="n">builder</span> <span class="o">=</span> <span class="k">new</span> <span class="n">TransactionalTopologyBuilder</span><span class="o">(</span><span class="s">"global-count"</span><span class="o">,</span> <span class="s">"spout"</span><span class="o">,</span> <span class="n">spout</span><span class="o">,</span> <span class="mi">3</span><span class="o">);</span>
<span class="n">builder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"partial-count"</span><span class="o">,</span> <span class="k">new</span> <span class="n">BatchCount</span><span class="o">(),</span> <span class="mi">5</span><span class="o">)</span>
@@ -332,7 +332,7 @@
<p>The code for <code>finishBatch</code> in <code>UpdateGlobalCount</code> gets the current value from the database and compares its transaction id to the transaction id for this batch. If they are the same, it does nothing. Otherwise, it increments the value in the database by the partial count for this batch.</p>
-<p>A more involved transactional topology example that updates multiple databases idempotently can be found in storm-starter in the <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/storm/starter/TransactionalWords.java">TransactionalWords</a> class.</p>
+<p>A more involved transactional topology example that updates multiple databases idempotently can be found in storm-starter in the <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/org/apache/storm/starter/TransactionalWords.java">TransactionalWords</a> class.</p>
<h2 id="transactional-topology-api">Transactional Topology API</h2>
@@ -388,7 +388,7 @@
<h4 id="partitioned-transactional-spout">Partitioned Transactional Spout</h4>
-<p>A common kind of transactional spout is one that reads the batches from a set of partitions across many queue brokers. For example, this is how <a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/storm/kafka/TransactionalKafkaSpout.java">TransactionalKafkaSpout</a> works. An <code>IPartitionedTransactionalSpout</code> automates the bookkeeping work of managing the state for each partition to ensure idempotent replayability. See <a href="javadocs/org/apache/storm/transactional/partitioned/IPartitionedTransactionalSpout.html">the Javadoc</a> for more details.</p>
+<p>A common kind of transactional spout is one that reads the batches from a set of partitions across many queue brokers. For example, this is how <a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/org/apache/storm/kafka/TransactionalKafkaSpout.java">TransactionalKafkaSpout</a> works. An <code>IPartitionedTransactionalSpout</code> automates the bookkeeping work of managing the state for each partition to ensure idempotent replayability. See <a href="javadocs/org/apache/storm/transactional/partitioned/IPartitionedTransactionalSpout.html">the Javadoc</a> for more details.</p>
<h3 id="configuration">Configuration</h3>
@@ -460,7 +460,7 @@
<p>By failing all subsequent transactions on failure, no tuples are skipped. This also shows that a requirement of transactional spouts is that they always emit where the last transaction left off.</p>
-<p>A non-idempotent transactional spout is more concisely referred to as an "OpaqueTransactionalSpout" (opaque is the opposite of idempotent). <a href="javadocs/org/apache/storm/transactional/partitioned/IOpaquePartitionedTransactionalSpout.html">IOpaquePartitionedTransactionalSpout</a> is an interface for implementing opaque partitioned transactional spouts, of which <a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/storm/kafka/OpaqueTransactionalKafkaSpout.java">OpaqueTransactionalKafkaSpout</a> is an example. <code>OpaqueTransactionalKafkaSpout</code> can withstand losing individual Kafka nodes without sacrificing accuracy as long as you use the update strategy as explained in this section.</p>
+<p>A non-idempotent transactional spout is more concisely referred to as an "OpaqueTransactionalSpout" (opaque is the opposite of idempotent). <a href="javadocs/org/apache/storm/transactional/partitioned/IOpaquePartitionedTransactionalSpout.html">IOpaquePartitionedTransactionalSpout</a> is an interface for implementing opaque partitioned transactional spouts, of which <a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/org/apache/storm/kafka/OpaqueTransactionalKafkaSpout.java">OpaqueTransactionalKafkaSpout</a> is an example. <code>OpaqueTransactionalKafkaSpout</code> can withstand losing individual Kafka nodes without sacrificing accuracy as long as you use the update strategy as explained in this section.</p>
<h2 id="implementation">Implementation</h2>
Modified: storm/site/publish/releases/current/Trident-state.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Trident-state.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Trident-state.html (original)
+++ storm/site/publish/releases/current/Trident-state.html Wed Jun 1 06:20:57 2016
@@ -166,7 +166,7 @@
<li>Every tuple is in a batch (no tuples are skipped)</li>
</ol>
-<p>This is a pretty easy type of spout to understand, the stream is divided into fixed batches that never change. storm-contrib has <a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/storm/kafka/trident/TransactionalTridentKafkaSpout.java">an implementation of a transactional spout</a> for Kafka.</p>
+<p>This is a pretty easy type of spout to understand, the stream is divided into fixed batches that never change. storm-contrib has <a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TransactionalTridentKafkaSpout.java">an implementation of a transactional spout</a> for Kafka.</p>
<p>You might be wondering – why wouldn't you just always use a transactional spout? They're simple and easy to understand. One reason you might not use one is because they're not necessarily very fault-tolerant. For example, the way TransactionalTridentKafkaSpout works is the batch for a txid will contain tuples from all the Kafka partitions for a topic. Once a batch has been emitted, any time that batch is re-emitted in the future the exact same set of tuples must be emitted to meet the semantics of transactional spouts. Now suppose a batch is emitted from TransactionalTridentKafkaSpout, the batch fails to process, and at the same time one of the Kafka nodes goes down. You're now incapable of replaying the same batch as you did before (since the node is down and some partitions for the topic are not unavailable), and processing will halt. </p>
@@ -203,7 +203,7 @@ apple => [count=10, txid=2]
<li>Every tuple is <em>successfully</em> processed in exactly one batch. However, it's possible for a tuple to fail to process in one batch and then succeed to process in a later batch.</li>
</ol>
-<p><a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/storm/kafka/trident/OpaqueTridentKafkaSpout.java">OpaqueTridentKafkaSpout</a> is a spout that has this property and is fault-tolerant to losing Kafka nodes. Whenever it's time for OpaqueTridentKafkaSpout to emit a batch, it emits tuples starting from where the last batch finished emitting. This ensures that no tuple is ever skipped or successfully processed by multiple batches.</p>
+<p><a href="http://github.com/apache/storm/tree/v1.0.1/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/OpaqueTridentKafkaSpout.java">OpaqueTridentKafkaSpout</a> is a spout that has this property and is fault-tolerant to losing Kafka nodes. Whenever it's time for OpaqueTridentKafkaSpout to emit a batch, it emits tuples starting from where the last batch finished emitting. This ensures that no tuple is ever skipped or successfully processed by multiple batches.</p>
<p>With opaque transactional spouts, it's no longer possible to use the trick of skipping state updates if the transaction id in the database is the same as the transaction id for the current batch. This is because the batch may have changed between state updates.</p>
Modified: storm/site/publish/releases/current/Tutorial.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/Tutorial.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/Tutorial.html (original)
+++ storm/site/publish/releases/current/Tutorial.html Wed Jun 1 06:20:57 2016
@@ -358,7 +358,7 @@
<p>When a task for Bolt A emits a tuple to Bolt B, which task should it send the tuple to?</p>
-<p>A "stream grouping" answers this question by telling Storm how to send tuples between sets of tasks. Before we dig into the different kinds of stream groupings, let's take a look at another topology from <a href="http://github.com/apache/storm/blob/1.0.1/examples/storm-starter">storm-starter</a>. This <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/storm/starter/WordCountTopology.java">WordCountTopology</a> reads sentences off of a spout and streams out of <code>WordCountBolt</code> the total number of times it has seen that word before:</p>
+<p>A "stream grouping" answers this question by telling Storm how to send tuples between sets of tasks. Before we dig into the different kinds of stream groupings, let's take a look at another topology from <a href="http://github.com/apache/storm/blob/1.0.1/examples/storm-starter">storm-starter</a>. This <a href="http://github.com/apache/storm/blob/v1.0.1/examples/storm-starter/src/jvm/org/apache/storm/starter/WordCountTopology.java">WordCountTopology</a> reads sentences off of a spout and streams out of <code>WordCountBolt</code> the total number of times it has seen that word before:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">TopologyBuilder</span> <span class="n">builder</span> <span class="o">=</span> <span class="k">new</span> <span class="n">TopologyBuilder</span><span class="o">();</span>
<span class="n">builder</span><span class="o">.</span><span class="na">setSpout</span><span class="o">(</span><span class="s">"sentences"</span><span class="o">,</span> <span class="k">new</span> <span class="n">RandomSentenceSpout</span><span class="o">(),</span> <span class="mi">5</span><span class="o">);</span>
Modified: storm/site/publish/releases/current/index.html
URL: http://svn.apache.org/viewvc/storm/site/publish/releases/current/index.html?rev=1746399&r1=1746398&r2=1746399&view=diff
==============================================================================
--- storm/site/publish/releases/current/index.html (original)
+++ storm/site/publish/releases/current/index.html Wed Jun 1 06:20:57 2016
@@ -138,6 +138,18 @@
<p class="post-meta"></p>
+<blockquote>
+<h4 id="note">NOTE</h4>
+
+<p>In the latest version, the class packages have been changed from "backtype.storm" to "org.apache.storm" so the topology code compiled with older version won't run on the Storm 1.0.0 just like that. Backward compatibility is available through following configuration </p>
+
+<p><code>client.jartransformer.class: "org.apache.storm.hack.StormShadeTransformer"</code></p>
+
+<p>You need to add the above config in storm installation if you want to run the code compiled with older versions of storm. The config should be added in the machine you use to submit your topologies.</p>
+
+<p>Refer to <a href="https://issues.apache.org/jira/browse/STORM-1202">https://issues.apache.org/jira/browse/STORM-1202</a> for more details. </p>
+</blockquote>
+
<h3 id="basics-of-storm">Basics of Storm</h3>
<ul>