| <!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>Clojure DSL</title> |
| |
| <!-- Bootstrap core CSS --> |
| <link href="/assets/css/bootstrap.min.css" rel="stylesheet"> |
| <!-- Bootstrap theme --> |
| <link href="/assets/css/bootstrap-theme.min.css" rel="stylesheet"> |
| |
| <!-- Custom styles for this template --> |
| <link rel="stylesheet" href="http://fortawesome.github.io/Font-Awesome/assets/font-awesome/css/font-awesome.css"> |
| <link href="/css/style.css" rel="stylesheet"> |
| <link href="/assets/css/owl.theme.css" rel="stylesheet"> |
| <link href="/assets/css/owl.carousel.css" rel="stylesheet"> |
| <script type="text/javascript" src="/assets/js/jquery.min.js"></script> |
| <script type="text/javascript" src="/assets/js/bootstrap.min.js"></script> |
| <script type="text/javascript" src="/assets/js/owl.carousel.min.js"></script> |
| <script type="text/javascript" src="/assets/js/storm.js"></script> |
| <!-- Just for debugging purposes. Don't actually copy these 2 lines! --> |
| <!--[if lt IE 9]><script src="../../assets/js/ie8-responsive-file-warning.js"></script><![endif]--> |
| |
| <!-- HTML5 shim and Respond.js for IE8 support of HTML5 elements and media queries --> |
| <!--[if lt IE 9]> |
| <script src="https://oss.maxcdn.com/html5shiv/3.7.2/html5shiv.min.js"></script> |
| <script src="https://oss.maxcdn.com/respond/1.4.2/respond.min.js"></script> |
| <![endif]--> |
| </head> |
| |
| |
| <body> |
| <header> |
| <div class="container-fluid"> |
| <div class="row"> |
| <div class="col-md-5"> |
| <a href="/index.html"><img src="/images/logo.png" class="logo" /></a> |
| </div> |
| <div class="col-md-5"> |
| |
| <h1>Version: 2.3.0</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.3.0/index.html">2.3.0</a></li> |
| |
| |
| |
| <li><a href="/releases/2.2.1/index.html">2.2.1</a></li> |
| |
| |
| |
| <li><a href="/releases/2.2.0/index.html">2.2.0</a></li> |
| |
| |
| |
| <li><a href="/releases/2.1.0/index.html">2.1.0</a></li> |
| |
| |
| |
| <li><a href="/releases/2.0.0/index.html">2.0.0</a></li> |
| |
| |
| |
| <li><a href="/releases/1.2.3/index.html">1.2.3</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> |
| <li><a href="/Powered-By.html">PoweredBy</a></li> |
| </ul> |
| </li> |
| <li><a href="/2021/10/11/storm221-released.html" id="news">News</a></li> |
| </ul> |
| </nav> |
| </div> |
| </div> |
| |
| |
| |
| <div class="container-fluid"> |
| <h1 class="page-title">Clojure DSL</h1> |
| <div class="row"> |
| <div class="col-md-12"> |
| <!-- Documentation --> |
| |
| <p class="post-meta"></p> |
| |
| <div class="documentation-content"><p>Storm offers a Clojure DSL through the storm-clojure package for defining spouts, bolts, and topologies. The Clojure DSL has access to everything the Java API exposes, so if you're a Clojure user you can code Storm topologies without touching Java at all. The Clojure DSL is defined in the source in the <a href="http://github.com/apache/storm/blob/v2.3.0/storm-clojure/src/clj/org/apache/storm/clojure.clj">org.apache.storm.clojure</a> namespace.</p> |
| |
| <p>This page outlines all the pieces of the Clojure DSL, including:</p> |
| |
| <ol> |
| <li>Defining topologies</li> |
| <li><code>defbolt</code></li> |
| <li><code>defspout</code></li> |
| <li>Running topologies in local mode or on a cluster</li> |
| <li>Testing topologies</li> |
| </ol> |
| |
| <h3 id="defining-topologies">Defining topologies</h3> |
| |
| <p>To define a topology, use the <code>topology</code> function. <code>topology</code> takes in two arguments: a map of "spout specs" and a map of "bolt specs". Each spout and bolt spec wires the code for the component into the topology by specifying things like inputs and parallelism.</p> |
| |
| <p>Let's take a look at an example topology definition <a href="http://github.com/apache/storm/blob/v2.3.0/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj">from the storm-starter project</a>:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">topology</span><span class="w"> |
| </span><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="p">(</span><span class="nf">spout-spec</span><span class="w"> </span><span class="n">sentence-spout</span><span class="p">)</span><span class="w"> |
| </span><span class="s">"2"</span><span class="w"> </span><span class="p">(</span><span class="nf">spout-spec</span><span class="w"> </span><span class="p">(</span><span class="nf">sentence-spout-parameterized</span><span class="w"> |
| </span><span class="p">[</span><span class="s">"the cat jumped over the door"</span><span class="w"> |
| </span><span class="s">"greetings from a faraway land"</span><span class="p">])</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">2</span><span class="p">)}</span><span class="w"> |
| </span><span class="p">{</span><span class="s">"3"</span><span class="w"> </span><span class="p">(</span><span class="nf">bolt-spec</span><span class="w"> </span><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="no">:shuffle</span><span class="w"> </span><span class="s">"2"</span><span class="w"> </span><span class="no">:shuffle</span><span class="p">}</span><span class="w"> |
| </span><span class="n">split-sentence</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">5</span><span class="p">)</span><span class="w"> |
| </span><span class="s">"4"</span><span class="w"> </span><span class="p">(</span><span class="nf">bolt-spec</span><span class="w"> </span><span class="p">{</span><span class="s">"3"</span><span class="w"> </span><span class="p">[</span><span class="s">"word"</span><span class="p">]}</span><span class="w"> |
| </span><span class="n">word-count</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">6</span><span class="p">)})</span><span class="w"> |
| </span></code></pre></div> |
| <p>The maps of spout and bolt specs are maps from the component id to the corresponding spec. The component ids must be unique across the maps. Just like defining topologies in Java, component ids are used when declaring inputs for bolts in the topology.</p> |
| |
| <h4 id="spout-spec">spout-spec</h4> |
| |
| <p><code>spout-spec</code> takes as arguments the spout implementation (an object that implements <a href="javadocs/org/apache/storm/topology/IRichSpout.html">IRichSpout</a>) and optional keyword arguments. The only option that exists currently is the <code>:p</code> option, which specifies the parallelism for the spout. If you omit <code>:p</code>, the spout will execute as a single task.</p> |
| |
| <h4 id="bolt-spec">bolt-spec</h4> |
| |
| <p><code>bolt-spec</code> takes as arguments the input declaration for the bolt, the bolt implementation (an object that implements <a href="javadocs/org/apache/storm/topology/IRichBolt.html">IRichBolt</a>), and optional keyword arguments.</p> |
| |
| <p>The input declaration is a map from stream ids to stream groupings. A stream id can have one of two forms:</p> |
| |
| <ol> |
| <li><code>[==component id== ==stream id==]</code>: Subscribes to a specific stream on a component</li> |
| <li><code>==component id==</code>: Subscribes to the default stream on a component</li> |
| </ol> |
| |
| <p>A stream grouping can be one of the following:</p> |
| |
| <ol> |
| <li><code>:shuffle</code>: subscribes with a shuffle grouping</li> |
| <li>Vector of field names, like <code>["id" "name"]</code>: subscribes with a fields grouping on the specified fields</li> |
| <li><code>:global</code>: subscribes with a global grouping</li> |
| <li><code>:all</code>: subscribes with an all grouping</li> |
| <li><code>:direct</code>: subscribes with a direct grouping</li> |
| </ol> |
| |
| <p>See <a href="Concepts.html">Concepts</a> for more info on stream groupings. Here's an example input declaration showcasing the various ways to declare inputs:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">{[</span><span class="s">"2"</span><span class="w"> </span><span class="s">"1"</span><span class="p">]</span><span class="w"> </span><span class="no">:shuffle</span><span class="w"> |
| </span><span class="s">"3"</span><span class="w"> </span><span class="p">[</span><span class="s">"field1"</span><span class="w"> </span><span class="s">"field2"</span><span class="p">]</span><span class="w"> |
| </span><span class="p">[</span><span class="s">"4"</span><span class="w"> </span><span class="s">"2"</span><span class="p">]</span><span class="w"> </span><span class="no">:global</span><span class="p">}</span><span class="w"> |
| </span></code></pre></div> |
| <p>This input declaration subscribes to three streams total. It subscribes to stream "1" on component "2" with a shuffle grouping, subscribes to the default stream on component "3" with a fields grouping on the fields "field1" and "field2", and subscribes to stream "2" on component "4" with a global grouping.</p> |
| |
| <p>Like <code>spout-spec</code>, the only current supported keyword argument for <code>bolt-spec</code> is <code>:p</code> which specifies the parallelism for the bolt.</p> |
| |
| <h4 id="shell-bolt-spec">shell-bolt-spec</h4> |
| |
| <p><code>shell-bolt-spec</code> is used for defining bolts that are implemented in a non-JVM language. It takes as arguments the input declaration, the command line program to run, the name of the file implementing the bolt, an output specification, and then the same keyword arguments that <code>bolt-spec</code> accepts.</p> |
| |
| <p>Here's an example <code>shell-bolt-spec</code>:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">shell-bolt-spec</span><span class="w"> </span><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="no">:shuffle</span><span class="w"> </span><span class="s">"2"</span><span class="w"> </span><span class="p">[</span><span class="s">"id"</span><span class="p">]}</span><span class="w"> |
| </span><span class="s">"python"</span><span class="w"> |
| </span><span class="s">"mybolt.py"</span><span class="w"> |
| </span><span class="p">[</span><span class="s">"outfield1"</span><span class="w"> </span><span class="s">"outfield2"</span><span class="p">]</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">25</span><span class="p">)</span><span class="w"> |
| </span></code></pre></div> |
| <p>The syntax of output declarations is described in more detail in the <code>defbolt</code> section below. See <a href="Using-non-JVM-languages-with-Storm.html">Using non JVM languages with Storm</a> for more details on how multilang works within Storm.</p> |
| |
| <h3 id="defbolt">defbolt</h3> |
| |
| <p><code>defbolt</code> is used for defining bolts in Clojure. Bolts have the constraint that they must be serializable, and this is why you can't just reify <code>IRichBolt</code> to implement a bolt (closures aren't serializable). <code>defbolt</code> works around this restriction and provides a nicer syntax for defining bolts than just implementing a Java interface.</p> |
| |
| <p>At its fullest expressiveness, <code>defbolt</code> supports parameterized bolts and maintaining state in a closure around the bolt implementation. It also provides shortcuts for defining bolts that don't need this extra functionality. The signature for <code>defbolt</code> looks like the following:</p> |
| |
| <p>(defbolt <em>name</em> <em>output-declaration</em> *<em>option-map</em> & <em>impl</em>)</p> |
| |
| <p>Omitting the option map is equivalent to having an option map of <code>{:prepare false}</code>.</p> |
| |
| <h4 id="simple-bolts">Simple bolts</h4> |
| |
| <p>Let's start with the simplest form of <code>defbolt</code>. Here's an example bolt that splits a tuple containing a sentence into a tuple for each word:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">defbolt</span><span class="w"> </span><span class="n">split-sentence</span><span class="w"> </span><span class="p">[</span><span class="s">"word"</span><span class="p">]</span><span class="w"> </span><span class="p">[</span><span class="n">tuple</span><span class="w"> </span><span class="n">collector</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="k">let</span><span class="w"> </span><span class="p">[</span><span class="n">words</span><span class="w"> </span><span class="p">(</span><span class="nf">.split</span><span class="w"> </span><span class="p">(</span><span class="nf">.getString</span><span class="w"> </span><span class="n">tuple</span><span class="w"> </span><span class="mi">0</span><span class="p">)</span><span class="w"> </span><span class="s">" "</span><span class="p">)]</span><span class="w"> |
| </span><span class="p">(</span><span class="nb">doseq</span><span class="w"> </span><span class="p">[</span><span class="n">w</span><span class="w"> </span><span class="n">words</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">emit-bolt!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="p">[</span><span class="n">w</span><span class="p">]</span><span class="w"> </span><span class="no">:anchor</span><span class="w"> </span><span class="n">tuple</span><span class="p">))</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">ack!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="n">tuple</span><span class="p">)</span><span class="w"> |
| </span><span class="p">))</span><span class="w"> |
| </span></code></pre></div> |
| <p>Since the option map is omitted, this is a non-prepared bolt. The DSL simply expects an implementation for the <code>execute</code> method of <code>IRichBolt</code>. The implementation takes two parameters, the tuple and the <code>OutputCollector</code>, and is followed by the body of the <code>execute</code> function. The DSL automatically type-hints the parameters for you so you don't need to worry about reflection if you use Java interop.</p> |
| |
| <p>This implementation binds <code>split-sentence</code> to an actual <code>IRichBolt</code> object that you can use in topologies, like so:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">bolt-spec</span><span class="w"> </span><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="no">:shuffle</span><span class="p">}</span><span class="w"> |
| </span><span class="n">split-sentence</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">5</span><span class="p">)</span><span class="w"> |
| </span></code></pre></div> |
| <h4 id="parameterized-bolts">Parameterized bolts</h4> |
| |
| <p>Many times you want to parameterize your bolts with other arguments. For example, let's say you wanted to have a bolt that appends a suffix to every input string it receives, and you want that suffix to be set at runtime. You do this with <code>defbolt</code> by including a <code>:params</code> option in the option map, like so:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">defbolt</span><span class="w"> </span><span class="n">suffix-appender</span><span class="w"> </span><span class="p">[</span><span class="s">"word"</span><span class="p">]</span><span class="w"> </span><span class="p">{</span><span class="no">:params</span><span class="w"> </span><span class="p">[</span><span class="n">suffix</span><span class="p">]}</span><span class="w"> |
| </span><span class="p">[</span><span class="n">tuple</span><span class="w"> </span><span class="n">collector</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">emit-bolt!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="p">[(</span><span class="nb">str</span><span class="w"> </span><span class="p">(</span><span class="nf">.getString</span><span class="w"> </span><span class="n">tuple</span><span class="w"> </span><span class="mi">0</span><span class="p">)</span><span class="w"> </span><span class="n">suffix</span><span class="p">)]</span><span class="w"> </span><span class="no">:anchor</span><span class="w"> </span><span class="n">tuple</span><span class="p">)</span><span class="w"> |
| </span><span class="p">)</span><span class="w"> |
| </span></code></pre></div> |
| <p>Unlike the previous example, <code>suffix-appender</code> will be bound to a function that returns an <code>IRichBolt</code> rather than be an <code>IRichBolt</code> object directly. This is caused by specifying <code>:params</code> in its option map. So to use <code>suffix-appender</code> in a topology, you would do something like:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">bolt-spec</span><span class="w"> </span><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="no">:shuffle</span><span class="p">}</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">suffix-appender</span><span class="w"> </span><span class="s">"-suffix"</span><span class="p">)</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">10</span><span class="p">)</span><span class="w"> |
| </span></code></pre></div> |
| <h4 id="prepared-bolts">Prepared bolts</h4> |
| |
| <p>To do more complex bolts, such as ones that do joins and streaming aggregations, the bolt needs to store state. You can do this by creating a prepared bolt which is specified by including <code>{:prepare true}</code> in the option map. Consider, for example, this bolt that implements word counting:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">defbolt</span><span class="w"> </span><span class="n">word-count</span><span class="w"> </span><span class="p">[</span><span class="s">"word"</span><span class="w"> </span><span class="s">"count"</span><span class="p">]</span><span class="w"> </span><span class="p">{</span><span class="no">:prepare</span><span class="w"> </span><span class="n">true</span><span class="p">}</span><span class="w"> |
| </span><span class="p">[</span><span class="n">conf</span><span class="w"> </span><span class="n">context</span><span class="w"> </span><span class="n">collector</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="k">let</span><span class="w"> </span><span class="p">[</span><span class="n">counts</span><span class="w"> </span><span class="p">(</span><span class="nf">atom</span><span class="w"> </span><span class="p">{})]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">bolt</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">execute</span><span class="w"> </span><span class="p">[</span><span class="n">tuple</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="k">let</span><span class="w"> </span><span class="p">[</span><span class="n">word</span><span class="w"> </span><span class="p">(</span><span class="nf">.getString</span><span class="w"> </span><span class="n">tuple</span><span class="w"> </span><span class="mi">0</span><span class="p">)]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">swap!</span><span class="w"> </span><span class="n">counts</span><span class="w"> </span><span class="p">(</span><span class="nb">partial</span><span class="w"> </span><span class="nb">merge-with</span><span class="w"> </span><span class="nb">+</span><span class="p">)</span><span class="w"> </span><span class="p">{</span><span class="n">word</span><span class="w"> </span><span class="mi">1</span><span class="p">})</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">emit-bolt!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="p">[</span><span class="n">word</span><span class="w"> </span><span class="p">(</span><span class="o">@</span><span class="n">counts</span><span class="w"> </span><span class="n">word</span><span class="p">)]</span><span class="w"> </span><span class="no">:anchor</span><span class="w"> </span><span class="n">tuple</span><span class="p">)</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">ack!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="n">tuple</span><span class="p">)</span><span class="w"> |
| </span><span class="p">)))))</span><span class="w"> |
| </span></code></pre></div> |
| <p>The implementation for a prepared bolt is a function that takes as input the topology config, <code>TopologyContext</code>, and <code>OutputCollector</code>, and returns an implementation of the <code>IBolt</code> interface. This design allows you to have a closure around the implementation of <code>execute</code> and <code>cleanup</code>. </p> |
| |
| <p>In this example, the word counts are stored in the closure in a map called <code>counts</code>. The <code>bolt</code> macro is used to create the <code>IBolt</code> implementation. The <code>bolt</code> macro is a more concise way to implement the interface than reifying, and it automatically type-hints all of the method parameters. This bolt implements the execute method which updates the count in the map and emits the new word count.</p> |
| |
| <p>Note that the <code>execute</code> method in prepared bolts only takes as input the tuple since the <code>OutputCollector</code> is already in the closure of the function (for simple bolts the collector is a second parameter to the <code>execute</code> function).</p> |
| |
| <p>Prepared bolts can be parameterized just like simple bolts.</p> |
| |
| <h4 id="output-declarations">Output declarations</h4> |
| |
| <p>The Clojure DSL has a concise syntax for declaring the outputs of a bolt. The most general way to declare the outputs is as a map from stream id a stream spec. For example:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">{</span><span class="s">"1"</span><span class="w"> </span><span class="p">[</span><span class="s">"field1"</span><span class="w"> </span><span class="s">"field2"</span><span class="p">]</span><span class="w"> |
| </span><span class="s">"2"</span><span class="w"> </span><span class="p">(</span><span class="nf">direct-stream</span><span class="w"> </span><span class="p">[</span><span class="s">"f1"</span><span class="w"> </span><span class="s">"f2"</span><span class="w"> </span><span class="s">"f3"</span><span class="p">])</span><span class="w"> |
| </span><span class="s">"3"</span><span class="w"> </span><span class="p">[</span><span class="s">"f1"</span><span class="p">]}</span><span class="w"> |
| </span></code></pre></div> |
| <p>The stream id is a string, while the stream spec is either a vector of fields or a vector of fields wrapped by <code>direct-stream</code>. <code>direct stream</code> marks the stream as a direct stream (See <a href="Concepts.html">Concepts</a> and <a href="">Direct groupings</a> for more details on direct streams).</p> |
| |
| <p>If the bolt only has one output stream, you can define the default stream of the bolt by using a vector instead of a map for the output declaration. For example:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">[</span><span class="s">"word"</span><span class="w"> </span><span class="s">"count"</span><span class="p">]</span><span class="w"> |
| </span></code></pre></div> |
| <p>This declares the output of the bolt as the fields ["word" "count"] on the default stream id.</p> |
| |
| <h4 id="emitting-acking-and-failing">Emitting, acking, and failing</h4> |
| |
| <p>Rather than use the Java methods on <code>OutputCollector</code> directly, the DSL provides a nicer set of functions for using <code>OutputCollector</code>: <code>emit-bolt!</code>, <code>emit-direct-bolt!</code>, <code>ack!</code>, and <code>fail!</code>.</p> |
| |
| <ol> |
| <li><code>emit-bolt!</code>: takes as parameters the <code>OutputCollector</code>, the values to emit (a Clojure sequence), and keyword arguments for <code>:anchor</code> and <code>:stream</code>. <code>:anchor</code> can be a single tuple or a list of tuples, and <code>:stream</code> is the id of the stream to emit to. Omitting the keyword arguments emits an unanchored tuple to the default stream.</li> |
| <li><code>emit-direct-bolt!</code>: takes as parameters the <code>OutputCollector</code>, the task id to send the tuple to, the values to emit, and keyword arguments for <code>:anchor</code> and <code>:stream</code>. This function can only emit to streams declared as direct streams.</li> |
| <li><code>ack!</code>: takes as parameters the <code>OutputCollector</code> and the tuple to ack.</li> |
| <li><code>fail!</code>: takes as parameters the <code>OutputCollector</code> and the tuple to fail.</li> |
| </ol> |
| |
| <p>See <a href="Guaranteeing-message-processing.html">Guaranteeing message processing</a> for more info on acking and anchoring.</p> |
| |
| <h3 id="defspout">defspout</h3> |
| |
| <p><code>defspout</code> is used for defining spouts in Clojure. Like bolts, spouts must be serializable so you can't just reify <code>IRichSpout</code> to do spout implementations in Clojure. <code>defspout</code> works around this restriction and provides a nicer syntax for defining spouts than just implementing a Java interface.</p> |
| |
| <p>The signature for <code>defspout</code> looks like the following:</p> |
| |
| <p>(defspout <em>name</em> <em>output-declaration</em> *<em>option-map</em> & <em>impl</em>)</p> |
| |
| <p>If you leave out the option map, it defaults to {:prepare true}. The output declaration for <code>defspout</code> has the same syntax as <code>defbolt</code>.</p> |
| |
| <p>Here's an example <code>defspout</code> implementation from <a href="http://github.com/apache/storm/blob/v2.3.0/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj">storm-starter</a>:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">defspout</span><span class="w"> </span><span class="n">sentence-spout</span><span class="w"> </span><span class="p">[</span><span class="s">"sentence"</span><span class="p">]</span><span class="w"> |
| </span><span class="p">[</span><span class="n">conf</span><span class="w"> </span><span class="n">context</span><span class="w"> </span><span class="n">collector</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="k">let</span><span class="w"> </span><span class="p">[</span><span class="n">sentences</span><span class="w"> </span><span class="p">[</span><span class="s">"a little brown dog"</span><span class="w"> |
| </span><span class="s">"the man petted the dog"</span><span class="w"> |
| </span><span class="s">"four score and seven years ago"</span><span class="w"> |
| </span><span class="s">"an apple a day keeps the doctor away"</span><span class="p">]]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">spout</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">nextTuple</span><span class="w"> </span><span class="p">[]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">Thread/sleep</span><span class="w"> </span><span class="mi">100</span><span class="p">)</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">emit-spout!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="p">[(</span><span class="nf">rand-nth</span><span class="w"> </span><span class="n">sentences</span><span class="p">)])</span><span class="w"> |
| </span><span class="p">)</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">ack</span><span class="w"> </span><span class="p">[</span><span class="n">id</span><span class="p">]</span><span class="w"> |
| </span><span class="c1">;; You only need to define this method for reliable spouts</span><span class="w"> |
| </span><span class="c1">;; (such as one that reads off of a queue like Kestrel)</span><span class="w"> |
| </span><span class="c1">;; This is an unreliable spout, so it does nothing here</span><span class="w"> |
| </span><span class="p">))))</span><span class="w"> |
| </span></code></pre></div> |
| <p>The implementation takes in as input the topology config, <code>TopologyContext</code>, and <code>SpoutOutputCollector</code>. The implementation returns an <code>ISpout</code> object. Here, the <code>nextTuple</code> function emits a random sentence from <code>sentences</code>. </p> |
| |
| <p>This spout isn't reliable, so the <code>ack</code> and <code>fail</code> methods will never be called. A reliable spout will add a message id when emitting tuples, and then <code>ack</code> or <code>fail</code> will be called when the tuple is completed or failed respectively. See <a href="Guaranteeing-message-processing.html">Guaranteeing message processing</a> for more info on how reliability works within Storm.</p> |
| |
| <p><code>emit-spout!</code> takes in as parameters the <code>SpoutOutputCollector</code> and the new tuple to be emitted, and accepts as keyword arguments <code>:stream</code> and <code>:id</code>. <code>:stream</code> specifies the stream to emit to, and <code>:id</code> specifies a message id for the tuple (used in the <code>ack</code> and <code>fail</code> callbacks). Omitting these arguments emits an unanchored tuple to the default output stream.</p> |
| |
| <p>There is also a <code>emit-direct-spout!</code> function that emits a tuple to a direct stream and takes an additional argument as the second parameter of the task id to send the tuple to.</p> |
| |
| <p>Spouts can be parameterized just like bolts, in which case the symbol is bound to a function returning <code>IRichSpout</code> instead of the <code>IRichSpout</code> itself. You can also declare an unprepared spout which only defines the <code>nextTuple</code> method. Here is an example of an unprepared spout that emits random sentences parameterized at runtime:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">defspout</span><span class="w"> </span><span class="n">sentence-spout-parameterized</span><span class="w"> </span><span class="p">[</span><span class="s">"word"</span><span class="p">]</span><span class="w"> </span><span class="p">{</span><span class="no">:params</span><span class="w"> </span><span class="p">[</span><span class="n">sentences</span><span class="p">]</span><span class="w"> </span><span class="no">:prepare</span><span class="w"> </span><span class="n">false</span><span class="p">}</span><span class="w"> |
| </span><span class="p">[</span><span class="n">collector</span><span class="p">]</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">Thread/sleep</span><span class="w"> </span><span class="mi">500</span><span class="p">)</span><span class="w"> |
| </span><span class="p">(</span><span class="nf">emit-spout!</span><span class="w"> </span><span class="n">collector</span><span class="w"> </span><span class="p">[(</span><span class="nf">rand-nth</span><span class="w"> </span><span class="n">sentences</span><span class="p">)]))</span><span class="w"> |
| </span></code></pre></div> |
| <p>The following example illustrates how to use this spout in a <code>spout-spec</code>:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">(</span><span class="nf">spout-spec</span><span class="w"> </span><span class="p">(</span><span class="nf">sentence-spout-parameterized</span><span class="w"> |
| </span><span class="p">[</span><span class="s">"the cat jumped over the door"</span><span class="w"> |
| </span><span class="s">"greetings from a faraway land"</span><span class="p">])</span><span class="w"> |
| </span><span class="no">:p</span><span class="w"> </span><span class="mi">2</span><span class="p">)</span><span class="w"> |
| </span></code></pre></div> |
| <h3 id="running-topologies-in-local-mode-or-on-a-cluster">Running topologies in local mode or on a cluster</h3> |
| |
| <p>That's all there is to the Clojure DSL. To submit topologies in remote mode or local mode, just use the <code>StormSubmitter</code> class just like you would from Java.</p> |
| |
| <p>To create topology configs, it's easiest to use the <code>org.apache.storm.config</code> namespace which defines constants for all of the possible configs. The constants are the same as the static constants in the <code>Config</code> class, except with dashes instead of underscores. For example, here's a topology config that sets the number of workers to 15 and configures the topology in debug mode:</p> |
| <div class="highlight"><pre><code class="language-clojure" data-lang="clojure"><span class="p">{</span><span class="n">TOPOLOGY-DEBUG</span><span class="w"> </span><span class="n">true</span><span class="w"> |
| </span><span class="n">TOPOLOGY-WORKERS</span><span class="w"> </span><span class="mi">15</span><span class="p">}</span><span class="w"> |
| </span></code></pre></div> |
| <h3 id="testing-topologies">Testing topologies</h3> |
| |
| <p><a href="http://www.pixelmachine.org/2011/12/17/Testing-Storm-Topologies.html">This blog post</a> and its <a href="http://www.pixelmachine.org/2011/12/21/Testing-Storm-Topologies-Part-2.html">follow-up</a> give a good overview of Storm's powerful built-in facilities for testing topologies in Clojure.</p> |
| </div> |
| |
| |
| </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 Apache Storm</h5> |
| <p>Apache Storm integrates with any queueing system and any database system. Apache Storm's spout abstraction makes it easy to integrate a new queuing system. Likewise, integrating Apache 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 Apache 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 © 2021 <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> |
| |
