Updated on 2022-06-01 GMT+08:00

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 of MRS. Determine the versions of the JAR files described in this section based on the actual situation.

The Flux framework is a framework provided by Storm 0.10.0. 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.

  1. 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

    Sample code for defining the topology name:

    name: "yaml-topology"

    Sample code 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"
    
     #Define the properties used by KafkaBolt.
    - id: "kafkaProducerProps"
        className: "java.util.Properties"
        configMethods:
          - name: "put"
            args:
              - "bootstrap.servers"
              - "${metadata.broker.list}"
          - name: "put"
            args:
              - "acks"
              - "1"
          - name: "put"
            args:
              - "key.serializer"
              - "org.apache.kafka.common.serialization.StringSerializer"
          - name: "put"
            args:
              - "value.serializer"
              - "org.apache.kafka.common.serialization.StringSerializer"
    

    Sample code 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"

    Sample code 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
    
    - id: "kafkaBolt"
      className: "org.apache.storm.kafka.bolt.KafkaBolt"
      configMethods:
        - name: "withTopicSelector"
          args: 
            - ref: "defaultTopicSelector"
        - name: "withProducerProperties"
          args: [ref: "kafkaProducerProps"]
        - name: "withTupleToKafkaMapper"
          args:
            - ref: "fieldNameBasedTupleToKafkaMapper"
    

    Sample code 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
  2. 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 backtype.storm.generated.StormTopology type. This method is the same as the getTopology() method.

Procedure for Developing an Application

  1. Verify that the Storm component has been installed and is running correctly. If the services need to connect to other components, install the required components and ensure that the components are running properly.
  2. Import storm-examples to the Eclipse development environment. For details, see Preparing the Windows Development Environment.
  3. Develop client services. For details, see the related YAML application examples in the src/main/resources/flux-examples directory of the storm-examples project.
  4. Obtain the related configuration files.

    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.

Example of the Flux Configuration File

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: "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"  #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"  #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

  1. Run the mvn package command. After the command is executed successfully, the storm-examples-1.0.jar file is generated in the target directory.
  2. Copy the JAR file and developed YAML file and related properties files copied to any directory on the hose where the Storm client is located, for example, /opt.
  3. Run the related command to submit the topology.

    storm jar /opt/jartarget/storm-examples-1.0.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/storm-examples-1.0.jar org.apache.storm.flux.Flux --local /opt/my-topology.yaml

    If a service is set to the local mode, ensure that the submitting environment is a normal one. Currently, services in local mode cannot be submitted by using commands in a security environment.

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

    storm jar /opt/jartarget/storm-examples-1.0.jar org.apache.storm.flux.Flux --remote /opt/my-topology.yaml --filter /opt/my-prop.properties

  4. After the topology is submitted successfully, log in to the Storm UI to view the topology.