Now that we have a running cluster, let's revisit our earlier word count example and modify it so we can deploy it to a cluster as well as run it in local mode. The previous example used Storm's LocalCluster class to run in local mode:

LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());

Submitting a topology to a remote cluster is simply a matter of using Storm's StormSubmitter class, which exposes a method with the same name and signature:

StormSubmitter.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());

When developing Storm topologies, you usually aren't going to want to change code and recompile them to switch between running in local mode and deploying to a cluster. The standard way to handle this is to add an if/else block that makes that determination based on a command-line argument. In our updated example, if there are no command line arguments, we run the...