http://git-wip-us.apache.org/repos/asf/storm-site/blob/ff14ea94/content/releases/0.10.1/Trident-API-Overview.html ---------------------------------------------------------------------- diff --git a/content/releases/0.10.1/Trident-API-Overview.html b/content/releases/0.10.1/Trident-API-Overview.html deleted file mode 100644 index adbb244..0000000 --- a/content/releases/0.10.1/Trident-API-Overview.html +++ /dev/null @@ -1,678 +0,0 @@ -<!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>Trident API Overview</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: 0.10.1</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.2.0/index.html">1.2.0</a></li> - - - - <li><a href="/releases/1.1.2/index.html">1.1.2</a></li> - - - - <li><a href="/releases/1.1.1/index.html">1.1.1</a></li> - - - - <li><a href="/releases/1.1.0/index.html">1.1.0</a></li> - - - - - - <li><a href="/releases/1.0.6/index.html">1.0.6</a></li> - - - - <li><a href="/releases/1.0.4/index.html">1.0.4</a></li> - - - - - - <li><a href="/releases/1.0.3/index.html">1.0.3</a></li> - - - - - - - - - - <li><a href="/releases/0.10.2/index.html">0.10.2</a></li> - - - - <li><a href="/releases/0.10.1/index.html">0.10.1</a></li> - - - - - - - - <li><a href="/releases/0.9.7/index.html">0.9.7</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="/2018/02/15/storm120-released.html" id="news">News</a></li> - </ul> - </nav> - </div> -</div> - - - - <div class="container-fluid"> - <h1 class="page-title">Trident API Overview</h1> - <div class="row"> - <div class="col-md-12"> - <!-- Documentation --> - -<p class="post-meta"></p> - -<p>The core data model in Trident is the "Stream", processed as a series of batches. A stream is partitioned among the nodes in the cluster, and operations applied to a stream are applied in parallel across each partition.</p> - -<p>There are five kinds of operations in Trident:</p> - -<ol> -<li>Operations that apply locally to each partition and cause no network transfer</li> -<li>Repartitioning operations that repartition a stream but otherwise don't change the contents (involves network transfer)</li> -<li>Aggregation operations that do network transfer as part of the operation</li> -<li>Operations on grouped streams</li> -<li>Merges and joins</li> -</ol> - -<h2 id="partition-local-operations">Partition-local operations</h2> - -<p>Partition-local operations involve no network transfer and are applied to each batch partition independently.</p> - -<h3 id="functions">Functions</h3> - -<p>A function takes in a set of input fields and emits zero or more tuples as output. The fields of the output tuple are appended to the original input tuple in the stream. If a function emits no tuples, the original input tuple is filtered out. Otherwise, the input tuple is duplicated for each output tuple. Suppose you have 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">MyFunction</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="k">for</span><span class="o">(</span><span class="kt">int</span> <span class="n">i</span><span class="o">=</span><span class="mi">0</span><span class="o">;</span> <span class="n">i</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="n">i</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">i</span><span class="o">));</span> - <span class="o">}</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>Now suppose you have a stream in the variable "mystream" with the fields ["a", "b", "c"] with the following tuples:</p> -<div class="highlight"><pre><code class="language-" data-lang="">[1, 2, 3] -[4, 1, 6] -[3, 0, 8] -</code></pre></div> -<p>If you run this code:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</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">"b"</span><span class="o">),</span> <span class="k">new</span> <span class="n">MyFunction</span><span class="o">(),</span> <span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"d"</span><span class="o">)))</span> -</code></pre></div> -<p>The resulting tuples would have fields ["a", "b", "c", "d"] and look like this:</p> -<div class="highlight"><pre><code class="language-" data-lang="">[1, 2, 3, 0] -[1, 2, 3, 1] -[4, 1, 6, 0] -</code></pre></div> -<h3 id="filters">Filters</h3> - -<p>Filters take in a tuple as input and decide whether or not to keep that tuple or not. Suppose you had this filter:</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="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">1</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="o">==</span> <span class="mi">2</span><span class="o">;</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>Now suppose you had these tuples with fields ["a", "b", "c"]:</p> -<div class="highlight"><pre><code class="language-" data-lang="">[1, 2, 3] -[2, 1, 1] -[2, 3, 4] -</code></pre></div> -<p>If you ran this code:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">filter</span><span class="o">(</span><span class="k">new</span> <span class="n">MyFilter</span><span class="o">())</span> -</code></pre></div> -<p>The resulting tuples would be:</p> -<div class="highlight"><pre><code class="language-" data-lang="">[1, 2, 3] -</code></pre></div> -<h3 id="map-and-flatmap">map and flatMap</h3> - -<p><code>map</code> returns a stream consisting of the result of applying the given mapping function to the tuples of the stream. This -can be used to apply a one-one transformation to the tuples.</p> - -<p>For example, if there is a stream of words and you wanted to convert it to a stream of upper case words, -you could define a mapping function as follows,</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">UpperCase</span> <span class="kd">extends</span> <span class="n">MapFunction</span> <span class="o">{</span> - <span class="nd">@Override</span> - <span class="kd">public</span> <span class="n">Values</span> <span class="nf">execute</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">input</span><span class="o">)</span> <span class="o">{</span> - <span class="k">return</span> <span class="k">new</span> <span class="nf">Values</span><span class="o">(</span><span class="n">input</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="na">toUpperCase</span><span class="o">());</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>The mapping function can then be applied on the stream to produce a stream of uppercase words.</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="n">UpperCase</span><span class="o">())</span> -</code></pre></div> -<p><code>flatMap</code> is similar to <code>map</code> but has the effect of applying a one-to-many transformation to the values of the stream, -and then flattening the resulting elements into a new stream.</p> - -<p>For example, if there is a stream of sentences and you wanted to convert it to a stream of words, -you could define a flatMap function as follows,</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">FlatMapFunction</span> <span class="o">{</span> - <span class="nd">@Override</span> - <span class="kd">public</span> <span class="n">Iterable</span><span class="o"><</span><span class="n">Values</span><span class="o">></span> <span class="nf">execute</span><span class="o">(</span><span class="n">TridentTuple</span> <span class="n">input</span><span class="o">)</span> <span class="o">{</span> - <span class="n">List</span><span class="o"><</span><span class="n">Values</span><span class="o">></span> <span class="n">valuesList</span> <span class="o">=</span> <span class="k">new</span> <span class="n">ArrayList</span><span class="o"><>();</span> - <span class="k">for</span> <span class="o">(</span><span class="n">String</span> <span class="n">word</span> <span class="o">:</span> <span class="n">input</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="na">split</span><span class="o">(</span><span class="s">" "</span><span class="o">))</span> <span class="o">{</span> - <span class="n">valuesList</span><span class="o">.</span><span class="na">add</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="k">return</span> <span class="n">valuesList</span><span class="o">;</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>The flatMap function can then be applied on the stream of sentences to produce a stream of words,</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">flatMap</span><span class="o">(</span><span class="k">new</span> <span class="n">Split</span><span class="o">())</span> -</code></pre></div> -<p>Of course these operations can be chained, so a stream of uppercase words can be obtained from a stream of sentences as follows,</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">flatMap</span><span class="o">(</span><span class="k">new</span> <span class="n">Split</span><span class="o">()).</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="n">UpperCase</span><span class="o">())</span> -</code></pre></div> -<h3 id="peek">peek</h3> - -<p><code>peek</code> can be used to perform an additional action on each trident tuple as they flow through the stream. - This could be useful for debugging to see the tuples as they flow past a certain point in a pipeline.</p> - -<p>For example, the below code would print the result of converting the words to uppercase before they are passed to <code>groupBy</code> -<code>java - mystream.flatMap(new Split()).map(new UpperCase()) - .peek(new Consumer() { - @Override - public void accept(TridentTuple input) { - System.out.println(input.getString(0)); - } - }) - .groupBy(new Fields("word")) - .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count")) -</code></p> - -<h3 id="min-and-minby">min and minBy</h3> - -<p><code>min</code> and <code>minBy</code> operations return minimum value on each partition of a batch of tuples in a trident stream.</p> - -<p>Suppose, a trident stream contains fields ["device-id", "count"] and the following partitions of tuples</p> -<div class="highlight"><pre><code class="language-" data-lang="">Partition 0: -[123, 2] -[113, 54] -[23, 28] -[237, 37] -[12, 23] -[62, 17] -[98, 42] - -Partition 1: -[64, 18] -[72, 54] -[2, 28] -[742, 71] -[98, 45] -[62, 12] -[19, 174] - - -Partition 2: -[27, 94] -[82, 23] -[9, 86] -[53, 71] -[74, 37] -[51, 49] -[37, 98] - -</code></pre></div> -<p><code>minBy</code> operation can be applied on the above stream of tuples like below which results in emitting tuples with minimum values of <code>count</code> field in each partition.</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">mystream</span><span class="o">.</span><span class="na">minBy</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> -</code></pre></div> -<p>Result of the above code on mentioned partitions is:</p> -<div class="highlight"><pre><code class="language-" data-lang="">Partition 0: -[123, 2] - - -Partition 1: -[62, 12] - - -Partition 2: -[82, 23] - -</code></pre></div> -<p>You can look at other <code>min</code> and <code>minBy</code> operations on Stream -<code>java - public <T> Stream minBy(String inputFieldName, Comparator<T> comparator) - public Stream min(Comparator<TridentTuple> comparator) -</code> -Below example shows how these APIs can be used to find minimum using respective Comparators on a tuple. </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="n">FixedBatchSpout</span><span class="o">(</span><span class="n">allFields</span><span class="o">,</span> <span class="mi">10</span><span class="o">,</span> <span class="n">Vehicle</span><span class="o">.</span><span class="na">generateVehicles</span><span class="o">(</span><span class="mi">20</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">vehiclesStream</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">each</span><span class="o">(</span><span class="n">allFields</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"##### vehicles"</span><span class="o">));</span> - - <span class="n">Stream</span> <span class="n">slowVehiclesStream</span> <span class="o">=</span> - <span class="n">vehiclesStream</span> - <span class="o">.</span><span class="na">min</span><span class="o">(</span><span class="k">new</span> <span class="n">SpeedComparator</span><span class="o">())</span> <span class="c1">// Comparator w.r.t speed on received tuple.</span> - <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="n">vehicleField</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"#### slowest vehicle"</span><span class="o">));</span> - - <span class="n">vehiclesStream</span> - <span class="o">.</span><span class="na">minBy</span><span class="o">(</span><span class="n">Vehicle</span><span class="o">.</span><span class="na">FIELD_NAME</span><span class="o">,</span> <span class="k">new</span> <span class="n">EfficiencyComparator</span><span class="o">())</span> <span class="c1">// Comparator w.r.t efficiency on received tuple.</span> - <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="n">vehicleField</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"#### least efficient vehicle"</span><span class="o">));</span> - -</code></pre></div> -<p>Example applications of these APIs can be located at <a href="https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentMinMaxOfDevicesTopology.java";>TridentMinMaxOfDevicesTopology</a> and <a href="https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentMinMaxOfVehiclesTopology.java";>TridentMinMaxOfVehiclesTopology</a> </p> - -<h3 id="max-and-maxby">max and maxBy</h3> - -<p><code>max</code> and <code>maxBy</code> operations return maximum value on each partition of a batch of tuples in a trident stream.</p> - -<p>Suppose, a trident stream contains fields ["device-id", "count"] as mentioned in the above section.</p> - -<p><code>max</code> and <code>maxBy</code> operations can be applied on the above stream of tuples like below which results in emitting tuples with maximum values of <code>count</code> field for each partition.</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"> <span class="n">mystream</span><span class="o">.</span><span class="na">maxBy</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> -</code></pre></div> -<p>Result of the above code on mentioned partitions is:</p> -<div class="highlight"><pre><code class="language-" data-lang="">Partition 0: -[113, 54] - - -Partition 1: -[19, 174] - - -Partition 2: -[37, 98] - -</code></pre></div> -<p>You can look at other <code>max</code> and <code>maxBy</code> functions on Stream</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"> - <span class="kd">public</span> <span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="n">Stream</span> <span class="nf">maxBy</span><span class="o">(</span><span class="n">String</span> <span class="n">inputFieldName</span><span class="o">,</span> <span class="n">Comparator</span><span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="n">comparator</span><span class="o">)</span> - <span class="kd">public</span> <span class="n">Stream</span> <span class="nf">max</span><span class="o">(</span><span class="n">Comparator</span><span class="o"><</span><span class="n">TridentTuple</span><span class="o">></span> <span class="n">comparator</span><span class="o">)</span> - -</code></pre></div> -<p>Below example shows how these APIs can be used to find maximum using respective Comparators on a tuple.</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="n">FixedBatchSpout</span><span class="o">(</span><span class="n">allFields</span><span class="o">,</span> <span class="mi">10</span><span class="o">,</span> <span class="n">Vehicle</span><span class="o">.</span><span class="na">generateVehicles</span><span class="o">(</span><span class="mi">20</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">vehiclesStream</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">each</span><span class="o">(</span><span class="n">allFields</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"##### vehicles"</span><span class="o">));</span> - - <span class="n">vehiclesStream</span> - <span class="o">.</span><span class="na">max</span><span class="o">(</span><span class="k">new</span> <span class="n">SpeedComparator</span><span class="o">())</span> <span class="c1">// Comparator w.r.t speed on received tuple.</span> - <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="n">vehicleField</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"#### fastest vehicle"</span><span class="o">))</span> - <span class="o">.</span><span class="na">project</span><span class="o">(</span><span class="n">driverField</span><span class="o">)</span> - <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="n">driverField</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"##### fastest driver"</span><span class="o">));</span> - - <span class="n">vehiclesStream</span> - <span class="o">.</span><span class="na">maxBy</span><span class="o">(</span><span class="n">Vehicle</span><span class="o">.</span><span class="na">FIELD_NAME</span><span class="o">,</span> <span class="k">new</span> <span class="n">EfficiencyComparator</span><span class="o">())</span> <span class="c1">// Comparator w.r.t efficiency on received tuple.</span> - <span class="o">.</span><span class="na">each</span><span class="o">(</span><span class="n">vehicleField</span><span class="o">,</span> <span class="k">new</span> <span class="n">Debug</span><span class="o">(</span><span class="s">"#### most efficient vehicle"</span><span class="o">));</span> - -</code></pre></div> -<p>Example applications of these APIs can be located at <a href="https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentMinMaxOfDevicesTopology.java";>TridentMinMaxOfDevicesTopology</a> and <a href="https://github.com/apache/storm/blob/master/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentMinMaxOfVehiclesTopology.java";>TridentMinMaxOfVehiclesTopology</a> </p> - -<h3 id="partitionaggregate">partitionAggregate</h3> - -<p>partitionAggregate runs a function on each partition of a batch of tuples. Unlike functions, the tuples emitted by partitionAggregate replace the input tuples given to it. Consider this example:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">partitionAggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"b"</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>Suppose the input stream contained fields ["a", "b"] and the following partitions of tuples:</p> -<div class="highlight"><pre><code class="language-" data-lang="">Partition 0: -["a", 1] -["b", 2] - -Partition 1: -["a", 3] -["c", 8] - -Partition 2: -["e", 1] -["d", 9] -["d", 10] -</code></pre></div> -<p>Then the output stream of that code would contain these tuples with one field called "sum":</p> -<div class="highlight"><pre><code class="language-" data-lang="">Partition 0: -[3] - -Partition 1: -[11] - -Partition 2: -[20] -</code></pre></div> -<p>There are three different interfaces for defining aggregators: CombinerAggregator, ReducerAggregator, and Aggregator.</p> - -<p>Here's the interface for CombinerAggregator:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">CombinerAggregator</span><span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="kd">extends</span> <span class="n">Serializable</span> <span class="o">{</span> - <span class="n">T</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="n">T</span> <span class="nf">combine</span><span class="o">(</span><span class="n">T</span> <span class="n">val1</span><span class="o">,</span> <span class="n">T</span> <span class="n">val2</span><span class="o">);</span> - <span class="n">T</span> <span class="nf">zero</span><span class="o">();</span> -<span class="o">}</span> -</code></pre></div> -<p>A CombinerAggregator returns a single tuple with a single field as output. CombinerAggregators run the init function on each input tuple and use the combine function to combine values until there's only one value left. If there's no tuples in the partition, the CombinerAggregator emits the output of the zero function. For example, here's the implementation of Count:</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">Count</span> <span class="kd">implements</span> <span class="n">CombinerAggregator</span><span class="o"><</span><span class="n">Long</span><span class="o">></span> <span class="o">{</span> - <span class="kd">public</span> <span class="n">Long</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="k">return</span> <span class="mi">1L</span><span class="o">;</span> - <span class="o">}</span> - - <span class="kd">public</span> <span class="n">Long</span> <span class="nf">combine</span><span class="o">(</span><span class="n">Long</span> <span class="n">val1</span><span class="o">,</span> <span class="n">Long</span> <span class="n">val2</span><span class="o">)</span> <span class="o">{</span> - <span class="k">return</span> <span class="n">val1</span> <span class="o">+</span> <span class="n">val2</span><span class="o">;</span> - <span class="o">}</span> - - <span class="kd">public</span> <span class="n">Long</span> <span class="nf">zero</span><span class="o">()</span> <span class="o">{</span> - <span class="k">return</span> <span class="mi">0L</span><span class="o">;</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>The benefits of CombinerAggregators are seen when you use them with the aggregate method instead of partitionAggregate. In that case, Trident automatically optimizes the computation by doing partial aggregations before transferring tuples over the network.</p> - -<p>A ReducerAggregator has the following interface:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">ReducerAggregator</span><span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="kd">extends</span> <span class="n">Serializable</span> <span class="o">{</span> - <span class="n">T</span> <span class="nf">init</span><span class="o">();</span> - <span class="n">T</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">T</span> <span class="n">curr</span><span class="o">,</span> <span class="n">TridentTuple</span> <span class="n">tuple</span><span class="o">);</span> -<span class="o">}</span> -</code></pre></div> -<p>A ReducerAggregator produces an initial value with init, and then it iterates on that value for each input tuple to produce a single tuple with a single value as output. For example, here's how you would define Count as a ReducerAggregator:</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">Count</span> <span class="kd">implements</span> <span class="n">ReducerAggregator</span><span class="o"><</span><span class="n">Long</span><span class="o">></span> <span class="o">{</span> - <span class="kd">public</span> <span class="n">Long</span> <span class="nf">init</span><span class="o">()</span> <span class="o">{</span> - <span class="k">return</span> <span class="mi">0L</span><span class="o">;</span> - <span class="o">}</span> - - <span class="kd">public</span> <span class="n">Long</span> <span class="nf">reduce</span><span class="o">(</span><span class="n">Long</span> <span class="n">curr</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">curr</span> <span class="o">+</span> <span class="mi">1</span><span class="o">;</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>ReducerAggregator can also be used with persistentAggregate, as you'll see later.</p> - -<p>The most general interface for performing aggregations is Aggregator, which looks like this:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">Aggregator</span><span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="kd">extends</span> <span class="n">Operation</span> <span class="o">{</span> - <span class="n">T</span> <span class="nf">init</span><span class="o">(</span><span class="n">Object</span> <span class="n">batchId</span><span class="o">,</span> <span class="n">TridentCollector</span> <span class="n">collector</span><span class="o">);</span> - <span class="kt">void</span> <span class="nf">aggregate</span><span class="o">(</span><span class="n">T</span> <span class="n">state</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="kt">void</span> <span class="nf">complete</span><span class="o">(</span><span class="n">T</span> <span class="n">state</span><span class="o">,</span> <span class="n">TridentCollector</span> <span class="n">collector</span><span class="o">);</span> -<span class="o">}</span> -</code></pre></div> -<p>Aggregators can emit any number of tuples with any number of fields. They can emit tuples at any point during execution. Aggregators execute in the following way:</p> - -<ol> -<li>The init method is called before processing the batch. The return value of init is an Object that will represent the state of the aggregation and will be passed into the aggregate and complete methods.</li> -<li>The aggregate method is called for each input tuple in the batch partition. This method can update the state and optionally emit tuples.</li> -<li>The complete method is called when all tuples for the batch partition have been processed by aggregate. </li> -</ol> - -<p>Here's how you would implement Count as an 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">CountAgg</span> <span class="kd">extends</span> <span class="n">BaseAggregator</span><span class="o"><</span><span class="n">CountState</span><span class="o">></span> <span class="o">{</span> - <span class="kd">static</span> <span class="kd">class</span> <span class="nc">CountState</span> <span class="o">{</span> - <span class="kt">long</span> <span class="n">count</span> <span class="o">=</span> <span class="mi">0</span><span class="o">;</span> - <span class="o">}</span> - - <span class="kd">public</span> <span class="n">CountState</span> <span class="nf">init</span><span class="o">(</span><span class="n">Object</span> <span class="n">batchId</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="k">return</span> <span class="k">new</span> <span class="nf">CountState</span><span class="o">();</span> - <span class="o">}</span> - - <span class="kd">public</span> <span class="kt">void</span> <span class="nf">aggregate</span><span class="o">(</span><span class="n">CountState</span> <span class="n">state</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">state</span><span class="o">.</span><span class="na">count</span><span class="o">+=</span><span class="mi">1</span><span class="o">;</span> - <span class="o">}</span> - - <span class="kd">public</span> <span class="kt">void</span> <span class="nf">complete</span><span class="o">(</span><span class="n">CountState</span> <span class="n">state</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">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">state</span><span class="o">.</span><span class="na">count</span><span class="o">));</span> - <span class="o">}</span> -<span class="o">}</span> -</code></pre></div> -<p>Sometimes you want to execute multiple aggregators at the same time. This is called chaining and can be accomplished like this:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">chainedAgg</span><span class="o">()</span> - <span class="o">.</span><span class="na">partitionAggregate</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">partitionAggregate</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"b"</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> - <span class="o">.</span><span class="na">chainEnd</span><span class="o">()</span> -</code></pre></div> -<p>This code will run the Count and Sum aggregators on each partition. The output will contain a single tuple with the fields ["count", "sum"].</p> - -<h3 id="statequery-and-partitionpersist">stateQuery and partitionPersist</h3> - -<p>stateQuery and partitionPersist query and update sources of state, respectively. You can read about how to use them on <a href="Trident-state.html">Trident state doc</a>.</p> - -<h3 id="projection">projection</h3> - -<p>The projection method on Stream keeps only the fields specified in the operation. If you had a Stream with fields ["a", "b", "c", "d"] and you ran this code:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</span><span class="o">.</span><span class="na">project</span><span class="o">(</span><span class="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"b"</span><span class="o">,</span> <span class="s">"d"</span><span class="o">))</span> -</code></pre></div> -<p>The output stream would contain only the fields ["b", "d"].</p> - -<h2 id="repartitioning-operations">Repartitioning operations</h2> - -<p>Repartitioning operations run a function to change how the tuples are partitioned across tasks. The number of partitions can also change as a result of repartitioning (for example, if the parallelism hint is greater after repartioning). Repartitioning requires network transfer. Here are the repartitioning functions:</p> - -<ol> -<li>shuffle: Use random round robin algorithm to evenly redistribute tuples across all target partitions</li> -<li>broadcast: Every tuple is replicated to all target partitions. This can useful during DRPC â for example, if you need to do a stateQuery on every partition of data.</li> -<li>partitionBy: partitionBy takes in a set of fields and does semantic partitioning based on that set of fields. The fields are hashed and modded by the number of target partitions to select the target partition. partitionBy guarantees that the same set of fields always goes to the same target partition.</li> -<li>global: All tuples are sent to the same partition. The same partition is chosen for all batches in the stream.</li> -<li>batchGlobal: All tuples in the batch are sent to the same partition. Different batches in the stream may go to different partitions. </li> -<li>partition: This method takes in a custom partitioning function that implements backtype.storm.grouping.CustomStreamGrouping</li> -</ol> - -<h2 id="aggregation-operations">Aggregation operations</h2> - -<p>Trident has aggregate and persistentAggregate methods for doing aggregations on Streams. aggregate is run on each batch of the stream in isolation, while persistentAggregate will aggregation on all tuples across all batches in the stream and store the result in a source of state.</p> - -<p>Running aggregate on a Stream does a global aggregation. When you use a ReducerAggregator or an Aggregator, the stream is first repartitioned into a single partition, and then the aggregation function is run on that partition. When you use a CombinerAggregator, on the other hand, first Trident will compute partial aggregations of each partition, then repartition to a single partition, and then finish the aggregation after the network transfer. CombinerAggregator's are far more efficient and should be used when possible.</p> - -<p>Here's an example of using aggregate to get a global count for a batch:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">mystream</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">"count"</span><span class="o">))</span> -</code></pre></div> -<p>Like partitionAggregate, aggregators for aggregate can be chained. However, if you chain a CombinerAggregator with a non-CombinerAggregator, Trident is unable to do the partial aggregation optimization.</p> - -<p>You can read more about how to use persistentAggregate in the <a href="Trident-state.html">Trident state doc</a>.</p> - -<h2 id="operations-on-grouped-streams">Operations on grouped streams</h2> - -<p>The groupBy operation repartitions the stream by doing a partitionBy on the specified fields, and then within each partition groups tuples together whose group fields are equal. For example, here's an illustration of a groupBy operation:</p> - -<p><img src="images/grouping.png" alt="Grouping"></p> - -<p>If you run aggregators on a grouped stream, the aggregation will be run within each group instead of against the whole batch. persistentAggregate can also be run on a GroupedStream, in which case the results will be stored in a <a href="http://github.com/apache/storm/blob/v0.10.1/storm-core/src/jvm/storm/trident/state/map/MapState.java";>MapState</a> with the key being the grouping fields. You can read more about persistentAggregate in the <a href="Trident-state.html">Trident state doc</a>.</p> - -<p>Like regular streams, aggregators on grouped streams can be chained.</p> - -<h2 id="merges-and-joins">Merges and joins</h2> - -<p>The last part of the API is combining different streams together. The simplest way to combine streams is to merge them into one stream. You can do that with the TridentTopology#merge method, like so:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">topology</span><span class="o">.</span><span class="na">merge</span><span class="o">(</span><span class="n">stream1</span><span class="o">,</span> <span class="n">stream2</span><span class="o">,</span> <span class="n">stream3</span><span class="o">);</span> -</code></pre></div> -<p>Trident will name the output fields of the new, merged stream as the output fields of the first stream.</p> - -<p>Another way to combine streams is with a join. Now, a standard join, like the kind from SQL, require finite input. So they don't make sense with infinite streams. Joins in Trident only apply within each small batch that comes off of the spout. </p> - -<p>Here's an example join between a stream containing fields ["key", "val1", "val2"] and another stream containing ["x", "val1"]:</p> -<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">topology</span><span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">stream1</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="n">stream2</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="k">new</span> <span class="n">Fields</span><span class="o">(</span><span class="s">"key"</span><span class="o">,</span> <span class="s">"a"</span><span class="o">,</span> <span class="s">"b"</span><span class="o">,</span> <span class="s">"c"</span><span class="o">));</span> -</code></pre></div> -<p>This joins stream1 and stream2 together using "key" and "x" as the join fields for each respective stream. Then, Trident requires that all the output fields of the new stream be named, since the input streams could have overlapping field names. The tuples emitted from the join will contain:</p> - -<ol> -<li>First, the list of join fields. In this case, "key" corresponds to "key" from stream1 and "x" from stream2.</li> -<li>Next, a list of all non-join fields from all streams, in order of how the streams were passed to the join method. In this case, "a" and "b" correspond to "val1" and "val2" from stream1, and "c" corresponds to "val1" from stream2.</li> -</ol> - -<p>When a join happens between streams originating from different spouts, those spouts will be synchronized with how they emit batches. That is, a batch of processing will include tuples from each spout.</p> - -<p>You might be wondering â how do you do something like a "windowed join", where tuples from one side of the join are joined against the last hour of tuples from the other side of the join.</p> - -<p>To do this, you would make use of partitionPersist and stateQuery. The last hour of tuples from one side of the join would be stored and rotated in a source of state, keyed by the join field. Then the stateQuery would do lookups by the join field to perform the "join".</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> -
http://git-wip-us.apache.org/repos/asf/storm-site/blob/ff14ea94/content/releases/0.10.1/Trident-spouts.html ---------------------------------------------------------------------- diff --git a/content/releases/0.10.1/Trident-spouts.html b/content/releases/0.10.1/Trident-spouts.html deleted file mode 100644 index 942171e..0000000 --- a/content/releases/0.10.1/Trident-spouts.html +++ /dev/null @@ -1,288 +0,0 @@ -<!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>Trident Spouts</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: 0.10.1</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.2.0/index.html">1.2.0</a></li> - - - - <li><a href="/releases/1.1.2/index.html">1.1.2</a></li> - - - - <li><a href="/releases/1.1.1/index.html">1.1.1</a></li> - - - - <li><a href="/releases/1.1.0/index.html">1.1.0</a></li> - - - - - - <li><a href="/releases/1.0.6/index.html">1.0.6</a></li> - - - - <li><a href="/releases/1.0.4/index.html">1.0.4</a></li> - - - - - - <li><a href="/releases/1.0.3/index.html">1.0.3</a></li> - - - - - - - - - - <li><a href="/releases/0.10.2/index.html">0.10.2</a></li> - - - - <li><a href="/releases/0.10.1/index.html">0.10.1</a></li> - - - - - - - - <li><a href="/releases/0.9.7/index.html">0.9.7</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="/2018/02/15/storm120-released.html" id="news">News</a></li> - </ul> - </nav> - </div> -</div> - - - - <div class="container-fluid"> - <h1 class="page-title">Trident Spouts</h1> - <div class="row"> - <div class="col-md-12"> - <!-- Documentation --> - -<p class="post-meta"></p> - -<h1 id="trident-spouts">Trident spouts</h1> - -<p>Like in the vanilla Storm API, spouts are the source of streams in a Trident topology. On top of the vanilla Storm spouts, Trident exposes additional APIs for more sophisticated spouts.</p> - -<p>There is an inextricable link between how you source your data streams and how you update state (e.g. databases) based on those data streams. See <a href="Trident-state.html">Trident state doc</a> for an explanation of this â understanding this link is imperative for understanding the spout options available.</p> - -<p>Regular Storm spouts will be non-transactional spouts in a Trident topology. To use a regular Storm IRichSpout, create the stream like this in a TridentTopology:</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="n">TridentTopology</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">"myspoutid"</span><span class="o">,</span> <span class="k">new</span> <span class="n">MyRichSpout</span><span class="o">());</span> -</code></pre></div> -<p>All spouts in a Trident topology are required to be given a unique identifier for the stream â this identifier must be unique across all topologies run on the cluster. Trident will use this identifier to store metadata about what the spout has consumed in Zookeeper, including the txid and any metadata associated with the spout.</p> - -<p>You can configure the Zookeeper storage of spout metadata via the following configuration options:</p> - -<ol> -<li><code>transactional.zookeeper.servers</code>: A list of Zookeeper hostnames </li> -<li><code>transactional.zookeeper.port</code>: The port of the Zookeeper cluster</li> -<li><code>transactional.zookeeper.root</code>: The root dir in Zookeeper where metadata is stored. Metadata will be stored at the path <root path>/<spout id></li> -</ol> - -<h2 id="pipelining">Pipelining</h2> - -<p>By default, Trident processes a single batch at a time, waiting for the batch to succeed or fail before trying another batch. You can get significantly higher throughput â and lower latency of processing of each batch â by pipelining the batches. You configure the maximum amount of batches to be processed simultaneously with the "topology.max.spout.pending" property. </p> - -<p>Even while processing multiple batches simultaneously, Trident will order any state updates taking place in the topology among batches. For example, suppose you're doing a global count aggregation into a database. The idea is that while you're updating the count in the database for batch 1, you can still be computing the partial counts for batches 2 through 10. Trident won't move on to the state updates for batch 2 until the state updates for batch 1 have succeeded. This is essential for achieving exactly-once processing semantics, as outline in <a href="Trident-state.html">Trident state doc</a>.</p> - -<h2 id="trident-spout-types">Trident spout types</h2> - -<p>Here are the following spout APIs available:</p> - -<ol> -<li><a href="http://github.com/apache/storm/blob/v0.10.1/storm-core/src/jvm/storm/trident/spout/ITridentSpout.java";>ITridentSpout</a>: The most general API that can support transactional or opaque transactional semantics. Generally you'll use one of the partitioned flavors of this API rather than this one directly.</li> -<li><a href="http://github.com/apache/storm/blob/v0.10.1/storm-core/src/jvm/storm/trident/spout/IBatchSpout.java";>IBatchSpout</a>: A non-transactional spout that emits batches of tuples at a time</li> -<li><a href="http://github.com/apache/storm/blob/v0.10.1/storm-core/src/jvm/storm/trident/spout/IPartitionedTridentSpout.java";>IPartitionedTridentSpout</a>: A transactional spout that reads from a partitioned data source (like a cluster of Kafka servers)</li> -<li><a href="http://github.com/apache/storm/blob/v0.10.1/storm-core/src/jvm/storm/trident/spout/IOpaquePartitionedTridentSpout.java";>IOpaquePartitionedTridentSpout</a>: An opaque transactional spout that reads from a partitioned data source</li> -</ol> - -<p>And, like mentioned in the beginning of this tutorial, you can use regular IRichSpout's as well.</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> -
