You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2016/03/22 16:31:48 UTC
[06/31] storm git commit: STORM-1617: Release Specific Documentation
0.9.x
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/Trident-tutorial.md
----------------------------------------------------------------------
diff --git a/docs/Trident-tutorial.md b/docs/Trident-tutorial.md
new file mode 100644
index 0000000..862dd8b
--- /dev/null
+++ b/docs/Trident-tutorial.md
@@ -0,0 +1,253 @@
+---
+layout: documentation
+---
+# Trident tutorial
+
+Trident is a high-level abstraction for doing realtime computing on top of Storm. It allows you to seamlessly intermix high throughput (millions of messages per second), stateful stream processing with low latency distributed querying. If you're familiar with high level batch processing tools like Pig or Cascading, the concepts of Trident will be very familiar – Trident has joins, aggregations, grouping, functions, and filters. In addition to these, Trident adds primitives for doing stateful, incremental processing on top of any database or persistence store. Trident has consistent, exactly-once semantics, so it is easy to reason about Trident topologies.
+
+## Illustrative example
+
+Let's look at an illustrative example of Trident. This example will do two things:
+
+1. Compute streaming word count from an input stream of sentences
+2. Implement queries to get the sum of the counts for a list of words
+
+For the purposes of illustration, this example will read an infinite stream of sentences from the following source:
+
+```java
+FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
+ new Values("the cow jumped over the moon"),
+ new Values("the man went to the store and bought some candy"),
+ new Values("four score and seven years ago"),
+ new Values("how many apples can you eat"));
+spout.setCycle(true);
+```
+
+This spout cycles through that set of sentences over and over to produce the sentence stream. Here's the code to do the streaming word count part of the computation:
+
+```java
+TridentTopology topology = new TridentTopology();
+TridentState wordCounts =
+ topology.newStream("spout1", spout)
+ .each(new Fields("sentence"), new Split(), new Fields("word"))
+ .groupBy(new Fields("word"))
+ .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
+ .parallelismHint(6);
+```
+
+Let's go through the code line by line. First a TridentTopology object is created, which exposes the interface for constructing Trident computations. TridentTopology has a method called newStream that creates a new stream of data in the topology reading from an input source. In this case, the input source is just the FixedBatchSpout defined from before. Input sources can also be queue brokers like Kestrel or Kafka. Trident keeps track of a small amount of state for each input source (metadata about what it has consumed) in Zookeeper, and the "spout1" string here specifies the node in Zookeeper where Trident should keep that metadata.
+
+Trident processes the stream as small batches of tuples. For example, the incoming stream of sentences might be divided into batches like so:
+
+
+
+Generally the size of those small batches will be on the order of thousands or millions of tuples, depending on your incoming throughput.
+
+Trident provides a fully fledged batch processing API to process those small batches. The API is very similar to what you see in high level abstractions for Hadoop like Pig or Cascading: you can do group by's, joins, aggregations, run functions, run filters, and so on. Of course, processing each small batch in isolation isn't that interesting, so Trident provides functions for doing aggregations across batches and persistently storing those aggregations – whether in memory, in Memcached, in Cassandra, or some other store. Finally, Trident has first-class functions for querying sources of realtime state. That state could be updated by Trident (like in this example), or it could be an independent source of state.
+
+Back to the example, the spout emits a stream containing one field called "sentence". The next line of the topology definition applies the Split function to each tuple in the stream, taking the "sentence" field and splitting it into words. Each sentence tuple creates potentially many word tuples – for instance, the sentence "the cow jumped over the moon" creates six "word" tuples. Here's the definition of Split:
+
+```java
+public class Split extends BaseFunction {
+ public void execute(TridentTuple tuple, TridentCollector collector) {
+ String sentence = tuple.getString(0);
+ for(String word: sentence.split(" ")) {
+ collector.emit(new Values(word));
+ }
+ }
+}
+```
+
+As you can see, it's really simple. It simply grabs the sentence, splits it on whitespace, and emits a tuple for each word.
+
+The rest of the topology computes word count and keeps the results persistently stored. First the stream is grouped by the "word" field. Then, each group is persistently aggregated using the Count aggregator. The persistentAggregate function knows how to store and update the results of the aggregation in a source of state. In this example, the word counts are kept in memory, but this can be trivially swapped to use Memcached, Cassandra, or any other persistent store. Swapping this topology to store counts in Memcached is as simple as replacing the persistentAggregate line with this (using [trident-memcached](https://github.com/nathanmarz/trident-memcached)), where the "serverLocations" is a list of host/ports for the Memcached cluster:
+
+```java
+.persistentAggregate(MemcachedState.transactional(serverLocations), new Count(), new Fields("count"))
+MemcachedState.transactional()
+```
+
+The values stored by persistentAggregate represents the aggregation of all batches ever emitted by the stream.
+
+One of the cool things about Trident is that it has fully fault-tolerant, exactly-once processing semantics. This makes it easy to reason about your realtime processing. Trident persists state in a way so that if failures occur and retries are necessary, it won't perform multiple updates to the database for the same source data.
+
+The persistentAggregate method transforms a Stream into a TridentState object. In this case the TridentState object represents all the word counts. We will use this TridentState object to implement the distributed query portion of the computation.
+
+The next part of the topology implements a low latency distributed query on the word counts. The query takes as input a whitespace separated list of words and return the sum of the counts for those words. These queries are executed just like normal RPC calls, except they are parallelized in the background. Here's an example of how you might invoke one of these queries:
+
+```java
+DRPCClient client = new DRPCClient("drpc.server.location", 3772);
+System.out.println(client.execute("words", "cat dog the man");
+// prints the JSON-encoded result, e.g.: "[[5078]]"
+```
+
+As you can see, it looks just like a regular remote procedure call (RPC), except it's executing in parallel across a Storm cluster. The latency for small queries like this are typically around 10ms. More intense DRPC queries can take longer of course, although the latency largely depends on how many resources you have allocated for the computation.
+
+The implementation of the distributed query portion of the topology looks like this:
+
+```java
+topology.newDRPCStream("words")
+ .each(new Fields("args"), new Split(), new Fields("word"))
+ .groupBy(new Fields("word"))
+ .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
+ .each(new Fields("count"), new FilterNull())
+ .aggregate(new Fields("count"), new Sum(), new Fields("sum"));
+```
+
+The same TridentTopology object is used to create the DRPC stream, and the function is named "words". The function name corresponds to the function name given in the first argument of execute when using a DRPCClient.
+
+Each DRPC request is treated as its own little batch processing job that takes as input a single tuple representing the request. The tuple contains one field called "args" that contains the argument provided by the client. In this case, the argument is a whitespace separated list of words.
+
+First, the Split function is used to split the arguments for the request into its constituent words. The stream is grouped by "word", and the stateQuery operator is used to query the TridentState object that the first part of the topology generated. stateQuery takes in a source of state – in this case, the word counts computed by the other portion of the topology – and a function for querying that state. In this case, the MapGet function is invoked, which gets the count for each word. Since the DRPC stream is grouped the exact same way as the TridentState was (by the "word" field), each word query is routed to the exact partition of the TridentState object that manages updates for that word.
+
+Next, words that didn't have a count are filtered out via the FilterNull filter and the counts are summed using the Sum aggregator to get the result. Then, Trident automatically sends the result back to the waiting client.
+
+Trident is intelligent about how it executes a topology to maximize performance. There's two interesting things happening automatically in this topology:
+
+1. Operations that read from or write to state (like persistentAggregate and stateQuery) automatically batch operations to that state. So if there's 20 updates that need to be made to the database for the current batch of processing, rather than do 20 read requests and 20 writes requests to the database, Trident will automatically batch up the reads and writes, doing only 1 read request and 1 write request (and in many cases, you can use caching in your State implementation to eliminate the read request). So you get the best of both words of convenience – being able to express your computation in terms of what should be done with each tuple – and performance.
+2. Trident aggregators are heavily optimized. Rather than transfer all tuples for a group to the same machine and then run the aggregator, Trident will do partial aggregations when possible before sending tuples over the network. For example, the Count aggregator computes the count on each partition, sends the partial count over the network, and then sums together all the partial counts to get the total count. This technique is similar to the use of combiners in MapReduce.
+
+Let's look at another example of Trident.
+
+## Reach
+
+The next example is a pure DRPC topology that computes the reach of a URL on demand. Reach is the number of unique people exposed to a URL on Twitter. To compute reach, you need to fetch all the people who ever tweeted a URL, fetch all the followers of all those people, unique that set of followers, and that count that uniqued set. Computing reach is too intense for a single machine – it can require thousands of database calls and tens of millions of tuples. With Storm and Trident, you can parallelize the computation of each step across a cluster.
+
+This topology will read from two sources of state. One database maps URLs to a list of people who tweeted that URL. The other database maps a person to a list of followers for that person. The topology definition looks like this:
+
+```java
+TridentState urlToTweeters =
+ topology.newStaticState(getUrlToTweetersState());
+TridentState tweetersToFollowers =
+ topology.newStaticState(getTweeterToFollowersState());
+
+topology.newDRPCStream("reach")
+ .stateQuery(urlToTweeters, new Fields("args"), new MapGet(), new Fields("tweeters"))
+ .each(new Fields("tweeters"), new ExpandList(), new Fields("tweeter"))
+ .shuffle()
+ .stateQuery(tweetersToFollowers, new Fields("tweeter"), new MapGet(), new Fields("followers"))
+ .parallelismHint(200)
+ .each(new Fields("followers"), new ExpandList(), new Fields("follower"))
+ .groupBy(new Fields("follower"))
+ .aggregate(new One(), new Fields("one"))
+ .parallelismHint(20)
+ .aggregate(new Count(), new Fields("reach"));
+```
+
+The topology creates TridentState objects representing each external database using the newStaticState method. These can then be queried in the topology. Like all sources of state, queries to these databases will be automatically batched for maximum efficiency.
+
+The topology definition is straightforward – it's just a simple batch processing job. First, the urlToTweeters database is queried to get the list of people who tweeted the URL for this request. That returns a list, so the ExpandList function is invoked to create a tuple for each tweeter.
+
+Next, the followers for each tweeter must be fetched. It's important that this step be parallelized, so shuffle is invoked to evenly distribute the tweeters among all workers for the topology. Then, the followers database is queried to get the list of followers for each tweeter. You can see that this portion of the topology is given a large parallelism since this is the most intense portion of the computation.
+
+Next, the set of followers is uniqued and counted. This is done in two steps. First a "group by" is done on the batch by "follower", running the "One" aggregator on each group. The "One" aggregator simply emits a single tuple containing the number one for each group. Then, the ones are summed together to get the unique count of the followers set. Here's the definition of the "One" aggregator:
+
+```java
+public class One implements CombinerAggregator<Integer> {
+ public Integer init(TridentTuple tuple) {
+ return 1;
+ }
+
+ public Integer combine(Integer val1, Integer val2) {
+ return 1;
+ }
+
+ public Integer zero() {
+ return 1;
+ }
+}
+```
+
+This is a "combiner aggregator", which knows how to do partial aggregations before transferring tuples over the network to maximize efficiency. Sum is also defined as a combiner aggregator, so the global sum done at the end of the topology will be very efficient.
+
+Let's now look at Trident in more detail.
+
+## Fields and tuples
+
+The Trident data model is the TridentTuple which is a named list of values. During a topology, tuples are incrementally built up through a sequence of operations. Operations generally take in a set of input fields and emit a set of "function fields". The input fields are used to select a subset of the tuple as input to the operation, while the "function fields" name the fields the operation emits.
+
+Consider this example. Suppose you have a stream called "stream" that contains the fields "x", "y", and "z". To run a filter MyFilter that takes in "y" as input, you would say:
+
+```java
+stream.each(new Fields("y"), new MyFilter())
+```
+
+Suppose the implementation of MyFilter is this:
+
+```java
+public class MyFilter extends BaseFilter {
+ public boolean isKeep(TridentTuple tuple) {
+ return tuple.getInteger(0) < 10;
+ }
+}
+```
+
+This will keep all tuples whose "y" field is less than 10. The TridentTuple given as input to MyFilter will only contain the "y" field. Note that Trident is able to project a subset of a tuple extremely efficiently when selecting the input fields: the projection is essentially free.
+
+Let's now look at how "function fields" work. Suppose you had this function:
+
+```java
+public class AddAndMultiply extends BaseFunction {
+ public void execute(TridentTuple tuple, TridentCollector collector) {
+ int i1 = tuple.getInteger(0);
+ int i2 = tuple.getInteger(1);
+ collector.emit(new Values(i1 + i2, i1 * i2));
+ }
+}
+```
+
+This function takes two numbers as input and emits two new values: the addition of the numbers and the multiplication of the numbers. Suppose you had a stream with the fields "x", "y", and "z". You would use this function like this:
+
+```java
+stream.each(new Fields("x", "y"), new AddAndMultiply(), new Fields("added", "multiplied"));
+```
+
+The output of functions is additive: the fields are added to the input tuple. So the output of this each call would contain tuples with the five fields "x", "y", "z", "added", and "multiplied". "added" corresponds to the first value emitted by AddAndMultiply, while "multiplied" corresponds to the second value.
+
+With aggregators, on the other hand, the function fields replace the input tuples. So if you had a stream containing the fields "val1" and "val2", and you did this:
+
+```java
+stream.aggregate(new Fields("val2"), new Sum(), new Fields("sum"))
+```
+
+The output stream would only contain a single tuple with a single field called "sum", representing the sum of all "val2" fields in that batch.
+
+With grouped streams, the output will contain the grouping fields followed by the fields emitted by the aggregator. For example:
+
+```java
+stream.groupBy(new Fields("val1"))
+ .aggregate(new Fields("val2"), new Sum(), new Fields("sum"))
+```
+
+In this example, the output will contain the fields "val1" and "sum".
+
+## State
+
+A key problem to solve with realtime computation is how to manage state so that updates are idempotent in the face of failures and retries. It's impossible to eliminate failures, so when a node dies or something else goes wrong, batches need to be retried. The question is – how do you do state updates (whether external databases or state internal to the topology) so that it's like each message was only processed only once?
+
+This is a tricky problem, and can be illustrated with the following example. Suppose that you're doing a count aggregation of your stream and want to store the running count in a database. If you store only the count in the database and it's time to apply a state update for a batch, there's no way to know if you applied that state update before. The batch could have been attempted before, succeeded in updating the database, and then failed at a later step. Or the batch could have been attempted before and failed to update the database. You just don't know.
+
+Trident solves this problem by doing two things:
+
+1. Each batch is given a unique id called the "transaction id". If a batch is retried it will have the exact same transaction id.
+2. State updates are ordered among batches. That is, the state updates for batch 3 won't be applied until the state updates for batch 2 have succeeded.
+
+With these two primitives, you can achieve exactly-once semantics with your state updates. Rather than store just the count in the database, what you can do instead is store the transaction id with the count in the database as an atomic value. Then, when updating the count, you can just compare the transaction id in the database with the transaction id for the current batch. If they're the same, you skip the update – because of the strong ordering, you know for sure that the value in the database incorporates the current batch. If they're different, you increment the count.
+
+Of course, you don't have to do this logic manually in your topologies. This logic is wrapped by the State abstraction and done automatically. Nor is your State object required to implement the transaction id trick: if you don't want to pay the cost of storing the transaction id in the database, you don't have to. In that case the State will have at-least-once-processing semantics in the case of failures (which may be fine for your application). You can read more about how to implement a State and the various fault-tolerance tradeoffs possible [in this doc](Trident-state.html).
+
+A State is allowed to use whatever strategy it wants to store state. So it could store state in an external database or it could keep the state in-memory but backed by HDFS (like how HBase works). State's are not required to hold onto state forever. For example, you could have an in-memory State implementation that only keeps the last X hours of data available and drops anything older. Take a look at the implementation of the [Memcached integration](https://github.com/nathanmarz/trident-memcached/blob/master/src/jvm/trident/memcached/MemcachedState.java) for an example State implementation.
+
+## Execution of Trident topologies
+
+Trident topologies compile down into as efficient of a Storm topology as possible. Tuples are only sent over the network when a repartitioning of the data is required, such as if you do a groupBy or a shuffle. So if you had this Trident topology:
+
+
+
+It would compile into Storm spouts/bolts like this:
+
+
+
+## Conclusion
+
+Trident makes realtime computation elegant. You've seen how high throughput stream processing, state manipulation, and low-latency querying can be seamlessly intermixed via Trident's API. Trident lets you express your realtime computations in a natural way while still getting maximal performance.
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/Troubleshooting.md
----------------------------------------------------------------------
diff --git a/docs/Troubleshooting.md b/docs/Troubleshooting.md
new file mode 100644
index 0000000..c9df298
--- /dev/null
+++ b/docs/Troubleshooting.md
@@ -0,0 +1,144 @@
+---
+layout: documentation
+---
+## Troubleshooting
+
+This page lists issues people have run into when using Storm along with their solutions.
+
+### Worker processes are crashing on startup with no stack trace
+
+Possible symptoms:
+
+ * Topologies work with one node, but workers crash with multiple nodes
+
+Solutions:
+
+ * You may have a misconfigured subnet, where nodes can't locate other nodes based on their hostname. ZeroMQ sometimes crashes the process when it can't resolve a host. There are two solutions:
+ * Make a mapping from hostname to IP address in /etc/hosts
+ * Set up an internal DNS so that nodes can locate each other based on hostname.
+
+### Nodes are unable to communicate with each other
+
+Possible symptoms:
+
+ * Every spout tuple is failing
+ * Processing is not working
+
+Solutions:
+
+ * Storm doesn't work with ipv6. You can force ipv4 by adding `-Djava.net.preferIPv4Stack=true` to the supervisor child options and restarting the supervisor.
+ * You may have a misconfigured subnet. See the solutions for `Worker processes are crashing on startup with no stack trace`
+
+### Topology stops processing tuples after awhile
+
+Symptoms:
+
+ * Processing works fine for awhile, and then suddenly stops and spout tuples start failing en masse.
+
+Solutions:
+
+ * This is a known issue with ZeroMQ 2.1.10. Downgrade to ZeroMQ 2.1.7.
+
+### Not all supervisors appear in Storm UI
+
+Symptoms:
+
+ * Some supervisor processes are missing from the Storm UI
+ * List of supervisors in Storm UI changes on refreshes
+
+Solutions:
+
+ * Make sure the supervisor local dirs are independent (e.g., not sharing a local dir over NFS)
+ * Try deleting the local dirs for the supervisors and restarting the daemons. Supervisors create a unique id for themselves and store it locally. When that id is copied to other nodes, Storm gets confused.
+
+### "Multiple defaults.yaml found" error
+
+Symptoms:
+
+ * When deploying a topology with "storm jar", you get this error
+
+Solution:
+
+ * You're most likely including the Storm jars inside your topology jar. When packaging your topology jar, don't include the Storm jars as Storm will put those on the classpath for you.
+
+### "NoSuchMethodError" when running storm jar
+
+Symptoms:
+
+ * When running storm jar, you get a cryptic "NoSuchMethodError"
+
+Solution:
+
+ * You're deploying your topology with a different version of Storm than you built your topology against. Make sure the storm client you use comes from the same version as the version you compiled your topology against.
+
+
+### Kryo ConcurrentModificationException
+
+Symptoms:
+
+ * At runtime, you get a stack trace like the following:
+
+```
+java.lang.RuntimeException: java.util.ConcurrentModificationException
+ at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:84)
+ at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:55)
+ at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:56)
+ at backtype.storm.disruptor$consume_loop_STAR_$fn__1597.invoke(disruptor.clj:67)
+ at backtype.storm.util$async_loop$fn__465.invoke(util.clj:377)
+ at clojure.lang.AFn.run(AFn.java:24)
+ at java.lang.Thread.run(Thread.java:679)
+Caused by: java.util.ConcurrentModificationException
+ at java.util.LinkedHashMap$LinkedHashIterator.nextEntry(LinkedHashMap.java:390)
+ at java.util.LinkedHashMap$EntryIterator.next(LinkedHashMap.java:409)
+ at java.util.LinkedHashMap$EntryIterator.next(LinkedHashMap.java:408)
+ at java.util.HashMap.writeObject(HashMap.java:1016)
+ at sun.reflect.GeneratedMethodAccessor17.invoke(Unknown Source)
+ at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
+ at java.lang.reflect.Method.invoke(Method.java:616)
+ at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:959)
+ at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1480)
+ at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1416)
+ at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
+ at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:346)
+ at backtype.storm.serialization.SerializableSerializer.write(SerializableSerializer.java:21)
+ at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:554)
+ at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:77)
+ at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:18)
+ at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:472)
+ at backtype.storm.serialization.KryoValuesSerializer.serializeInto(KryoValuesSerializer.java:27)
+```
+
+Solution:
+
+ * This means that you're emitting a mutable object as an output tuple. Everything you emit into the output collector must be immutable. What's happening is that your bolt is modifying the object while it is being serialized to be sent over the network.
+
+
+### NullPointerException from deep inside Storm
+
+Symptoms:
+
+ * You get a NullPointerException that looks something like:
+
+```
+java.lang.RuntimeException: java.lang.NullPointerException
+ at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:84)
+ at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:55)
+ at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:56)
+ at backtype.storm.disruptor$consume_loop_STAR_$fn__1596.invoke(disruptor.clj:67)
+ at backtype.storm.util$async_loop$fn__465.invoke(util.clj:377)
+ at clojure.lang.AFn.run(AFn.java:24)
+ at java.lang.Thread.run(Thread.java:662)
+Caused by: java.lang.NullPointerException
+ at backtype.storm.serialization.KryoTupleSerializer.serialize(KryoTupleSerializer.java:24)
+ at backtype.storm.daemon.worker$mk_transfer_fn$fn__4126$fn__4130.invoke(worker.clj:99)
+ at backtype.storm.util$fast_list_map.invoke(util.clj:771)
+ at backtype.storm.daemon.worker$mk_transfer_fn$fn__4126.invoke(worker.clj:99)
+ at backtype.storm.daemon.executor$start_batch_transfer__GT_worker_handler_BANG_$fn__3904.invoke(executor.clj:205)
+ at backtype.storm.disruptor$clojure_handler$reify__1584.onEvent(disruptor.clj:43)
+ at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:81)
+ ... 6 more
+```
+
+Solution:
+
+ * This is caused by having multiple threads issue methods on the `OutputCollector`. All emits, acks, and fails must happen on the same thread. One subtle way this can happen is if you make a `IBasicBolt` that emits on a separate thread. `IBasicBolt`'s automatically ack after execute is called, so this would cause multiple threads to use the `OutputCollector` leading to this exception. When using a basic bolt, all emits must happen in the same thread that runs `execute`.
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/Tutorial.md
----------------------------------------------------------------------
diff --git a/docs/Tutorial.md b/docs/Tutorial.md
new file mode 100644
index 0000000..73bf9a4
--- /dev/null
+++ b/docs/Tutorial.md
@@ -0,0 +1,310 @@
+---
+layout: documentation
+---
+In this tutorial, you'll learn how to create Storm topologies and deploy them to a Storm cluster. Java will be the main language used, but a few examples will use Python to illustrate Storm's multi-language capabilities.
+
+## Preliminaries
+
+This tutorial uses examples from the [storm-starter](http://github.com/nathanmarz/storm-starter) project. It's recommended that you clone the project and follow along with the examples. Read [Setting up a development environment](Setting-up-development-environment.html) and [Creating a new Storm project](Creating-a-new-Storm-project.html) to get your machine set up.
+
+## Components of a Storm cluster
+
+A Storm cluster is superficially similar to a Hadoop cluster. Whereas on Hadoop you run "MapReduce jobs", on Storm you run "topologies". "Jobs" and "topologies" themselves are very different -- one key difference is that a MapReduce job eventually finishes, whereas a topology processes messages forever (or until you kill it).
+
+There are two kinds of nodes on a Storm cluster: the master node and the worker nodes. The master node runs a daemon called "Nimbus" that is similar to Hadoop's "JobTracker". Nimbus is responsible for distributing code around the cluster, assigning tasks to machines, and monitoring for failures.
+
+Each worker node runs a daemon called the "Supervisor". The supervisor listens for work assigned to its machine and starts and stops worker processes as necessary based on what Nimbus has assigned to it. Each worker process executes a subset of a topology; a running topology consists of many worker processes spread across many machines.
+
+
+
+All coordination between Nimbus and the Supervisors is done through a [Zookeeper](http://zookeeper.apache.org/) cluster. Additionally, the Nimbus daemon and Supervisor daemons are fail-fast and stateless; all state is kept in Zookeeper or on local disk. This means you can kill -9 Nimbus or the Supervisors and they'll start back up like nothing happened. This design leads to Storm clusters being incredibly stable.
+
+## Topologies
+
+To do realtime computation on Storm, you create what are called "topologies". A topology is a graph of computation. Each node in a topology contains processing logic, and links between nodes indicate how data should be passed around between nodes.
+
+Running a topology is straightforward. First, you package all your code and dependencies into a single jar. Then, you run a command like the following:
+
+```
+storm jar all-my-code.jar backtype.storm.MyTopology arg1 arg2
+```
+
+This runs the class `backtype.storm.MyTopology` with the arguments `arg1` and `arg2`. The main function of the class defines the topology and submits it to Nimbus. The `storm jar` part takes care of connecting to Nimbus and uploading the jar.
+
+Since topology definitions are just Thrift structs, and Nimbus is a Thrift service, you can create and submit topologies using any programming language. The above example is the easiest way to do it from a JVM-based language. See [Running topologies on a production cluster](Running-topologies-on-a-production-cluster.html)] for more information on starting and stopping topologies.
+
+## Streams
+
+The core abstraction in Storm is the "stream". A stream is an unbounded sequence of tuples. Storm provides the primitives for transforming a stream into a new stream in a distributed and reliable way. For example, you may transform a stream of tweets into a stream of trending topics.
+
+The basic primitives Storm provides for doing stream transformations are "spouts" and "bolts". Spouts and bolts have interfaces that you implement to run your application-specific logic.
+
+A spout is a source of streams. For example, a spout may read tuples off of a [Kestrel](http://github.com/nathanmarz/storm-kestrel) queue and emit them as a stream. Or a spout may connect to the Twitter API and emit a stream of tweets.
+
+A bolt consumes any number of input streams, does some processing, and possibly emits new streams. Complex stream transformations, like computing a stream of trending topics from a stream of tweets, require multiple steps and thus multiple bolts. Bolts can do anything from run functions, filter tuples, do streaming aggregations, do streaming joins, talk to databases, and more.
+
+Networks of spouts and bolts are packaged into a "topology" which is the top-level abstraction that you submit to Storm clusters for execution. A topology is a graph of stream transformations where each node is a spout or bolt. Edges in the graph indicate which bolts are subscribing to which streams. When a spout or bolt emits a tuple to a stream, it sends the tuple to every bolt that subscribed to that stream.
+
+
+
+Links between nodes in your topology indicate how tuples should be passed around. For example, if there is a link between Spout A and Bolt B, a link from Spout A to Bolt C, and a link from Bolt B to Bolt C, then everytime Spout A emits a tuple, it will send the tuple to both Bolt B and Bolt C. All of Bolt B's output tuples will go to Bolt C as well.
+
+Each node in a Storm topology executes in parallel. In your topology, you can specify how much parallelism you want for each node, and then Storm will spawn that number of threads across the cluster to do the execution.
+
+A topology runs forever, or until you kill it. Storm will automatically reassign any failed tasks. Additionally, Storm guarantees that there will be no data loss, even if machines go down and messages are dropped.
+
+## Data model
+
+Storm uses tuples as its data model. A tuple is a named list of values, and a field in a tuple can be an object of any type. Out of the box, Storm supports all the primitive types, strings, and byte arrays as tuple field values. To use an object of another type, you just need to implement [a serializer](Serialization.html) for the type.
+
+Every node in a topology must declare the output fields for the tuples it emits. For example, this bolt declares that it emits 2-tuples with the fields "double" and "triple":
+
+```java
+public class DoubleAndTripleBolt extends BaseRichBolt {
+ private OutputCollectorBase _collector;
+
+ @Override
+ public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
+ _collector = collector;
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ int val = input.getInteger(0);
+ _collector.emit(input, new Values(val*2, val*3));
+ _collector.ack(input);
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("double", "triple"));
+ }
+}
+```
+
+The `declareOutputFields` function declares the output fields `["double", "triple"]` for the component. The rest of the bolt will be explained in the upcoming sections.
+
+## A simple topology
+
+Let's take a look at a simple topology to explore the concepts more and see how the code shapes up. Let's look at the `ExclamationTopology` definition from storm-starter:
+
+```java
+TopologyBuilder builder = new TopologyBuilder();
+builder.setSpout("words", new TestWordSpout(), 10);
+builder.setBolt("exclaim1", new ExclamationBolt(), 3)
+ .shuffleGrouping("words");
+builder.setBolt("exclaim2", new ExclamationBolt(), 2)
+ .shuffleGrouping("exclaim1");
+```
+
+This topology contains a spout and two bolts. The spout emits words, and each bolt appends the string "!!!" to its input. The nodes are arranged in a line: the spout emits to the first bolt which then emits to the second bolt. If the spout emits the tuples ["bob"] and ["john"], then the second bolt will emit the words ["bob!!!!!!"] and ["john!!!!!!"].
+
+This code defines the nodes using the `setSpout` and `setBolt` methods. These methods take as input a user-specified id, an object containing the processing logic, and the amount of parallelism you want for the node. In this example, the spout is given id "words" and the bolts are given ids "exclaim1" and "exclaim2".
+
+The object containing the processing logic implements the [IRichSpout](javadocs/backtype/storm/topology/IRichSpout.html) interface for spouts and the [IRichBolt](javadocs/backtype/storm/topology/IRichBolt.html) interface for bolts.
+
+The last parameter, how much parallelism you want for the node, is optional. It indicates how many threads should execute that component across the cluster. If you omit it, Storm will only allocate one thread for that node.
+
+`setBolt` returns an [InputDeclarer](javadocs/backtype/storm/topology/InputDeclarer.html) object that is used to define the inputs to the Bolt. Here, component "exclaim1" declares that it wants to read all the tuples emitted by component "words" using a shuffle grouping, and component "exclaim2" declares that it wants to read all the tuples emitted by component "exclaim1" using a shuffle grouping. "shuffle grouping" means that tuples should be randomly distributed from the input tasks to the bolt's tasks. There are many ways to group data between components. These will be explained in a few sections.
+
+If you wanted component "exclaim2" to read all the tuples emitted by both component "words" and component "exclaim1", you would write component "exclaim2"'s definition like this:
+
+```java
+builder.setBolt("exclaim2", new ExclamationBolt(), 5)
+ .shuffleGrouping("words")
+ .shuffleGrouping("exclaim1");
+```
+
+As you can see, input declarations can be chained to specify multiple sources for the Bolt.
+
+Let's dig into the implementations of the spouts and bolts in this topology. Spouts are responsible for emitting new messages into the topology. `TestWordSpout` in this topology emits a random word from the list ["nathan", "mike", "jackson", "golda", "bertels"] as a 1-tuple every 100ms. The implementation of `nextTuple()` in TestWordSpout looks like this:
+
+```java
+public void nextTuple() {
+ Utils.sleep(100);
+ final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"};
+ final Random rand = new Random();
+ final String word = words[rand.nextInt(words.length)];
+ _collector.emit(new Values(word));
+}
+```
+
+As you can see, the implementation is very straightforward.
+
+`ExclamationBolt` appends the string "!!!" to its input. Let's take a look at the full implementation for `ExclamationBolt`:
+
+```java
+public static class ExclamationBolt implements IRichBolt {
+ OutputCollector _collector;
+
+ public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
+ _collector = collector;
+ }
+
+ public void execute(Tuple tuple) {
+ _collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
+ _collector.ack(tuple);
+ }
+
+ public void cleanup() {
+ }
+
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word"));
+ }
+
+ public Map getComponentConfiguration() {
+ return null;
+ }
+}
+```
+
+The `prepare` method provides the bolt with an `OutputCollector` that is used for emitting tuples from this bolt. Tuples can be emitted at anytime from the bolt -- in the `prepare`, `execute`, or `cleanup` methods, or even asynchronously in another thread. This `prepare` implementation simply saves the `OutputCollector` as an instance variable to be used later on in the `execute` method.
+
+The `execute` method receives a tuple from one of the bolt's inputs. The `ExclamationBolt` grabs the first field from the tuple and emits a new tuple with the string "!!!" appended to it. If you implement a bolt that subscribes to multiple input sources, you can find out which component the [Tuple](javadocs/backtype/storm/tuple/Tuple.html) came from by using the `Tuple#getSourceComponent` method.
+
+There's a few other things going in in the `execute` method, namely that the input tuple is passed as the first argument to `emit` and the input tuple is acked on the final line. These are part of Storm's reliability API for guaranteeing no data loss and will be explained later in this tutorial.
+
+The `cleanup` method is called when a Bolt is being shutdown and should cleanup any resources that were opened. There's no guarantee that this method will be called on the cluster: for example, if the machine the task is running on blows up, there's no way to invoke the method. The `cleanup` method is intended for when you run topologies in [local mode](Local-mode.html) (where a Storm cluster is simulated in process), and you want to be able to run and kill many topologies without suffering any resource leaks.
+
+The `declareOutputFields` method declares that the `ExclamationBolt` emits 1-tuples with one field called "word".
+
+The `getComponentConfiguration` method allows you to configure various aspects of how this component runs. This is a more advanced topic that is explained further on [Configuration](Configuration.html).
+
+Methods like `cleanup` and `getComponentConfiguration` are often not needed in a bolt implementation. You can define bolts more succinctly by using a base class that provides default implementations where appropriate. `ExclamationBolt` can be written more succinctly by extending `BaseRichBolt`, like so:
+
+```java
+public static class ExclamationBolt extends BaseRichBolt {
+ OutputCollector _collector;
+
+ public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
+ _collector = collector;
+ }
+
+ public void execute(Tuple tuple) {
+ _collector.emit(tuple, new Values(tuple.getString(0) + "!!!"));
+ _collector.ack(tuple);
+ }
+
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word"));
+ }
+}
+```
+
+## Running ExclamationTopology in local mode
+
+Let's see how to run the `ExclamationTopology` in local mode and see that it's working.
+
+Storm has two modes of operation: local mode and distributed mode. In local mode, Storm executes completely in process by simulating worker nodes with threads. Local mode is useful for testing and development of topologies. When you run the topologies in storm-starter, they'll run in local mode and you'll be able to see what messages each component is emitting. You can read more about running topologies in local mode on [Local mode](Local-mode.html).
+
+In distributed mode, Storm operates as a cluster of machines. When you submit a topology to the master, you also submit all the code necessary to run the topology. The master will take care of distributing your code and allocating workers to run your topology. If workers go down, the master will reassign them somewhere else. You can read more about running topologies on a cluster on [Running topologies on a production cluster](Running-topologies-on-a-production-cluster.html)].
+
+Here's the code that runs `ExclamationTopology` in local mode:
+
+```java
+Config conf = new Config();
+conf.setDebug(true);
+conf.setNumWorkers(2);
+
+LocalCluster cluster = new LocalCluster();
+cluster.submitTopology("test", conf, builder.createTopology());
+Utils.sleep(10000);
+cluster.killTopology("test");
+cluster.shutdown();
+```
+
+First, the code defines an in-process cluster by creating a `LocalCluster` object. Submitting topologies to this virtual cluster is identical to submitting topologies to distributed clusters. It submits a topology to the `LocalCluster` by calling `submitTopology`, which takes as arguments a name for the running topology, a configuration for the topology, and then the topology itself.
+
+The name is used to identify the topology so that you can kill it later on. A topology will run indefinitely until you kill it.
+
+The configuration is used to tune various aspects of the running topology. The two configurations specified here are very common:
+
+1. **TOPOLOGY_WORKERS** (set with `setNumWorkers`) specifies how many _processes_ you want allocated around the cluster to execute the topology. Each component in the topology will execute as many _threads_. The number of threads allocated to a given component is configured through the `setBolt` and `setSpout` methods. Those _threads_ exist within worker _processes_. Each worker _process_ contains within it some number of _threads_ for some number of components. For instance, you may have 300 threads specified across all your components and 50 worker processes specified in your config. Each worker process will execute 6 threads, each of which of could belong to a different component. You tune the performance of Storm topologies by tweaking the parallelism for each component and the number of worker processes those threads should run within.
+2. **TOPOLOGY_DEBUG** (set with `setDebug`), when set to true, tells Storm to log every message every emitted by a component. This is useful in local mode when testing topologies, but you probably want to keep this turned off when running topologies on the cluster.
+
+There's many other configurations you can set for the topology. The various configurations are detailed on [the Javadoc for Config](javadocs/backtype/storm/Config.html).
+
+To learn about how to set up your development environment so that you can run topologies in local mode (such as in Eclipse), see [Creating a new Storm project](Creating-a-new-Storm-project.html).
+
+## Stream groupings
+
+A stream grouping tells a topology how to send tuples between two components. Remember, spouts and bolts execute in parallel as many tasks across the cluster. If you look at how a topology is executing at the task level, it looks something like this:
+
+
+
+When a task for Bolt A emits a tuple to Bolt B, which task should it send the tuple to?
+
+A "stream grouping" answers this question by telling Storm how to send tuples between sets of tasks. Before we dig into the different kinds of stream groupings, let's take a look at another topology from [storm-starter](http://github.com/nathanmarz/storm-starter). This [WordCountTopology](https://github.com/nathanmarz/storm-starter/blob/master/src/jvm/storm/starter/WordCountTopology.java) reads sentences off of a spout and streams out of `WordCountBolt` the total number of times it has seen that word before:
+
+```java
+TopologyBuilder builder = new TopologyBuilder();
+
+builder.setSpout("sentences", new RandomSentenceSpout(), 5);
+builder.setBolt("split", new SplitSentence(), 8)
+ .shuffleGrouping("sentences");
+builder.setBolt("count", new WordCount(), 12)
+ .fieldsGrouping("split", new Fields("word"));
+```
+
+`SplitSentence` emits a tuple for each word in each sentence it receives, and `WordCount` keeps a map in memory from word to count. Each time `WordCount` receives a word, it updates its state and emits the new word count.
+
+There's a few different kinds of stream groupings.
+
+The simplest kind of grouping is called a "shuffle grouping" which sends the tuple to a random task. A shuffle grouping is used in the `WordCountTopology` to send tuples from `RandomSentenceSpout` to the `SplitSentence` bolt. It has the effect of evenly distributing the work of processing the tuples across all of `SplitSentence` bolt's tasks.
+
+A more interesting kind of grouping is the "fields grouping". A fields grouping is used between the `SplitSentence` bolt and the `WordCount` bolt. It is critical for the functioning of the `WordCount` bolt that the same word always go to the same task. Otherwise, more than one task will see the same word, and they'll each emit incorrect values for the count since each has incomplete information. A fields grouping lets you group a stream by a subset of its fields. This causes equal values for that subset of fields to go to the same task. Since `WordCount` subscribes to `SplitSentence`'s output stream using a fields grouping on the "word" field, the same word always goes to the same task and the bolt produces the correct output.
+
+Fields groupings are the basis of implementing streaming joins and streaming aggregations as well as a plethora of other use cases. Underneath the hood, fields groupings are implemented using mod hashing.
+
+There's a few other kinds of stream groupings. You can read more about them on [Concepts](Concepts.html).
+
+## Defining Bolts in other languages
+
+Bolts can be defined in any language. Bolts written in another language are executed as subprocesses, and Storm communicates with those subprocesses with JSON messages over stdin/stdout. The communication protocol just requires an ~100 line adapter library, and Storm ships with adapter libraries for Ruby, Python, and Fancy.
+
+Here's the definition of the `SplitSentence` bolt from `WordCountTopology`:
+
+```java
+public static class SplitSentence extends ShellBolt implements IRichBolt {
+ public SplitSentence() {
+ super("python", "splitsentence.py");
+ }
+
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word"));
+ }
+}
+```
+
+`SplitSentence` overrides `ShellBolt` and declares it as running using `python` with the arguments `splitsentence.py`. Here's the implementation of `splitsentence.py`:
+
+```python
+import storm
+
+class SplitSentenceBolt(storm.BasicBolt):
+ def process(self, tup):
+ words = tup.values[0].split(" ")
+ for word in words:
+ storm.emit([word])
+
+SplitSentenceBolt().run()
+```
+
+For more information on writing spouts and bolts in other languages, and to learn about how to create topologies in other languages (and avoid the JVM completely), see [Using non-JVM languages with Storm](Using-non-JVM-languages-with-Storm.html).
+
+## Guaranteeing message processing
+
+Earlier on in this tutorial, we skipped over a few aspects of how tuples are emitted. Those aspects were part of Storm's reliability API: how Storm guarantees that every message coming off a spout will be fully processed. See [Guaranteeing message processing](Guaranteeing-message-processing.html) for information on how this works and what you have to do as a user to take advantage of Storm's reliability capabilities.
+
+## Transactional topologies
+
+Storm guarantees that every message will be played through the topology at least once. A common question asked is "how do you do things like counting on top of Storm? Won't you overcount?" Storm has a feature called transactional topologies that let you achieve exactly-once messaging semantics for most computations. Read more about transactional topologies [here](Transactional-topologies.html).
+
+## Distributed RPC
+
+This tutorial showed how to do basic stream processing on top of Storm. There's lots more things you can do with Storm's primitives. One of the most interesting applications of Storm is Distributed RPC, where you parallelize the computation of intense functions on the fly. Read more about Distributed RPC [here](Distributed-RPC.html).
+
+## Conclusion
+
+This tutorial gave a broad overview of developing, testing, and deploying Storm topologies. The rest of the documentation dives deeper into all the aspects of using Storm.
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/Understanding-the-parallelism-of-a-Storm-topology.md
----------------------------------------------------------------------
diff --git a/docs/Understanding-the-parallelism-of-a-Storm-topology.md b/docs/Understanding-the-parallelism-of-a-Storm-topology.md
new file mode 100644
index 0000000..adc4c41
--- /dev/null
+++ b/docs/Understanding-the-parallelism-of-a-Storm-topology.md
@@ -0,0 +1,121 @@
+---
+layout: documentation
+---
+# What makes a running topology: worker processes, executors and tasks
+
+Storm distinguishes between the following three main entities that are used to actually run a topology in a Storm cluster:
+
+1. Worker processes
+2. Executors (threads)
+3. Tasks
+
+Here is a simple illustration of their relationships:
+
+
+
+A _worker process_ executes a subset of a topology. A worker process belongs to a specific topology and may run one or more executors for one or more components (spouts or bolts) of this topology. A running topology consists of many such processes running on many machines within a Storm cluster.
+
+An _executor_ is a thread that is spawned by a worker process. It may run one or more tasks for the same component (spout or bolt).
+
+A _task_ performs the actual data processing — each spout or bolt that you implement in your code executes as many tasks across the cluster. The number of tasks for a component is always the same throughout the lifetime of a topology, but the number of executors (threads) for a component can change over time. This means that the following condition holds true: ``#threads ≤ #tasks``. By default, the number of tasks is set to be the same as the number of executors, i.e. Storm will run one task per thread.
+
+# Configuring the parallelism of a topology
+
+Note that in Storm’s terminology "parallelism" is specifically used to describe the so-called _parallelism hint_, which means the initial number of executor (threads) of a component. In this document though we use the term "parallelism" in a more general sense to describe how you can configure not only the number of executors but also the number of worker processes and the number of tasks of a Storm topology. We will specifically call out when "parallelism" is used in the normal, narrow definition of Storm.
+
+The following sections give an overview of the various configuration options and how to set them in your code. There is more than one way of setting these options though, and the table lists only some of them. Storm currently has the following [order of precedence for configuration settings](Configuration.html): ``defaults.yaml`` < ``storm.yaml`` < topology-specific configuration < internal component-specific configuration < external component-specific configuration.
+
+## Number of worker processes
+
+* Description: How many worker processes to create _for the topology_ across machines in the cluster.
+* Configuration option: [TOPOLOGY_WORKERS](javadocs/backtype/storm/Config.html#TOPOLOGY_WORKERS)
+* How to set in your code (examples):
+ * [Config#setNumWorkers](javadocs/backtype/storm/Config.html)
+
+## Number of executors (threads)
+
+* Description: How many executors to spawn _per component_.
+* Configuration option: ?
+* How to set in your code (examples):
+ * [TopologyBuilder#setSpout()](javadocs/backtype/storm/topology/TopologyBuilder.html)
+ * [TopologyBuilder#setBolt()](javadocs/backtype/storm/topology/TopologyBuilder.html)
+ * Note that as of Storm 0.8 the ``parallelism_hint`` parameter now specifies the initial number of executors (not tasks!) for that bolt.
+
+## Number of tasks
+
+* Description: How many tasks to create _per component_.
+* Configuration option: [TOPOLOGY_TASKS](javadocs/backtype/storm/Config.html#TOPOLOGY_TASKS)
+* How to set in your code (examples):
+ * [ComponentConfigurationDeclarer#setNumTasks()](javadocs/backtype/storm/topology/ComponentConfigurationDeclarer.html)
+
+
+Here is an example code snippet to show these settings in practice:
+
+```java
+topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
+ .setNumTasks(4)
+ .shuffleGrouping("blue-spout);
+```
+
+In the above code we configured Storm to run the bolt ``GreenBolt`` with an initial number of two executors and four associated tasks. Storm will run two tasks per executor (thread). If you do not explicitly configure the number of tasks, Storm will run by default one task per executor.
+
+# Example of a running topology
+
+The following illustration shows how a simple topology would look like in operation. The topology consists of three components: one spout called ``BlueSpout`` and two bolts called ``GreenBolt`` and ``YellowBolt``. The components are linked such that ``BlueSpout`` sends its output to ``GreenBolt``, which in turns sends its own output to ``YellowBolt``.
+
+
+
+The ``GreenBolt`` was configured as per the code snippet above whereas ``BlueSpout`` and ``YellowBolt`` only set the parallelism hint (number of executors). Here is the relevant code:
+
+```java
+Config conf = new Config();
+conf.setNumWorkers(2); // use two worker processes
+
+topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // set parallelism hint to 2
+
+topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
+ .setNumTasks(4)
+ .shuffleGrouping("blue-spout");
+
+topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6)
+ .shuffleGrouping("green-bolt");
+
+StormSubmitter.submitTopology(
+ "mytopology",
+ conf,
+ topologyBuilder.createTopology()
+ );
+```
+
+And of course Storm comes with additional configuration settings to control the parallelism of a topology, including:
+
+* [TOPOLOGY_MAX_TASK_PARALLELISM](javadocs/backtype/storm/Config.html#TOPOLOGY_MAX_TASK_PARALLELISM): This setting puts a ceiling on the number of executors that can be spawned for a single component. It is typically used during testing to limit the number of threads spawned when running a topology in local mode. You can set this option via e.g. [Config#setMaxTaskParallelism()](javadocs/backtype/storm/Config.html).
+
+# How to change the parallelism of a running topology
+
+A nifty feature of Storm is that you can increase or decrease the number of worker processes and/or executors without being required to restart the cluster or the topology. The act of doing so is called rebalancing.
+
+You have two options to rebalance a topology:
+
+1. Use the Storm web UI to rebalance the topology.
+2. Use the CLI tool storm rebalance as described below.
+
+Here is an example of using the CLI tool:
+
+```
+# Reconfigure the topology "mytopology" to use 5 worker processes,
+# the spout "blue-spout" to use 3 executors and
+# the bolt "yellow-bolt" to use 10 executors.
+
+$ storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10
+```
+
+# References for this article
+
+* [Concepts](Concepts.html)
+* [Configuration](Configuration.html)
+* [Running topologies on a production cluster](Running-topologies-on-a-production-cluster.html)]
+* [Local mode](Local-mode.html)
+* [Tutorial](Tutorial.html)
+* [Storm API documentation](javadocs/), most notably the class ``Config``
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/Using-non-JVM-languages-with-Storm.md
----------------------------------------------------------------------
diff --git a/docs/Using-non-JVM-languages-with-Storm.md b/docs/Using-non-JVM-languages-with-Storm.md
new file mode 100644
index 0000000..7b2a2f2
--- /dev/null
+++ b/docs/Using-non-JVM-languages-with-Storm.md
@@ -0,0 +1,52 @@
+---
+layout: documentation
+---
+- two pieces: creating topologies and implementing spouts and bolts in other languages
+- creating topologies in another language is easy since topologies are just thrift structures (link to storm.thrift)
+- implementing spouts and bolts in another language is called a "multilang components" or "shelling"
+ - Here's a specification of the protocol: [Multilang protocol](Multilang-protocol.html)
+ - the thrift structure lets you define multilang components explicitly as a program and a script (e.g., python and the file implementing your bolt)
+ - In Java, you override ShellBolt or ShellSpout to create multilang components
+ - note that output fields declarations happens in the thrift structure, so in Java you create multilang components like the following:
+ - declare fields in java, processing code in the other language by specifying it in constructor of shellbolt
+ - multilang uses json messages over stdin/stdout to communicate with the subprocess
+ - storm comes with ruby, python, and fancy adapters that implement the protocol. show an example of python
+ - python supports emitting, anchoring, acking, and logging
+- "storm shell" command makes constructing jar and uploading to nimbus easy
+ - makes jar and uploads it
+ - calls your program with host/port of nimbus and the jarfile id
+
+## Notes on implementing a DSL in a non-JVM language
+
+The right place to start is src/storm.thrift. Since Storm topologies are just Thrift structures, and Nimbus is a Thrift daemon, you can create and submit topologies in any language.
+
+When you create the Thrift structs for spouts and bolts, the code for the spout or bolt is specified in the ComponentObject struct:
+
+```
+union ComponentObject {
+ 1: binary serialized_java;
+ 2: ShellComponent shell;
+ 3: JavaObject java_object;
+}
+```
+
+For a non-JVM DSL, you would want to make use of "2" and "3". ShellComponent lets you specify a script to run that component (e.g., your python code). And JavaObject lets you specify native java spouts and bolts for the component (and Storm will use reflection to create that spout or bolt).
+
+There's a "storm shell" command that will help with submitting a topology. Its usage is like this:
+
+```
+storm shell resources/ python topology.py arg1 arg2
+```
+
+storm shell will then package resources/ into a jar, upload the jar to Nimbus, and call your topology.py script like this:
+
+```
+python topology.py arg1 arg2 {nimbus-host} {nimbus-port} {uploaded-jar-location}
+```
+
+Then you can connect to Nimbus using the Thrift API and submit the topology, passing {uploaded-jar-location} into the submitTopology method. For reference, here's the submitTopology definition:
+
+```
+void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology)
+ throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite);
+```
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_config.yml
----------------------------------------------------------------------
diff --git a/docs/_config.yml b/docs/_config.yml
new file mode 100644
index 0000000..b05bcef
--- /dev/null
+++ b/docs/_config.yml
@@ -0,0 +1,18 @@
+# Site settings
+title: Apache Storm
+baseurl: "" # the subpath of your site, e.g. /blog/
+url: "http://storm.apache.org" # the base hostname & protocol for your site
+twitter_username: stormprocessor
+github_username: apache/storm
+
+# Build settings
+markdown: redcarpet
+redcarpet:
+ extensions: ["no_intra_emphasis", "fenced_code_blocks", "autolink", "tables", "with_toc_data"]
+
+keep_files: [".git", ".svn"]
+encoding: "utf-8"
+exclude:
+ - READEME.md
+
+storm_release_only: true
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_includes/footer.html
----------------------------------------------------------------------
diff --git a/docs/_includes/footer.html b/docs/_includes/footer.html
new file mode 100644
index 0000000..1696720
--- /dev/null
+++ b/docs/_includes/footer.html
@@ -0,0 +1,55 @@
+<footer>
+ <div class="container-fluid">
+ <div class="row">
+ <div class="col-md-3">
+ <div class="footer-widget">
+ <h5>Meetups</h5>
+ <ul class="latest-news">
+ {% for meetup in site.data.meetups %}
+ <li><a href="{{meetup.url}}">{{meetup.title}}</a> <span class="small">({{meetup.location}})</span></li>
+ {% endfor %}
+ <!-- <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>
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_includes/head.html
----------------------------------------------------------------------
diff --git a/docs/_includes/head.html b/docs/_includes/head.html
new file mode 100644
index 0000000..8f51c94
--- /dev/null
+++ b/docs/_includes/head.html
@@ -0,0 +1,34 @@
+ <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>{% if page.title %}{{ page.title }}{% else %}{{ site.title }}{% endif %}</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>
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_includes/header.html
----------------------------------------------------------------------
diff --git a/docs/_includes/header.html b/docs/_includes/header.html
new file mode 100644
index 0000000..b9a2b03
--- /dev/null
+++ b/docs/_includes/header.html
@@ -0,0 +1,59 @@
+<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">
+ {% if page.version %}
+ <h1>Version: {{page.version}}</h1>
+ {% endif %}
+ </div>
+ <div class="col-md-2">
+ <a href="/releases.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">
+ {% for r in site.data.releases %}
+ {% if r.documented %}
+ <li><a href="/releases/{{r.name}}/index.html">{{r.name}}</a></li>
+ {% endif %}
+ {% endfor %}
+ </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="{{ site.posts[0].url }}" id="news">News</a></li>
+ </ul>
+ </nav>
+ </div>
+</div>
+
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_layouts/about.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/about.html b/docs/_layouts/about.html
new file mode 100644
index 0000000..7ca3e79
--- /dev/null
+++ b/docs/_layouts/about.html
@@ -0,0 +1,43 @@
+---
+layout: default
+title: Project Information
+items:
+ -
+ - "/about/simple-api.html"
+ - "Simple API"
+ -
+ - "/about/scalable.html"
+ - "Scalable"
+ -
+ - "/about/fault-tolerant.html"
+ - "Fault tolerant"
+ -
+ - "/about/guarantees-data-processing.html"
+ - "Guarantees data processing"
+ -
+ - "/about/multi-language.html"
+ - "Use with any language"
+ -
+ - "/about/deployment.html"
+ - "Easy to deploy and operate"
+ -
+ - "/about/free-and-open-source.html"
+ - "Free and open source"
+---
+<div class="download-block">
+ <div class="row">
+ <div class="col-md-3">
+ <ul class="news">
+ {% for post in page.items %}
+ <li>
+ <a href="{{ post[0] }}">{{ post[1] }}</a>
+ </li>
+ {% endfor %}
+ </ul>
+ </div>
+ <div class="col-md-9">
+ {{ content }}
+
+ </div>
+ </div>
+</div>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_layouts/default.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/default.html b/docs/_layouts/default.html
new file mode 100644
index 0000000..80b404e
--- /dev/null
+++ b/docs/_layouts/default.html
@@ -0,0 +1,18 @@
+<!DOCTYPE html>
+<html>
+ {% include head.html %}
+ <body>
+ {% include header.html %}
+ <div class="container-fluid">
+ <h1 class="page-title">{{ page.title }}</h1>
+ <div class="row">
+ <div class="col-md-12">
+ {{ content }}
+ </div>
+ </div>
+ </div>
+{% include footer.html %}
+</body>
+
+</html>
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_layouts/documentation.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/documentation.html b/docs/_layouts/documentation.html
new file mode 100644
index 0000000..81cc09f
--- /dev/null
+++ b/docs/_layouts/documentation.html
@@ -0,0 +1,9 @@
+---
+layout: default
+---
+<!-- Documentation -->
+
+<p class="post-meta">{{ page.date | date: "%b %-d, %Y" }}{% if page.author %} by {{ page.author }}{% endif %}{% if page.meta %} • {{ page.meta }}{% endif %}</p>
+
+{{ content }}
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_layouts/page.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/page.html b/docs/_layouts/page.html
new file mode 100644
index 0000000..e230861
--- /dev/null
+++ b/docs/_layouts/page.html
@@ -0,0 +1,5 @@
+---
+layout: default
+---
+ {{ content }}
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_layouts/post.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/post.html b/docs/_layouts/post.html
new file mode 100644
index 0000000..5080868
--- /dev/null
+++ b/docs/_layouts/post.html
@@ -0,0 +1,61 @@
+<!DOCTYPE html>
+<html>
+
+ {% include head.html %}
+
+ <body>
+
+ {% include header.html %}
+ <div class="container-fluid">
+ <div class="row">
+ <div class="col-md-12">
+ <div class="row">
+ <div class="col-md-3">
+ <ul class="news" id="news-list">
+ {% for post in site.posts %}
+ <li><a href="{{ post.url }}">{{ post.title }}</a></li>
+ {% endfor %}
+ </ul>
+ </div>
+ <div class="col-md-9" id="news-content">
+ <h1 class="page-title">
+ {{ page.title }}
+ </h1>
+
+ <div class="row" style="margin: -15px;">
+ <div class="col-md-12">
+ <p class="text-muted credit pull-left">Posted on {{ page.date | date: "%b %-d, %Y" }}{% if page.author %} by {{ page.author }}{% endif %}{% if page.meta %} • {{ page.meta }}{% endif %}</p>
+ <div class="pull-right">
+ <a
+ href="https://twitter.com/share"
+ class="twitter-share-button"
+ data-count=none
+ >Tweet</a>
+ <script> !function(d,s,id){
+ var js,
+ fjs=d.getElementsByTagName(s)[0],
+ p=/^http:/.test(d.location)?'http':'https';
+ if(!d.getElementById(id)){
+ js=d.createElement(s);
+ js.id=id;
+ js.src=p+'://platform.twitter.com/widgets.js';
+ fjs.parentNode.insertBefore(js,fjs);
+ }
+ }(document, 'script', 'twitter-wjs');
+ </script>
+ </div>
+ </div>
+ </div>
+ <div>
+ {{ content }}
+ </div>
+ </div>
+ </div>
+ </div>
+ </div>
+ </div>
+ {% include footer.html %}
+ </body>
+
+</html>
+
http://git-wip-us.apache.org/repos/asf/storm/blob/d491c3ff/docs/_plugins/releases.rb
----------------------------------------------------------------------
diff --git a/docs/_plugins/releases.rb b/docs/_plugins/releases.rb
new file mode 100644
index 0000000..f28ccd2
--- /dev/null
+++ b/docs/_plugins/releases.rb
@@ -0,0 +1,84 @@
+module Releases
+ class Generator < Jekyll::Generator
+ def dir_to_releasename(dir)
+ ret = nil
+ splitdir = dir.split("/").select{ |a| a != ""};
+ if (splitdir[0] == 'releases')
+ ret = splitdir[1]
+ if (ret == 'current')
+ ret = File.readlink(splitdir.join("/")).split("/")[-1]
+ end
+ end
+ return ret
+ end
+
+ def set_if_unset(hash, key, value)
+ hash[key] = hash[key] || value;
+ end
+
+ def parse_version(version_string)
+ return version_string.split('.').map{|e| e.to_i}
+ end
+
+ def release_from_pom()
+ text= `mvn -f ../pom.xml help:evaluate -Dexpression=project.version`
+ return text.split("\n").select{|a| !a.start_with?('[')}[0]
+ end
+
+ def branch_from_git()
+ return `git rev-parse --abbrev-ref HEAD`
+ end
+
+ def generate(site)
+ if site.config['storm_release_only']
+ release_name = release_from_pom()
+ puts "release: #{release_name}"
+ git_branch = branch_from_git()
+ puts "branch: #{git_branch}"
+ for page in site.pages do
+ page.data['version'] = release_name;
+ page.data['git-tree-base'] = "http://github.com/apache/storm/tree/#{git_branch}"
+ page.data['git-blob-base'] = "http://github.com/apache/storm/blob/#{git_branch}"
+ end
+ return
+ end
+
+ releases = Hash.new
+ if (site.data['releases'])
+ for rel_data in site.data['releases'] do
+ releases[rel_data['name']] = rel_data
+ end
+ end
+
+ for page in site.pages do
+ release_name = dir_to_releasename(page.dir)
+ if (release_name != nil)
+ if !releases.has_key?(release_name)
+ releases[release_name] = {'name' => release_name};
+ end
+ releases[release_name]['documented'] = true
+ end
+ end
+
+ releases.each { |release_name, release_data|
+ set_if_unset(release_data, 'git-tag-or-branch', "v#{release_data['name']}")
+ set_if_unset(release_data, 'git-tree-base', "http://github.com/apache/storm/tree/#{release_data['git-tag-or-branch']}")
+ set_if_unset(release_data, 'git-blob-base', "http://github.com/apache/storm/blob/#{release_data['git-tag-or-branch']}")
+ set_if_unset(release_data, 'base-name', "apache-storm-#{release_data['name']}")
+ set_if_unset(release_data, 'has-download', !release_name.end_with?('-SNAPSHOT'))
+ }
+
+ for page in site.pages do
+ release_name = dir_to_releasename(page.dir)
+ if (release_name != nil)
+ release_data = releases[release_name]
+ page.data['version'] = release_name;
+ page.data['git-tree-base'] = release_data['git-tree-base'];
+ page.data['git-blob-base'] = release_data['git-blob-base'];
+ end
+ end
+ site.data['releases'] = releases.values.sort{|x,y| parse_version(y['name']) <=>
+ parse_version(x['name'])};
+ end
+ end
+end