Metron Performance Tuning Guide

Overview

There are quite a few options you will be confronted with when tuning your Storm topologies and this is largely trial and error. As a general rule of thumb, we recommend starting with the defaults and smaller numbers in terms of parallelism while iteratively working up until the desired performance is achieved. You will find the offset lag tool indispensable while verifying your settings.

We won’t go into a full discussion about Storm’s architecture - see references section for more info - but there are some general rules of thumb that should be followed. It’s first important to understand the ways you can impact parallelism in a Storm topology.

• num tasks
• num executors (parallelism hint)
• num workers

Tasks are instances of a given spout or bolt, executors are threads in a process, and workers are jvm processes. You’ll want the number of tasks as a multiple of the number of executors, the number of executors as multiple of the number of workers, and the number of workers as a multiple of the number of machines. The main reason for this approach is that it will give a uniform distribution of work to each machine and jvm process. More often than not, your number of tasks will be equal to the number of executors, which is the default in Storm. Flux does not actually provide a way to independently set number of tasks, so for enrichments and indexing, which use Flux, num tasks will always equal num executors.

You can change the number of workers via the Storm property topology.workers

Other Storm Settings

topology.max.spout.pending

This is the maximum number of tuples that can be in flight (ie, not yet acked) at any given time within your topology. You set this as a form of backpressure to ensure you don’t flood your topology.

topology.ackers.executors

This specifies how many threads should be dedicated to tuple acking. We found that setting this equal to the number of partitions in your inbound Kafka topic worked well.

spout-config.json

{
    ...
    "spout.pollTimeoutMs" : 200,
    "spout.maxUncommittedOffsets" : 10000000,
    "spout.offsetCommitPeriodMs" : 30000
}

Above is a snippet for configuring parsers. These are the spout recommended defaults from Storm and are currently the defaults provided in the Kafka spout itself. In fact, if you find the recommended defaults work fine for you, then you can omit these settings altogether.

Where to Find Tuning Properties

Important: The parser topologies are deployed via a builder pattern that takes parameters from the CLI as set via Ambari. The enrichment and indexing topologies are configured using a Storm Flux file, a configuration properties file, and Ambari. Here is a setting materialization summary for each of the topology types:

• Parsers
  • Management UI -> parser json config and CLI -> Storm
• Enrichment
  • Ambari UI -> properties file -> Flux -> Storm
• Indexing
  • Ambari UI -> properties file -> Flux -> Storm

Parsers

This is a mapping of the various performance tuning properties for parsers and how they are materialized.

See more detail on starting parsers here

Enrichment

Note These recommendations are based on the deprecated split-join enrichment topology. See Enrichment Performance for tuning recommendations for the new default unified enrichment topology.

This is a mapping of the various performance tuning properties for enrichments and how they are materialized.

Flux file found here - $METRON_HOME/flux/enrichment/remote-splitjoin.yaml

Note 1: Changes to Flux file properties that are managed by Ambari will render Ambari unable to further manage the property.

Note 2: Many of these settings will be irrelevant in the alternate non-split-join topology

When adding Kafka spout properties, there are 3 ways you’ll do this.

1. Ambari: If they are properties managed by Ambari (noted in the table under ‘Ambari Property Name’), look for the setting in Ambari.

2. Flux -> kafkaProps: add a new key/value to the kafkaProps section HashMap on line 201. For example, if you want to set the Kafka Spout consumer’s session.timeout.ms to 30 seconds, you would add the following:

          -   name: "put"
              args:
                  - "session.timeout.ms"
                  - 30000
   

3. Flux -> kafkaConfig: add a new setter to the kafkaConfig section on line 230. For example, if you want to set the Kafka Spout consumer’s poll timeout to 200 milliseconds, you would add the following under configMethods:

            -   name: "setPollTimeoutMs"
                args:
                    - 200
   

Indexing (Batch)

This is a mapping of the various performance tuning properties for indexing and how they are materialized.

Flux file can be found here - $METRON_HOME/flux/indexing/batch/remote.yaml.

Note: Changes to Flux file properties that are managed by Ambari will render Ambari unable to further manage the property.

Note: HDFS sync policy is not currently managed via Ambari. You will need to modify the Flux file directly to accommodate this setting. e.g.

Add a new setter to the hdfsWriter around line 56. Lines 53-55 provided for context.

 53             -   name: "withRotationPolicy"
 54                 args:
 55                     - ref: "hdfsRotationPolicy
 56             -   name: "withSyncPolicy"
 57                 args:
 58                     - ref: "hdfsSyncPolicy

Add an hdfsSyncPolicy after the hdfsRotationPolicy that appears on line 41. e.g.

 41     -   id: "hdfsRotationPolicy"
...
 45           - "${bolt.hdfs.rotation.policy.units}"
 46
 47     -   id: "hdfsSyncPolicy"
 48         className: "org.apache.storm.hdfs.bolt.sync.CountSyncPolicy"
 49         constructorArgs:
 50           -  100000

Tooling

Kafka

• consumer group offset lag viewer
• There is a GUI tool to make creating, modifying, and generally managing your Kafka topics a bit easier - see https://github.com/yahoo/kafka-manager
• console consumer - useful for quickly verifying topic contents

Storm

• Storm UI - http://www.malinga.me/reading-and-understanding-the-storm-ui-storm-ui-explained/

Example - Viewing Kafka Offset Lags

First we need to setup some environment variables

export BROKERLIST=<your broker comma-delimated list of host:ports>
export ZOOKEEPER=<your zookeeper comma-delimated list of host:ports>
export KAFKA_HOME=<kafka home dir>
export METRON_HOME=<your metron home>
export HDP_HOME=<your HDP home>

If you have Kerberos enabled, setup the security protocol

$ cat /tmp/consumergroup.config
security.protocol=SASL_PLAINTEXT

Now run the following command for a running topology’s consumer group. In this example we are using enrichments.

${KAFKA_HOME}/bin/kafka-consumer-groups.sh \
    --command-config=/tmp/consumergroup.config \
    --describe \
    --group enrichments \
    --bootstrap-server $BROKERLIST \
    --new-consumer

This will return a table with the following output depicting offsets for all partitions and consumers associated with the specified consumer group:

GROUP                          TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER
enrichments                    enrichments        9          29746066        29746067        1               consumer-2_/xxx.xxx.xxx.xxx
enrichments                    enrichments        3          29754325        29754326        1               consumer-1_/xxx.xxx.xxx.xxx
enrichments                    enrichments        43         29754331        29754332        1               consumer-6_/xxx.xxx.xxx.xxx
...

Note: You won’t see any output until a topology is actually running because the consumer groups only exist while consumers in the spouts are up and running.

The primary column we’re concerned with paying attention to is the LAG column, which is the current delta calculation between the current and end offset for the partition. This tells us how close we are to keeping up with incoming data. And, as we found through multiple trials, whether there are any problems with specific consumers getting stuck.

Taking this one step further, it’s probably more useful if we can watch the offsets and lags change over time. In order to do this we’ll add a “watch” command and set the refresh rate to 10 seconds.

watch -n 10 -d ${KAFKA_HOME}/bin/kafka-consumer-groups.sh \
    --command-config=/tmp/consumergroup.config \
    --describe \
    --group enrichments \
    --bootstrap-server $BROKERLIST \
    --new-consumer

Every 10 seconds the command will re-run and the screen will be refreshed with new information. The most useful bit is that the watch command will highlight the differences from the current output and the last output screens.