You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by wh...@apache.org on 2016/01/12 23:13:23 UTC
[05/10] storm git commit: STORM-1366. Add documentation for StormSQL
into the website.
http://git-wip-us.apache.org/repos/asf/storm/blob/dfa66d7d/_site/documentation/Trident-tutorial.html
----------------------------------------------------------------------
diff --git a/_site/documentation/Trident-tutorial.html b/_site/documentation/Trident-tutorial.html
index ef7ea78..a0f3a1b 100644
--- a/_site/documentation/Trident-tutorial.html
+++ b/_site/documentation/Trident-tutorial.html
@@ -102,20 +102,20 @@
</ol>
<p>For the purposes of illustration, this example will read an infinite stream of sentences from the following source:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">FixedBatchSpout</span> <span class="n">spout</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">FixedBatchSpout</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"sentence"</span><span class="o">),</span> <span class="mi">3</span><span class="o">,</span>
- <span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="s">"the cow jumped over the moon"</span><span class="o">),</span>
- <span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="s">"the man went to the store and bought some candy"</span><span class="o">),</span>
- <span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="s">"four score and seven years ago"</span><span class="o">),</span>
- <span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="s">"how many apples can you eat"</span><span class="o">));</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><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="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"sentence"</span><span class="o">),</span> <span class="mi">3</span><span class="o">,</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"the cow jumped over the moon"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"the man went to the store and bought some candy"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"four score and seven years ago"</span><span class="o">),</span>
+ <span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="s">"how many apples can you eat"</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>
</code></pre></div>
<p>This spout cycles through that set of sentences over and over to produce the sentence stream. Here's the code to do the streaming word count part of the computation:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">TridentTopology</span> <span class="n">topology</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">TridentTopology</span><span class="o">();</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><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">TridentState</span> <span class="n">wordCounts</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="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"sentence"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">Split</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">persistentAggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">MemoryMapState</span><span class="o">.</span><span class="na">Factory</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Count</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"count"</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="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"sentence"</span><span class="o">),</span> <span class="k">new</span> <span class="n">Split</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="o">.</span><span class="na">groupBy</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="o">.</span><span class="na">persistentAggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">MemoryMapState</span><span class="o">.</span><span class="na">Factory</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Count</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">))</span>
<span class="o">.</span><span class="na">parallelismHint</span><span class="o">(</span><span class="mi">6</span><span class="o">);</span>
</code></pre></div>
<p>Let's go through the code line by line. First a TridentTopology object is created, which exposes the interface for constructing Trident computations. TridentTopology has a method called newStream that creates a new stream of data in the topology reading from an input source. In this case, the input source is just the FixedBatchSpout defined from before. Input sources can also be queue brokers like Kestrel or Kafka. Trident keeps track of a small amount of state for each input source (metadata about what it has consumed) in Zookeeper, and the "spout1" string here specifies the node in Zookeeper where Trident should keep that metadata.</p>
@@ -130,10 +130,10 @@
<p>Back to the example, the spout emits a stream containing one field called "sentence". The next line of the topology definition applies the Split function to each tuple in the stream, taking the "sentence" field and splitting it into words. Each sentence tuple creates potentially many word tuples – for instance, the sentence "the cow jumped over the moon" creates six "word" tuples. Here's the definition of Split:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">class</span> <span class="nc">Split</span> <span class="kd">extends</span> <span class="n">BaseFunction</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">execute</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">,</span> <span class="n">TridentCollector</span> <span class="n">collector</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">void</span> <span class="n">execute</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">,</span> <span class="n">TridentCollector</span> <span class="n">collector</span><span class="o">)</span> <span class="o">{</span>
<span class="n">String</span> <span class="n">sentence</span> <span class="o">=</span> <span class="n">tuple</span><span class="o">.</span><span class="na">getString</span><span class="o">(</span><span class="mi">0</span><span class="o">);</span>
- <span class="k">for</span><span class="o">(</span><span class="n">String</span> <span class="nl">word:</span> <span class="n">sentence</span><span class="o">.</span><span class="na">split</span><span class="o">(</span><span class="s">" "</span><span class="o">))</span> <span class="o">{</span>
- <span class="n">collector</span><span class="o">.</span><span class="na">emit</span><span class="o">(</span><span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="n">word</span><span class="o">));</span>
+ <span class="k">for</span><span class="o">(</span><span class="n">String</span> <span class="nl">word:</span> <span class="n">sentence</span><span class="o">.</span><span class="na">split</span><span class="o">(</span><span class="s">" "</span><span class="o">))</span> <span class="o">{</span>
+ <span class="n">collector</span><span class="o">.</span><span class="na">emit</span><span class="o">(</span><span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="n">word</span><span class="o">));</span>
<span class="o">}</span>
<span class="o">}</span>
<span class="o">}</span>
@@ -141,7 +141,7 @@
<p>As you can see, it's really simple. It simply grabs the sentence, splits it on whitespace, and emits a tuple for each word.</p>
<p>The rest of the topology computes word count and keeps the results persistently stored. First the stream is grouped by the "word" field. Then, each group is persistently aggregated using the Count aggregator. The persistentAggregate function knows how to store and update the results of the aggregation in a source of state. In this example, the word counts are kept in memory, but this can be trivially swapped to use Memcached, Cassandra, or any other persistent store. Swapping this topology to store counts in Memcached is as simple as replacing the persistentAggregate line with this (using <a href="https://github.com/nathanmarz/trident-memcached">trident-memcached</a>), where the "serverLocations" is a list of host/ports for the Memcached cluster:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="o">.</span><span class="na">persistentAggregate</span><span class="o">(</span><span class="n">MemcachedState</span><span class="o">.</span><span class="na">transactional</span><span class="o">(</span><span class="n">serverLocations</span><span class="o">),</span> <span class="k">new</span> <span class="nf">Count</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">))</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="o">.</span><span class="na">persistentAggregate</span><span class="o">(</span><span class="n">MemcachedState</span><span class="o">.</span><span class="na">transactional</span><span class="o">(</span><span class="n">serverLocations</span><span class="o">),</span> <span class="k">new</span> <span class="n">Count</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">))</span>
<span class="n">MemcachedState</span><span class="o">.</span><span class="na">transactional</span><span class="o">()</span>
</code></pre></div>
<p>The values stored by persistentAggregate represents the aggregation of all batches ever emitted by the stream.</p>
@@ -151,19 +151,19 @@
<p>The persistentAggregate method transforms a Stream into a TridentState object. In this case the TridentState object represents all the word counts. We will use this TridentState object to implement the distributed query portion of the computation.</p>
<p>The next part of the topology implements a low latency distributed query on the word counts. The query takes as input a whitespace separated list of words and return the sum of the counts for those words. These queries are executed just like normal RPC calls, except they are parallelized in the background. Here's an example of how you might invoke one of these queries:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DRPCClient</span> <span class="n">client</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">DRPCClient</span><span class="o">(</span><span class="s">"drpc.server.location"</span><span class="o">,</span> <span class="mi">3772</span><span class="o">);</span>
-<span class="n">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="n">client</span><span class="o">.</span><span class="na">execute</span><span class="o">(</span><span class="s">"words"</span><span class="o">,</span> <span class="s">"cat dog the man"</span><span class="o">);</span>
-<span class="c1">// prints the JSON-encoded result, e.g.: "[[5078]]"</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">DRPCClient</span> <span class="n">client</span> <span class="o">=</span> <span class="k">new</span> <span class="n">DRPCClient</span><span class="o">(</span><span class="s">"drpc.server.location"</span><span class="o">,</span> <span class="mi">3772</span><span class="o">);</span>
+<span class="n">System</span><span class="o">.</span><span class="na">out</span><span class="o">.</span><span class="na">println</span><span class="o">(</span><span class="n">client</span><span class="o">.</span><span class="na">execute</span><span class="o">(</span><span class="s">"words"</span><span class="o">,</span> <span class="s">"cat dog the man"</span><span class="o">);</span>
+<span class="c1">// prints the JSON-encoded result, e.g.: "[[5078]]"</span>
</code></pre></div>
<p>As you can see, it looks just like a regular remote procedure call (RPC), except it's executing in parallel across a Storm cluster. The latency for small queries like this are typically around 10ms. More intense DRPC queries can take longer of course, although the latency largely depends on how many resources you have allocated for the computation.</p>
<p>The implementation of the distributed query portion of the topology looks like this:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">topology</span><span class="o">.</span><span class="na">newDRPCStream</span><span class="o">(</span><span class="s">"words"</span><span class="o">)</span>
- <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"args"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">Split</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">stateQuery</span><span class="o">(</span><span class="n">wordCounts</span><span class="o">,</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"word"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">MapGet</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">FilterNull</span><span class="o">())</span>
- <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">Sum</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"sum"</span><span class="o">));</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">topology</span><span class="o">.</span><span class="na">newDRPCStream</span><span class="o">(</span><span class="s">"words"</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"args"</span><span class="o">),</span> <span class="k">new</span> <span class="n">Split</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="o">.</span><span class="na">groupBy</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="o">.</span><span class="na">stateQuery</span><span class="o">(</span><span class="n">wordCounts</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="k">new</span> <span class="n">MapGet</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">),</span> <span class="k">new</span> <span class="n">FilterNull</span><span class="o">())</span>
+ <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"count"</span><span class="o">),</span> <span class="k">new</span> <span class="n">Sum</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"sum"</span><span class="o">));</span>
</code></pre></div>
<p>The same TridentTopology object is used to create the DRPC stream, and the function is named "words". The function name corresponds to the function name given in the first argument of execute when using a DRPCClient.</p>
@@ -192,17 +192,17 @@
<span class="n">TridentState</span> <span class="n">tweetersToFollowers</span> <span class="o">=</span>
<span class="n">topology</span><span class="o">.</span><span class="na">newStaticState</span><span class="o">(</span><span class="n">getTweeterToFollowersState</span><span class="o">());</span>
-<span class="n">topology</span><span class="o">.</span><span class="na">newDRPCStream</span><span class="o">(</span><span class="s">"reach"</span><span class="o">)</span>
- <span class="o">.</span><span class="na">stateQuery</span><span class="o">(</span><span class="n">urlToTweeters</span><span class="o">,</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"args"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">MapGet</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"tweeters"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"tweeters"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">ExpandList</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"tweeter"</span><span class="o">))</span>
+<span class="n">topology</span><span class="o">.</span><span class="na">newDRPCStream</span><span class="o">(</span><span class="s">"reach"</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">stateQuery</span><span class="o">(</span><span class="n">urlToTweeters</span><span class="o">,</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"args"</span><span class="o">),</span> <span class="k">new</span> <span class="n">MapGet</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"tweeters"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"tweeters"</span><span class="o">),</span> <span class="k">new</span> <span class="n">ExpandList</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"tweeter"</span><span class="o">))</span>
<span class="o">.</span><span class="na">shuffle</span><span class="o">()</span>
- <span class="o">.</span><span class="na">stateQuery</span><span class="o">(</span><span class="n">tweetersToFollowers</span><span class="o">,</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"tweeter"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">MapGet</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"followers"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">stateQuery</span><span class="o">(</span><span class="n">tweetersToFollowers</span><span class="o">,</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"tweeter"</span><span class="o">),</span> <span class="k">new</span> <span class="n">MapGet</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"followers"</span><span class="o">))</span>
<span class="o">.</span><span class="na">parallelismHint</span><span class="o">(</span><span class="mi">200</span><span class="o">)</span>
- <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"followers"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">ExpandList</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"follower"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"follower"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="nf">One</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"one"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"followers"</span><span class="o">),</span> <span class="k">new</span> <span class="n">ExpandList</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"follower"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"follower"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">One</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"one"</span><span class="o">))</span>
<span class="o">.</span><span class="na">parallelismHint</span><span class="o">(</span><span class="mi">20</span><span class="o">)</span>
- <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="nf">Count</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"reach"</span><span class="o">));</span>
+ <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">Count</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"reach"</span><span class="o">));</span>
</code></pre></div>
<p>The topology creates TridentState objects representing each external database using the newStaticState method. These can then be queried in the topology. Like all sources of state, queries to these databases will be automatically batched for maximum efficiency.</p>
@@ -212,15 +212,15 @@
<p>Next, the set of followers is uniqued and counted. This is done in two steps. First a "group by" is done on the batch by "follower", running the "One" aggregator on each group. The "One" aggregator simply emits a single tuple containing the number one for each group. Then, the ones are summed together to get the unique count of the followers set. Here's the definition of the "One" aggregator:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">class</span> <span class="nc">One</span> <span class="kd">implements</span> <span class="n">CombinerAggregator</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="o">{</span>
- <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">init</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="n">Integer</span> <span class="n">init</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">)</span> <span class="o">{</span>
<span class="k">return</span> <span class="mi">1</span><span class="o">;</span>
<span class="o">}</span>
- <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">combine</span><span class="o">(</span><span class="n">Integer</span> <span class="n">val1</span><span class="o">,</span> <span class="n">Integer</span> <span class="n">val2</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="n">Integer</span> <span class="n">combine</span><span class="o">(</span><span class="n">Integer</span> <span class="n">val1</span><span class="o">,</span> <span class="n">Integer</span> <span class="n">val2</span><span class="o">)</span> <span class="o">{</span>
<span class="k">return</span> <span class="mi">1</span><span class="o">;</span>
<span class="o">}</span>
- <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">zero</span><span class="o">()</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="n">Integer</span> <span class="n">zero</span><span class="o">()</span> <span class="o">{</span>
<span class="k">return</span> <span class="mi">1</span><span class="o">;</span>
<span class="o">}</span>
<span class="o">}</span>
@@ -234,11 +234,11 @@
<p>The Trident data model is the TridentTuple which is a named list of values. During a topology, tuples are incrementally built up through a sequence of operations. Operations generally take in a set of input fields and emit a set of "function fields". The input fields are used to select a subset of the tuple as input to the operation, while the "function fields" name the fields the operation emits.</p>
<p>Consider this example. Suppose you have a stream called "stream" that contains the fields "x", "y", and "z". To run a filter MyFilter that takes in "y" as input, you would say:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"y"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">MyFilter</span><span class="o">())</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"y"</span><span class="o">),</span> <span class="k">new</span> <span class="n">MyFilter</span><span class="o">())</span>
</code></pre></div>
<p>Suppose the implementation of MyFilter is this:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyFilter</span> <span class="kd">extends</span> <span class="n">BaseFilter</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="kt">boolean</span> <span class="nf">isKeep</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">boolean</span> <span class="n">isKeep</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">)</span> <span class="o">{</span>
<span class="k">return</span> <span class="n">tuple</span><span class="o">.</span><span class="na">getInteger</span><span class="o">(</span><span class="mi">0</span><span class="o">)</span> <span class="o"><</span> <span class="mi">10</span><span class="o">;</span>
<span class="o">}</span>
<span class="o">}</span>
@@ -247,26 +247,26 @@
<p>Let's now look at how "function fields" work. Suppose you had this function:</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">class</span> <span class="nc">AddAndMultiply</span> <span class="kd">extends</span> <span class="n">BaseFunction</span> <span class="o">{</span>
- <span class="kd">public</span> <span class="kt">void</span> <span class="nf">execute</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">,</span> <span class="n">TridentCollector</span> <span class="n">collector</span><span class="o">)</span> <span class="o">{</span>
+ <span class="kd">public</span> <span class="kt">void</span> <span class="n">execute</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">,</span> <span class="n">TridentCollector</span> <span class="n">collector</span><span class="o">)</span> <span class="o">{</span>
<span class="kt">int</span> <span class="n">i1</span> <span class="o">=</span> <span class="n">tuple</span><span class="o">.</span><span class="na">getInteger</span><span class="o">(</span><span class="mi">0</span><span class="o">);</span>
<span class="kt">int</span> <span class="n">i2</span> <span class="o">=</span> <span class="n">tuple</span><span class="o">.</span><span class="na">getInteger</span><span class="o">(</span><span class="mi">1</span><span class="o">);</span>
- <span class="n">collector</span><span class="o">.</span><span class="na">emit</span><span class="o">(</span><span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="n">i1</span> <span class="o">+</span> <span class="n">i2</span><span class="o">,</span> <span class="n">i1</span> <span class="o">*</span> <span class="n">i2</span><span class="o">));</span>
+ <span class="n">collector</span><span class="o">.</span><span class="na">emit</span><span class="o">(</span><span class="k">new</span> <span class="n">Values</span><span class="o">(</span><span class="n">i1</span> <span class="o">+</span> <span class="n">i2</span><span class="o">,</span> <span class="n">i1</span> <span class="o">*</span> <span class="n">i2</span><span class="o">));</span>
<span class="o">}</span>
<span class="o">}</span>
</code></pre></div>
<p>This function takes two numbers as input and emits two new values: the addition of the numbers and the multiplication of the numbers. Suppose you had a stream with the fields "x", "y", and "z". You would use this function like this:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"x"</span><span class="o">,</span> <span class="s">"y"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">AddAndMultiply</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"added"</span><span class="o">,</span> <span class="s">"multiplied"</span><span class="o">));</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"x"</span><span class="o">,</span> <span class="s">"y"</span><span class="o">),</span> <span class="k">new</span> <span class="n">AddAndMultiply</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"added"</span><span class="o">,</span> <span class="s">"multiplied"</span><span class="o">));</span>
</code></pre></div>
<p>The output of functions is additive: the fields are added to the input tuple. So the output of this each call would contain tuples with the five fields "x", "y", "z", "added", and "multiplied". "added" corresponds to the first value emitted by AddAndMultiply, while "multiplied" corresponds to the second value.</p>
<p>With aggregators, on the other hand, the function fields replace the input tuples. So if you had a stream containing the fields "val1" and "val2", and you did this:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"val2"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">Sum</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"sum"</span><span class="o">))</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"val2"</span><span class="o">),</span> <span class="k">new</span> <span class="n">Sum</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"sum"</span><span class="o">))</span>
</code></pre></div>
<p>The output stream would only contain a single tuple with a single field called "sum", representing the sum of all "val2" fields in that batch.</p>
<p>With grouped streams, the output will contain the grouping fields followed by the fields emitted by the aggregator. For example:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"val1"</span><span class="o">))</span>
- <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"val2"</span><span class="o">),</span> <span class="k">new</span> <span class="nf">Sum</span><span class="o">(),</span> <span class="k">new</span> <span class="nf">Fields</span><span class="o">(</span><span class="s">"sum"</span><span class="o">))</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">stream</span><span class="o">.</span><span class="na">groupBy</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"val1"</span><span class="o">))</span>
+ <span class="o">.</span><span class="na">aggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"val2"</span><span class="o">),</span> <span class="k">new</span> <span class="n">Sum</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"sum"</span><span class="o">))</span>
</code></pre></div>
<p>In this example, the output will contain the fields "val1" and "sum".</p>
http://git-wip-us.apache.org/repos/asf/storm/blob/dfa66d7d/_site/documentation/Troubleshooting.html
----------------------------------------------------------------------
diff --git a/_site/documentation/Troubleshooting.html b/_site/documentation/Troubleshooting.html
index 896f59f..e1d392c 100644
--- a/_site/documentation/Troubleshooting.html
+++ b/_site/documentation/Troubleshooting.html
@@ -157,7 +157,7 @@
<li>Try deleting the local dirs for the supervisors and restarting the daemons. Supervisors create a unique id for themselves and store it locally. When that id is copied to other nodes, Storm gets confused. </li>
</ul>
-<h3 id=""multiple-defaults.yaml-found"-error">"Multiple defaults.yaml found" error</h3>
+<h3 id="quot-multiple-defaults-yaml-found-quot-error">"Multiple defaults.yaml found" error</h3>
<p>Symptoms:</p>
@@ -171,7 +171,7 @@
<li>You're most likely including the Storm jars inside your topology jar. When packaging your topology jar, don't include the Storm jars as Storm will put those on the classpath for you.</li>
</ul>
-<h3 id=""nosuchmethoderror"-when-running-storm-jar">"NoSuchMethodError" when running storm jar</h3>
+<h3 id="quot-nosuchmethoderror-quot-when-running-storm-jar">"NoSuchMethodError" when running storm jar</h3>
<p>Symptoms:</p>
@@ -192,7 +192,7 @@
<ul>
<li>At runtime, you get a stack trace like the following:</li>
</ul>
-<div class="highlight"><pre><code class="language-text" data-lang="text">java.lang.RuntimeException: java.util.ConcurrentModificationException
+<div class="highlight"><pre><code class="language-" data-lang="">java.lang.RuntimeException: java.util.ConcurrentModificationException
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:84)
at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:55)
at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:56)
@@ -233,7 +233,7 @@ Caused by: java.util.ConcurrentModificationException
<ul>
<li>You get a NullPointerException that looks something like:</li>
</ul>
-<div class="highlight"><pre><code class="language-text" data-lang="text">java.lang.RuntimeException: java.lang.NullPointerException
+<div class="highlight"><pre><code class="language-" data-lang="">java.lang.RuntimeException: java.lang.NullPointerException
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:84)
at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:55)
at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:56)
@@ -252,7 +252,7 @@ Caused by: java.lang.NullPointerException
... 6 more
</code></pre></div>
<p>or </p>
-<div class="highlight"><pre><code class="language-text" data-lang="text">java.lang.RuntimeException: java.lang.NullPointerException
+<div class="highlight"><pre><code class="language-" data-lang="">java.lang.RuntimeException: java.lang.NullPointerException
at
backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:128)
~[storm-core-0.9.3.jar:0.9.3]
http://git-wip-us.apache.org/repos/asf/storm/blob/dfa66d7d/_site/documentation/Understanding-the-parallelism-of-a-Storm-topology.html
----------------------------------------------------------------------
diff --git a/_site/documentation/Understanding-the-parallelism-of-a-Storm-topology.html b/_site/documentation/Understanding-the-parallelism-of-a-Storm-topology.html
index 773af59..f2d2b16 100644
--- a/_site/documentation/Understanding-the-parallelism-of-a-Storm-topology.html
+++ b/_site/documentation/Understanding-the-parallelism-of-a-Storm-topology.html
@@ -90,7 +90,7 @@
<p class="post-meta"></p>
-<h2 id="what-makes-a-running-topology:-worker-processes,-executors-and-tasks">What makes a running topology: worker processes, executors and tasks</h2>
+<h2 id="what-makes-a-running-topology-worker-processes-executors-and-tasks">What makes a running topology: worker processes, executors and tasks</h2>
<p>Storm distinguishes between the following three main entities that are used to actually run a topology in a Storm cluster:</p>
@@ -128,7 +128,7 @@
</ul></li>
</ul>
-<h3 id="number-of-executors-(threads)">Number of executors (threads)</h3>
+<h3 id="number-of-executors-threads">Number of executors (threads)</h3>
<ul>
<li>Description: How many executors to spawn <em>per component</em>.</li>
@@ -155,9 +155,9 @@
</ul>
<p>Here is an example code snippet to show these settings in practice:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"green-bolt"</span><span class="o">,</span> <span class="k">new</span> <span class="nf">GreenBolt</span><span class="o">(),</span> <span class="mi">2</span><span class="o">)</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"green-bolt"</span><span class="o">,</span> <span class="k">new</span> <span class="n">GreenBolt</span><span class="o">(),</span> <span class="mi">2</span><span class="o">)</span>
<span class="o">.</span><span class="na">setNumTasks</span><span class="o">(</span><span class="mi">4</span><span class="o">)</span>
- <span class="o">.</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"blue-spout"</span><span class="o">);</span>
+ <span class="o">.</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"blue-spout"</span><span class="o">);</span>
</code></pre></div>
<p>In the above code we configured Storm to run the bolt <code>GreenBolt</code> with an initial number of two executors and four associated tasks. Storm will run two tasks per executor (thread). If you do not explicitly configure the number of tasks, Storm will run by default one task per executor.</p>
@@ -168,20 +168,20 @@
<p><img src="images/example-of-a-running-topology.png" alt="Example of a running topology in Storm"></p>
<p>The <code>GreenBolt</code> was configured as per the code snippet above whereas <code>BlueSpout</code> and <code>YellowBolt</code> only set the parallelism hint (number of executors). Here is the relevant code:</p>
-<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">Config</span> <span class="n">conf</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">Config</span><span class="o">();</span>
+<div class="highlight"><pre><code class="language-java" data-lang="java"><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="n">conf</span><span class="o">.</span><span class="na">setNumWorkers</span><span class="o">(</span><span class="mi">2</span><span class="o">);</span> <span class="c1">// use two worker processes</span>
-<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setSpout</span><span class="o">(</span><span class="s">"blue-spout"</span><span class="o">,</span> <span class="k">new</span> <span class="nf">BlueSpout</span><span class="o">(),</span> <span class="mi">2</span><span class="o">);</span> <span class="c1">// set parallelism hint to 2</span>
+<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setSpout</span><span class="o">(</span><span class="s">"blue-spout"</span><span class="o">,</span> <span class="k">new</span> <span class="n">BlueSpout</span><span class="o">(),</span> <span class="mi">2</span><span class="o">);</span> <span class="c1">// set parallelism hint to 2</span>
-<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"green-bolt"</span><span class="o">,</span> <span class="k">new</span> <span class="nf">GreenBolt</span><span class="o">(),</span> <span class="mi">2</span><span class="o">)</span>
+<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"green-bolt"</span><span class="o">,</span> <span class="k">new</span> <span class="n">GreenBolt</span><span class="o">(),</span> <span class="mi">2</span><span class="o">)</span>
<span class="o">.</span><span class="na">setNumTasks</span><span class="o">(</span><span class="mi">4</span><span class="o">)</span>
- <span class="o">.</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"blue-spout"</span><span class="o">);</span>
+ <span class="o">.</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"blue-spout"</span><span class="o">);</span>
-<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"yellow-bolt"</span><span class="o">,</span> <span class="k">new</span> <span class="nf">YellowBolt</span><span class="o">(),</span> <span class="mi">6</span><span class="o">)</span>
- <span class="o">.</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"green-bolt"</span><span class="o">);</span>
+<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">setBolt</span><span class="o">(</span><span class="s">"yellow-bolt"</span><span class="o">,</span> <span class="k">new</span> <span class="n">YellowBolt</span><span class="o">(),</span> <span class="mi">6</span><span class="o">)</span>
+ <span class="o">.</span><span class="na">shuffleGrouping</span><span class="o">(</span><span class="s">"green-bolt"</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">"mytopology"</span><span class="o">,</span>
+ <span class="s">"mytopology"</span><span class="o">,</span>
<span class="n">conf</span><span class="o">,</span>
<span class="n">topologyBuilder</span><span class="o">.</span><span class="na">createTopology</span><span class="o">()</span>
<span class="o">);</span>
@@ -204,9 +204,9 @@
</ol>
<p>Here is an example of using the CLI tool:</p>
-<div class="highlight"><pre><code class="language-text" data-lang="text">## Reconfigure the topology "mytopology" to use 5 worker processes,
-## the spout "blue-spout" to use 3 executors and
-## the bolt "yellow-bolt" to use 10 executors.
+<div class="highlight"><pre><code class="language-" data-lang="">## Reconfigure the topology "mytopology" to use 5 worker processes,
+## the spout "blue-spout" to use 3 executors and
+## the bolt "yellow-bolt" to use 10 executors.
$ storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10
</code></pre></div>
http://git-wip-us.apache.org/repos/asf/storm/blob/dfa66d7d/_site/documentation/Using-non-JVM-languages-with-Storm.html
----------------------------------------------------------------------
diff --git a/_site/documentation/Using-non-JVM-languages-with-Storm.html b/_site/documentation/Using-non-JVM-languages-with-Storm.html
index 1dab5ea..2483509 100644
--- a/_site/documentation/Using-non-JVM-languages-with-Storm.html
+++ b/_site/documentation/Using-non-JVM-languages-with-Storm.html
@@ -127,7 +127,7 @@
<p>The right place to start is src/storm.thrift. Since Storm topologies are just Thrift structures, and Nimbus is a Thrift daemon, you can create and submit topologies in any language.</p>
<p>When you create the Thrift structs for spouts and bolts, the code for the spout or bolt is specified in the ComponentObject struct:</p>
-<div class="highlight"><pre><code class="language-text" data-lang="text">union ComponentObject {
+<div class="highlight"><pre><code class="language-" data-lang="">union ComponentObject {
1: binary serialized_java;
2: ShellComponent shell;
3: JavaObject java_object;
@@ -136,13 +136,13 @@
<p>For a non-JVM DSL, you would want to make use of "2" and "3". ShellComponent lets you specify a script to run that component (e.g., your python code). And JavaObject lets you specify native java spouts and bolts for the component (and Storm will use reflection to create that spout or bolt).</p>
<p>There's a "storm shell" command that will help with submitting a topology. Its usage is like this:</p>
-<div class="highlight"><pre><code class="language-text" data-lang="text">storm shell resources/ python topology.py arg1 arg2
+<div class="highlight"><pre><code class="language-" data-lang="">storm shell resources/ python topology.py arg1 arg2
</code></pre></div>
<p>storm shell will then package resources/ into a jar, upload the jar to Nimbus, and call your topology.py script like this:</p>
-<div class="highlight"><pre><code class="language-text" data-lang="text">python topology.py arg1 arg2 {nimbus-host} {nimbus-port} {uploaded-jar-location}
+<div class="highlight"><pre><code class="language-" data-lang="">python topology.py arg1 arg2 {nimbus-host} {nimbus-port} {uploaded-jar-location}
</code></pre></div>
<p>Then you can connect to Nimbus using the Thrift API and submit the topology, passing {uploaded-jar-location} into the submitTopology method. For reference, here's the submitTopology definition:</p>
-<div class="highlight"><pre><code class="language-text" data-lang="text">void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology)
+<div class="highlight"><pre><code class="language-" data-lang="">void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology)
throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite);
</code></pre></div>