Flux Development Guideline

Scenario

This topic applies only to the scenario of submitting and deploying a topology using the Flux framework in the Storm component. Determine the versions of the jar packages described in this chapter based on the actual situation.

The Flux framework is a framework provided by Storm. This framework is used to improve the topology deployment usability. Using the Flux framework, users can use a YAML file to define and deploy a topology and run the storm jar command to submit the topology. This mode facilitates topology deployment and submitting, and reduces the service development cycle.

Basic Syntax Description

Using Flux to define a topology can be classified into two scenarios: defining a new topology and defining an existing topology.

Using Flux to define a new topology

Using Flux to define a topology indicates using a YAML file to describe a topology. A complete topology definition must contain the following parts:

Topology name
List of components used for defining the topology
Topology configuration
Topology definition, including the spout list, bolt list, and stream list

Code sample for defining the topology name:

name: "yaml-topology"

Code sample for defining the component list:

#Simple component definition
components:
- id: "stringScheme"
className: "org.apache.storm.kafka.StringScheme"

#Use a constructor to define a component.
- id: "defaultTopicSelector"
className: "org.apache.storm.kafka.bolt.selector.DefaultTopicSelector"
constructorArgs:
- "output"

#Reference parameters as input arguments in a constructor, and use the `ref` tag to describe the reference.
#When using a reference, ensure that the referenced object has been defined.
- id: "stringMultiScheme"
className: "org.apache.storm.spout.SchemeAsMultiScheme"
constructorArgs:
- ref: "stringScheme"

#Reference the configuration items in the specified properties file as input arguments in a constructor, and use the `${}` tag to describe the reference.
#If the properties file is referenced, use the --filter my-prop.properties mode to specify the path of the properties file when you run the storm jar command to submit the topology. 
- id: "zkHosts"
className: "org.apache.storm.kafka.ZkHosts"
constructorArgs: 
- "${kafka.zookeeper.root.list}"

#Reference environment variables as input arguments in a constructor, and use the `${ENV-[NAME]}` tag to describe the reference.
#NAME must be a defined environment variable.
- id: "zkHosts"
className: "org.apache.storm.kafka.ZkHosts"
constructorArgs: 
- "${ENV-ZK_HOSTS}"

#Use the `properties` keyword to initialize the internal private variables.
- id: spoutConfig
className: "org.apache.storm.kafka.SpoutConfig"
constructorArgs:
- ref: "zkHosts"
- "input"
- "/kafka/input"
- "myId"
properties:
- name: "scheme"
ref: "stringMultiScheme"

Code sample for defining the topology configuration:

config:
#Simple configuration item
topology.workers: 1

#If the configuration item value is a list, use `[]` to indicate it.
topology.auto-credentials: ["class1","class2"]

#The configuration item value is in the map structure.
kafka.broker.properties: 
metadata.broker.list: "${metadata.broker.list}"
producer.type: "async"
request.required.acks: "0"
serializer.class: "kafka.serializer.StringEncoder"

Code sample for defining the spout/bolt list:

#Define the spout list.
spouts:
- id: "spout1"
className: "org.apache.storm.kafka.KafkaSpout"
constructorArgs:
- ref: "spoutConfig"
parallelism: 1

#Define the bolt list.
bolts:
- id: "bolt1"
className: "com.huawei.storm.example.hbase.WordCounter"
parallelism: 1

#Use a method to initialize an object, and the keyword is `configMethods`.
- id: "bolt2"
className: "org.apache.storm.hbase.bolt.HBaseBolt"
constructorArgs:
- "WordCount"
- ref: "mapper"
configMethods:
- name: "withConfigKey"
args: ["hbase.conf"]
parallelism: 1

Code sample for defining the stream list:

#To define the stream mode, you must specify the grouping mode. The keyword is `grouping`, and keywords for grouping methods provided currently are as follows:
#`ALL`, `CUSTOM`, `DIRECT`, `SHUFFLE`, `LOCAL_OR_SHUFFLE`, `FIELDS`, `GLOBAL`, and `NONE`.
#`CUSTOM` is used for a customized group.

#For the definition of a simple stream, the grouping mode is SHUFFLE.
streams:
- name: "spout1 --> bolt1"
from: "spout1"
to: "bolt1"
grouping: 
type: SHUFFLE

#If the grouping mode is FIELDS, parameters must be entered.
- name: "bolt1 --> bolt2"
from: "bolt1"
to: "bolt2"
grouping:
type: FIELDS
args: ["word"]

#If the grouping mode is CUSTOM, you must specify a customized grouping class.
- name: "bolt-1 --> bolt2"
from: "bolt-1"
to: "bolt-2"
grouping:
type: CUSTOM
customClass:
className: "org.apache.storm.testing.NGrouping"
constructorArgs:
- 1

Using Flux to define an existing topology
If a topology already exists (for example, a topology has already been defined by Java code), you can still use the Flux framework to submit and deploy the topology. In this situation, you must use the getTopology() method in the current topology definition (for example, MyTopology.java). The definition in Java is as follows:
```
public StormTopology getTopology(Config config)
or
public StormTopology getTopology(Map<String, Object> config)
```
In this situation, you can use the following YAML file to define the topology:
```
name: "existing-topology"#You can specify the topology name to any value.
topologySource:
className: "custom-class"#Specify the client class.
```
You can specify another method name to obtain StormTopology (non-getTopology() method). The YAML file example is as follows:
```
name: "existing-topology"
topologySource:
className: "custom-class "
methodName: "getTopologyWithDifferentMethodName"
```
The specified method must accept an input parameter of the Map<String, Object> type or the Config type and return an object of the org.apache.storm.generated.StormTopology type. This method is the same as the getTopology() method.

