You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2018/02/26 19:37:58 UTC
[1/2] storm git commit: STORM-2973: Update storm-perf Kafka
topologies to use storm-kafka-client
Repository: storm
Updated Branches:
refs/heads/master 51fb1263f -> 21f4173f2
STORM-2973: Update storm-perf Kafka topologies to use storm-kafka-client
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f03e936f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f03e936f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f03e936f
Branch: refs/heads/master
Commit: f03e936fdfec92bdc3995635afaadec899b1700d
Parents: a38f0c5
Author: Stig Rohde Døssing <sr...@apache.org>
Authored: Sat Feb 24 17:13:51 2018 +0100
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Sun Feb 25 00:15:50 2018 +0100
----------------------------------------------------------------------
examples/storm-perf/README.markdown | 5 +-
examples/storm-perf/pom.xml | 43 +++--
.../src/main/conf/KafkaClientHdfsTopo.yaml | 26 +++
.../storm-perf/src/main/conf/KafkaHdfsTopo.yaml | 26 ---
.../src/main/conf/KafkaSpoutNullBoltTopo.yaml | 23 ---
.../apache/storm/perf/KafkaClientHdfsTopo.java | 172 ++++++++++++++++++
.../org/apache/storm/perf/KafkaHdfsTopo.java | 177 -------------------
.../storm/perf/KafkaSpoutNullBoltTopo.java | 120 -------------
8 files changed, 226 insertions(+), 366 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/README.markdown
----------------------------------------------------------------------
diff --git a/examples/storm-perf/README.markdown b/examples/storm-perf/README.markdown
index 8dea58b..b0c4ffa 100644
--- a/examples/storm-perf/README.markdown
+++ b/examples/storm-perf/README.markdown
@@ -26,9 +26,8 @@ Topologies that measure I/O with external systems are designed to run in cluster
4. **FileReadWordCountTopo:** Measures speed of word counting. The spout loads a file into memory and emits these lines in an infinite loop. Supports cluster mode only.
5. **HdfsSpoutNullBoltTopo:** Measures speed at which HdfsSpout can read from HDFS. Supports cluster mode only.
6. **StrGenSpoutHdfsBoltTopo:** Measures speed at which HdfsBolt can write to HDFS. Supports cluster mode only.
-7. **KafkaSpoutNullBoltTopo:** Measures speed at which the storm-kafka KafkaSpout can read from Kafka. Supports cluster mode only.
-8. **KafkaHdfsTopo:** Measures how fast Storm can read from Kafka and write to HDFS, using the storm-kafka spout. Supports cluster mode only
-9. **KafkaClientSpoutNullBoltTopo:** Measures the speed at which the storm-kafka-client KafkaSpout can read from Kafka. Supports cluster mode only.
+7. **KafkaClientHdfsTopo:** Measures how fast Storm can read from Kafka and write to HDFS, using the storm-kafka-client spout. Supports cluster mode only
+8. **KafkaClientSpoutNullBoltTopo:** Measures the speed at which the storm-kafka-client KafkaSpout can read from Kafka. Supports cluster mode only.
## How to run ?
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-perf/pom.xml b/examples/storm-perf/pom.xml
index af23137..f8eb965 100644
--- a/examples/storm-perf/pom.xml
+++ b/examples/storm-perf/pom.xml
@@ -35,27 +35,41 @@
<build>
<plugins>
<plugin>
- <artifactId>maven-assembly-plugin</artifactId>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
<configuration>
- <descriptorRefs>
- <descriptorRef>jar-with-dependencies</descriptorRef>
- </descriptorRefs>
- <archive>
- <manifest>
- <mainClass />
- </manifest>
- </archive>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.sf</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.dsa</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ <exclude>META-INF/*.rsa</exclude>
+ <exclude>META-INF/*.EC</exclude>
+ <exclude>META-INF/*.ec</exclude>
+ <exclude>META-INF/MSFTSIG.SF</exclude>
+ <exclude>META-INF/MSFTSIG.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
</configuration>
<executions>
<execution>
- <id>make-assembly</id>
<phase>package</phase>
<goals>
- <goal>single</goal>
+ <goal>shade</goal>
</goals>
+ <configuration>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
+ </configuration>
</execution>
</executions>
-
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
@@ -100,11 +114,6 @@
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
- <artifactId>storm-kafka</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
<artifactId>storm-hdfs</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/src/main/conf/KafkaClientHdfsTopo.yaml
----------------------------------------------------------------------
diff --git a/examples/storm-perf/src/main/conf/KafkaClientHdfsTopo.yaml b/examples/storm-perf/src/main/conf/KafkaClientHdfsTopo.yaml
new file mode 100755
index 0000000..2047d1b
--- /dev/null
+++ b/examples/storm-perf/src/main/conf/KafkaClientHdfsTopo.yaml
@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+spout.count : 1
+bolt.count : 1
+kafka.topic : "kafka_topic"
+kafka.bootstrap.hosts : "localhost:9092"
+hdfs.uri : "hdfs://hdfs.namenode:8020"
+hdfs.dir : "/tmp/storm"
+hdfs.batch : 1000
+
+# storm config overrides
+topology.workers : 1
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/src/main/conf/KafkaHdfsTopo.yaml
----------------------------------------------------------------------
diff --git a/examples/storm-perf/src/main/conf/KafkaHdfsTopo.yaml b/examples/storm-perf/src/main/conf/KafkaHdfsTopo.yaml
deleted file mode 100755
index a8ed2f2..0000000
--- a/examples/storm-perf/src/main/conf/KafkaHdfsTopo.yaml
+++ /dev/null
@@ -1,26 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-spout.count : 1
-bolt.count : 1
-kafka.topic : "kafka_topic"
-zk.uri : "zkhostname:2181"
-hdfs.uri : "hdfs://hdfs.namenode:8020"
-hdfs.dir : "/tmp/storm"
-hdfs.batch : 1000
-
-# storm config overrides
-topology.workers : 1
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/src/main/conf/KafkaSpoutNullBoltTopo.yaml
----------------------------------------------------------------------
diff --git a/examples/storm-perf/src/main/conf/KafkaSpoutNullBoltTopo.yaml b/examples/storm-perf/src/main/conf/KafkaSpoutNullBoltTopo.yaml
deleted file mode 100644
index cde4c2e..0000000
--- a/examples/storm-perf/src/main/conf/KafkaSpoutNullBoltTopo.yaml
+++ /dev/null
@@ -1,23 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-spout.count : 1
-bolt.count : 1
-kafka.topic : "kafka_topic"
-zk.uri : "zkhostname:2181"
-
-# storm config overrides
-topology.workers : 1
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
----------------------------------------------------------------------
diff --git a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
new file mode 100755
index 0000000..0432d9e
--- /dev/null
+++ b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.storm.perf;
+
+import java.util.Map;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.hdfs.bolt.HdfsBolt;
+import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;
+import org.apache.storm.hdfs.bolt.format.FileNameFormat;
+import org.apache.storm.hdfs.bolt.format.RecordFormat;
+import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
+import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy;
+import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
+import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
+import org.apache.storm.kafka.spout.KafkaSpout;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig;
+import org.apache.storm.perf.utils.Helper;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.utils.ObjectReader;
+import org.apache.storm.utils.Utils;
+
+/***
+ * This topo helps measure speed of reading from Kafka and writing to Hdfs.
+ * Spout Reads from Kafka.
+ * Bolt writes to Hdfs
+ */
+
+public class KafkaClientHdfsTopo {
+
+ // configs - topo parallelism
+ public static final String SPOUT_NUM = "spout.count";
+ public static final String BOLT_NUM = "bolt.count";
+ // configs - kafka spout
+ public static final String KAFKA_TOPIC = "kafka.topic";
+ public static final String KAFKA_BOOTSTRAP_HOSTS = "kafka.bootstrap.hosts";
+ // configs - hdfs bolt
+ public static final String HDFS_URI = "hdfs.uri";
+ public static final String HDFS_PATH = "hdfs.dir";
+ public static final String HDFS_BATCH = "hdfs.batch";
+
+
+ public static final int DEFAULT_SPOUT_NUM = 1;
+ public static final int DEFAULT_BOLT_NUM = 1;
+ public static final int DEFAULT_HDFS_BATCH = 1000;
+
+ // names
+ public static final String TOPOLOGY_NAME = "KafkaHdfsTopo";
+ public static final String SPOUT_ID = "kafkaSpout";
+ public static final String BOLT_ID = "hdfsBolt";
+
+
+ static StormTopology getTopology(Map<String, Object> config) {
+
+ final int spoutNum = getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
+ final int boltNum = getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM);
+
+ final int hdfsBatch = getInt(config, HDFS_BATCH, DEFAULT_HDFS_BATCH);
+
+ // 1 - Setup Kafka Spout --------
+ String bootstrapHosts = getStr(config, KAFKA_BOOTSTRAP_HOSTS);
+ String topicName = getStr(config, KAFKA_TOPIC);
+
+ KafkaSpoutConfig<String, String> spoutConfig = KafkaSpoutConfig.builder(bootstrapHosts, topicName)
+ .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST)
+ .build();
+
+ KafkaSpout<String, String> spout = new KafkaSpout<>(spoutConfig);
+
+ // 2 - Setup HFS Bolt --------
+ String hdfsUrls = getStr(config, HDFS_URI);
+ RecordFormat format = new LineWriter("value");
+ SyncPolicy syncPolicy = new CountSyncPolicy(hdfsBatch);
+ FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f, FileSizeRotationPolicy.Units.GB);
+
+ FileNameFormat fileNameFormat = new DefaultFileNameFormat().withPath(getStr(config, HDFS_PATH));
+
+ // Instantiate the HdfsBolt
+ HdfsBolt bolt = new HdfsBolt()
+ .withFsUrl(hdfsUrls)
+ .withFileNameFormat(fileNameFormat)
+ .withRecordFormat(format)
+ .withRotationPolicy(rotationPolicy)
+ .withSyncPolicy(syncPolicy);
+
+
+ // 3 - Setup Topology --------
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout(SPOUT_ID, spout, spoutNum);
+ builder.setBolt(BOLT_ID, bolt, boltNum)
+ .localOrShuffleGrouping(SPOUT_ID);
+
+ return builder.createTopology();
+ }
+
+
+ public static int getInt(Map map, Object key, int def) {
+ return ObjectReader.getInt(Utils.get(map, key, def));
+ }
+
+ public static String getStr(Map map, Object key) {
+ return (String) map.get(key);
+ }
+
+
+ /**
+ * Copies text file content from sourceDir to destinationDir. Moves source files into sourceDir after its done consuming
+ */
+ public static void main(String[] args) throws Exception {
+
+ if (args.length != 2) {
+ System.err.println("args: runDurationSec topConfFile");
+ return;
+ }
+
+ Integer durationSec = Integer.parseInt(args[0]);
+ String confFile = args[1];
+ Map<String, Object> topoConf = Utils.findAndReadConfigFile(confFile);
+ topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
+ topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
+ topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY, "org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
+
+ topoConf.putAll(Utils.readCommandLineOpts());
+ // Submit topology to Storm cluster
+ Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME, topoConf, getTopology(topoConf));
+ }
+
+ public static class LineWriter implements RecordFormat {
+ private String lineDelimiter = System.lineSeparator();
+ private String fieldName;
+
+ public LineWriter(String fieldName) {
+ this.fieldName = fieldName;
+ }
+
+ /**
+ * Overrides the default record delimiter.
+ *
+ * @param delimiter
+ * @return
+ */
+ public LineWriter withLineDelimiter(String delimiter) {
+ this.lineDelimiter = delimiter;
+ return this;
+ }
+
+ @Override
+ public byte[] format(Tuple tuple) {
+ return (tuple.getValueByField(fieldName).toString() + this.lineDelimiter).getBytes();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaHdfsTopo.java
----------------------------------------------------------------------
diff --git a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaHdfsTopo.java b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaHdfsTopo.java
deleted file mode 100755
index bdd35b6..0000000
--- a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaHdfsTopo.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
-
-package org.apache.storm.perf;
-
-import java.util.Map;
-import java.util.UUID;
-
-import org.apache.storm.Config;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.hdfs.bolt.HdfsBolt;
-import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;
-import org.apache.storm.hdfs.bolt.format.FileNameFormat;
-import org.apache.storm.hdfs.bolt.format.RecordFormat;
-import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
-import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy;
-import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
-import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
-import org.apache.storm.kafka.BrokerHosts;
-import org.apache.storm.kafka.KafkaSpout;
-import org.apache.storm.kafka.SpoutConfig;
-import org.apache.storm.kafka.StringMultiSchemeWithTopic;
-import org.apache.storm.kafka.ZkHosts;
-import org.apache.storm.perf.utils.Helper;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.utils.ObjectReader;
-import org.apache.storm.utils.Utils;
-
-/***
- * This topo helps measure speed of reading from Kafka and writing to Hdfs.
- * Spout Reads from Kafka.
- * Bolt writes to Hdfs
- */
-
-public class KafkaHdfsTopo {
-
- // configs - topo parallelism
- public static final String SPOUT_NUM = "spout.count";
- public static final String BOLT_NUM = "bolt.count";
- // configs - kafka spout
- public static final String KAFKA_TOPIC = "kafka.topic";
- public static final String ZOOKEEPER_URI = "zk.uri";
- // configs - hdfs bolt
- public static final String HDFS_URI = "hdfs.uri";
- public static final String HDFS_PATH = "hdfs.dir";
- public static final String HDFS_BATCH = "hdfs.batch";
-
-
- public static final int DEFAULT_SPOUT_NUM = 1;
- public static final int DEFAULT_BOLT_NUM = 1;
- public static final int DEFAULT_HDFS_BATCH = 1000;
-
- // names
- public static final String TOPOLOGY_NAME = "KafkaHdfsTopo";
- public static final String SPOUT_ID = "kafkaSpout";
- public static final String BOLT_ID = "hdfsBolt";
-
-
- static StormTopology getTopology(Map<String, Object> config) {
-
- final int spoutNum = getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
- final int boltNum = getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM);
-
- final int hdfsBatch = getInt(config, HDFS_BATCH, DEFAULT_HDFS_BATCH);
-
- // 1 - Setup Kafka Spout --------
- String zkConnString = getStr(config, ZOOKEEPER_URI);
- String topicName = getStr(config, KAFKA_TOPIC);
-
- BrokerHosts brokerHosts = new ZkHosts(zkConnString);
- SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topicName, "/" + topicName, UUID.randomUUID().toString());
- spoutConfig.scheme = new StringMultiSchemeWithTopic();
- spoutConfig.ignoreZkOffsets = true;
-
- KafkaSpout spout = new KafkaSpout(spoutConfig);
-
- // 2 - Setup HFS Bolt --------
- String hdfsUrls = getStr(config, HDFS_URI);
- RecordFormat format = new LineWriter("str");
- SyncPolicy syncPolicy = new CountSyncPolicy(hdfsBatch);
- FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f, FileSizeRotationPolicy.Units.GB);
-
- FileNameFormat fileNameFormat = new DefaultFileNameFormat().withPath(getStr(config, HDFS_PATH));
-
- // Instantiate the HdfsBolt
- HdfsBolt bolt = new HdfsBolt()
- .withFsUrl(hdfsUrls)
- .withFileNameFormat(fileNameFormat)
- .withRecordFormat(format)
- .withRotationPolicy(rotationPolicy)
- .withSyncPolicy(syncPolicy);
-
-
- // 3 - Setup Topology --------
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout(SPOUT_ID, spout, spoutNum);
- builder.setBolt(BOLT_ID, bolt, boltNum)
- .localOrShuffleGrouping(SPOUT_ID);
-
- return builder.createTopology();
- }
-
-
- public static int getInt(Map map, Object key, int def) {
- return ObjectReader.getInt(Utils.get(map, key, def));
- }
-
- public static String getStr(Map map, Object key) {
- return (String) map.get(key);
- }
-
-
- /**
- * Copies text file content from sourceDir to destinationDir. Moves source files into sourceDir after its done consuming
- */
- public static void main(String[] args) throws Exception {
-
- if (args.length != 2) {
- System.err.println("args: runDurationSec topConfFile");
- return;
- }
-
- Integer durationSec = Integer.parseInt(args[0]);
- String confFile = args[1];
- Map<String, Object> topoConf = Utils.findAndReadConfigFile(confFile);
- topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
- topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
- topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY, "org.apache.storm.policy.WaitStrategyPark");
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
-
- topoConf.putAll(Utils.readCommandLineOpts());
- // Submit topology to Storm cluster
- Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME, topoConf, getTopology(topoConf));
- }
-
- public static class LineWriter implements RecordFormat {
- private String lineDelimiter = System.lineSeparator();
- private String fieldName;
-
- public LineWriter(String fieldName) {
- this.fieldName = fieldName;
- }
-
- /**
- * Overrides the default record delimiter.
- *
- * @param delimiter
- * @return
- */
- public LineWriter withLineDelimiter(String delimiter) {
- this.lineDelimiter = delimiter;
- return this;
- }
-
- @Override
- public byte[] format(Tuple tuple) {
- return (tuple.getValueByField(fieldName).toString() + this.lineDelimiter).getBytes();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/f03e936f/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaSpoutNullBoltTopo.java
----------------------------------------------------------------------
diff --git a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaSpoutNullBoltTopo.java b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaSpoutNullBoltTopo.java
deleted file mode 100755
index d31bb4a..0000000
--- a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaSpoutNullBoltTopo.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
-* Licensed to the Apache Software Foundation (ASF) under one
-* or more contributor license agreements. See the NOTICE file
-* distributed with this work for additional information
-* regarding copyright ownership. The ASF licenses this file
-* to you under the Apache License, Version 2.0 (the
-* "License"); you may not use this file except in compliance
-* with the License. You may obtain a copy of the License at
-*
-* http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License
-*/
-
-package org.apache.storm.perf;
-
-import java.util.Map;
-import java.util.UUID;
-
-import org.apache.storm.Config;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.kafka.BrokerHosts;
-import org.apache.storm.kafka.KafkaSpout;
-import org.apache.storm.kafka.SpoutConfig;
-import org.apache.storm.kafka.StringMultiSchemeWithTopic;
-import org.apache.storm.kafka.ZkHosts;
-import org.apache.storm.perf.bolt.DevNullBolt;
-import org.apache.storm.perf.utils.Helper;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.utils.ObjectReader;
-import org.apache.storm.utils.Utils;
-
-/***
- * This topo helps measure speed of reading from Kafka
- * Spout Reads from Kafka.
- * Bolt acks and discards tuples
- */
-public class KafkaSpoutNullBoltTopo {
-
- // configs - topo parallelism
- public static final String SPOUT_NUM = "spout.count";
- public static final String BOLT_NUM = "bolt.count";
-
- // configs - kafka spout
- public static final String KAFKA_TOPIC = "kafka.topic";
- public static final String ZOOKEEPER_URI = "zk.uri";
-
-
- public static final int DEFAULT_SPOUT_NUM = 1;
- public static final int DEFAULT_BOLT_NUM = 1;
-
- // names
- public static final String TOPOLOGY_NAME = "KafkaSpoutNullBoltTopo";
- public static final String SPOUT_ID = "kafkaSpout";
- public static final String BOLT_ID = "devNullBolt";
-
-
- static StormTopology getTopology(Map<String, Object> config) {
-
- final int spoutNum = getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
- final int boltNum = getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM);
- // 1 - Setup Kafka Spout --------
-
- String zkConnString = getStr(config, ZOOKEEPER_URI);
- String topicName = getStr(config, KAFKA_TOPIC);
-
- BrokerHosts brokerHosts = new ZkHosts(zkConnString);
- SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topicName, "/" + topicName, UUID.randomUUID().toString());
- spoutConfig.scheme = new StringMultiSchemeWithTopic();
- spoutConfig.ignoreZkOffsets = true;
-
- KafkaSpout spout = new KafkaSpout(spoutConfig);
-
- // 2 - DevNull Bolt --------
- DevNullBolt bolt = new DevNullBolt();
-
- // 3 - Setup Topology --------
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout(SPOUT_ID, spout, spoutNum);
- builder.setBolt(BOLT_ID, bolt, boltNum)
- .localOrShuffleGrouping(SPOUT_ID);
-
- return builder.createTopology();
- }
-
-
- public static int getInt(Map map, Object key, int def) {
- return ObjectReader.getInt(Utils.get(map, key, def));
- }
-
- public static String getStr(Map map, Object key) {
- return (String) map.get(key);
- }
-
-
- /**
- * Copies text file content from sourceDir to destinationDir. Moves source files into sourceDir after its done consuming
- */
- public static void main(String[] args) throws Exception {
- if (args.length != 2) {
- System.err.println("args: runDurationSec confFile");
- return;
- }
- Integer durationSec = Integer.parseInt(args[0]);
- Map<String, Object> topoConf = Utils.findAndReadConfigFile(args[1]);
- topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
- topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
- topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY, "org.apache.storm.policy.WaitStrategyPark");
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
-
- topoConf.putAll(Utils.readCommandLineOpts());
- // Submit to Storm cluster
- Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME, topoConf, getTopology(topoConf));
- }
-}
[2/2] storm git commit: Merge branch 'STORM-2973' of
https://github.com/srdo/storm into asfgit-master
Posted by sr...@apache.org.
Merge branch 'STORM-2973' of https://github.com/srdo/storm into asfgit-master
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/21f4173f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/21f4173f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/21f4173f
Branch: refs/heads/master
Commit: 21f4173f25b44eea8b985987afb20c476751a50a
Parents: 51fb126 f03e936
Author: Stig Rohde Døssing <sr...@apache.org>
Authored: Mon Feb 26 20:37:26 2018 +0100
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Mon Feb 26 20:37:26 2018 +0100
----------------------------------------------------------------------
examples/storm-perf/README.markdown | 5 +-
examples/storm-perf/pom.xml | 43 +++--
.../src/main/conf/KafkaClientHdfsTopo.yaml | 26 +++
.../storm-perf/src/main/conf/KafkaHdfsTopo.yaml | 26 ---
.../src/main/conf/KafkaSpoutNullBoltTopo.yaml | 23 ---
.../apache/storm/perf/KafkaClientHdfsTopo.java | 172 ++++++++++++++++++
.../org/apache/storm/perf/KafkaHdfsTopo.java | 177 -------------------
.../storm/perf/KafkaSpoutNullBoltTopo.java | 120 -------------
8 files changed, 226 insertions(+), 366 deletions(-)
----------------------------------------------------------------------