Procedure for Developing an Application

Verify that the Storm component has been installed and is running properly. If the services need to connect to other components, install the required components and ensure that the components are running properly.
Import storm-examples to the IntelliJ IDEA development environment. For details, see Configuring and Importing Sample Projects.
Develop client services. For details, see the related YAML application examples in the src/main/resources/flux-examples directory of the storm-examples project.
Obtain the related configuration files.

For details, this step applies only to the scenarios when other components, such as HDFS and HBase, need to be accessed for service requirements. For details about how to obtain the related configuration files, see Storm-HDFS Development Guideline or Storm-HBase Development Guideline. If the services do not require the related configuration files, skip this step.
Obtain the related JAR packages.
- Obtain the following JAR file from the streaming-cql-<HD-Version>/lib directory on the Storm client:
  - flux-core-<version>.jar
  - flux-wrappers-<version>.jar
- Obtain other related JAR packages for the services. For example, to obtain the related JAR packages required for accessing HDFS, see 5. Other scenarios are similar.

The Flux configuration file example is as follows.

It is a complete YAML file example for accessing the Kafka service.

name: "simple_kafka"

components:
- id: "zkHosts"#Object name
className: "org.apache.storm.kafka.ZkHosts"  #Complete class name
constructorArgs:#Constructor
- "${kafka.zookeeper.root.list}"  #Constructor parameter

- id: "stringScheme" 
className: "org.apache.storm.kafka.StringScheme"

- id: "stringMultiScheme"
className: "org.apache.storm.spout.SchemeAsMultiScheme"
constructorArgs:
- ref: "stringScheme"  #A reference is used, and the value is stringScheme that has been defined.

- id: spoutConfig
className: "org.apache.storm.kafka.SpoutConfig"
constructorArgs:
- ref: "zkHosts"  #A reference is used.
- "input"
- "/kafka/input"
- "myId"
properties:  #Use properties to set the private variable whose name is "scheme" in this object.
- name: "scheme"
ref: "stringMultiScheme"

- id: "defaultTopicSelector" 
className: "org.apache.storm.kafka.bolt.selector.DefaultTopicSelector"
constructorArgs:
- "output"

- id: "fieldNameBasedTupleToKafkaMapper"
className: "org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper"
constructorArgs:
- "words"  #The first input argument in the constructor
- "count"  #The second input argument in the constructor

config: 
topology.workers: 1  #Set the number of workers of the topology to 1.
kafka.broker.properties:  #Set the parameters related to Kafka, and the values are in the map structure.
metadata.broker.list: "${metadata.broker.list}"
producer.type: "async"
request.required.acks: "0"
serializer.class: "kafka.serializer.StringEncoder"

spouts:
- id: "kafkaSpout"  #Spout name
className: "org.apache.storm.kafka.KafkaSpout"  #Spout class name
constructorArgs:  #Use a constructor to perform initialization.
- ref: "spoutConfig"  #Reference parameters as input arguments in a constructor.
parallelism: 1  #Set the concurrency of the spout to 1.

bolts: 
- id: "splitBolt"
className: "com.huawei.storm.example.common.SplitSentenceBolt"
parallelism: 1

- id: "countBolt" 
className: "com.huawei.storm.example.kafka.CountBolt"
parallelism: 1

- id: "kafkaBolt" 
className: "org.apache.storm.kafka.bolt.KafkaBolt"
configMethods:  #Invoke an internal method of the object to initialize the object.
- name: "withTopicSelector"  #Name of the invoked internal method
args:  #Parameter required by the internal method
- ref: "defaultTopicSelector"  #Only one input argument is set, and it is referenced.
- name: "withTupleToKafkaMapper"  #Invoke the second internal method.
args:
- ref: "fieldNameBasedTupleToKafkaMapper"

#Define the data stream.
streams:
- name: "kafkaSpout --> splitBolt"  #Name of the first data stream, which is only for display
from: "kafkaSpout"  #Data stream start, whose value is kafkaSpout defined in spouts
to: "splitBolt"  #Data stream end, whose value is splitBolt defined in bolts
grouping:  #Define the grouping mode.
type: LOCAL_OR_SHUFFLE  #The grouping mode is local_or_shuffle.

- name: "splitBolt --> countBolt"  #The second data stream
from: "splitBolt"
to: "countBolt"
grouping: 
type: FIELDS  #The grouping mode is fields.
args: ["word"]  #Parameters must be entered for the fields mode.

- name: "countBolt --> kafkaBolt"  #The third data stream
from: "countBolt"
to: "kafkaBolt"
grouping:
type: SHUFFLE  #The grouping mode is shuffle, and no parameter needs to be entered.

Running the Application and Viewing Results

Export the local JAR package. For details, see Packaging IntelliJ IDEA Code.
Combine the configuration files and JAR packages obtained respectively in 4 and 5, and export a complete service JAR package. For details, see Packaging Services.
Copy the developed YAML file and the related properties file to any directory on the hose where the Storm client is located, for example, /opt.
Run the following command to submit the topology:

storm jar /opt/jartarget/source.jar org.apache.storm.flux.Flux --remote /opt/my-topology.yaml

If the services are set to be started locally, run the following command to submit the topology:

storm jar /opt/jartarget/source.jar org.apache.storm.flux.Flux --local /opt/my-topology.yaml

Submit local services by running commands only in a normal mode environment.

If the properties file is used, run the following command to submit the topology:

storm jar /opt/jartarget/source.jar org.apache.storm.flux.Flux --remote /opt/my-topology.yaml --filter /opt/my-prop.properties
After the topology is submitted successfully, log in to the Storm UI to view the topology.