9491 [Thread-7-worker-receiver-thread-0] INFO backtype.storm.messaging.loader - Starting receive-thread: [stormId: Main-1-1458493086, port: 1027, thread-id: 0 ]
9969 [Thread-6] INFO backtype.storm.daemon.executor - Loading executor kafka-reader:[2 2]
10001 [Thread-6] INFO backtype.storm.daemon.executor - Loaded executor tasks kafka-reader:[2 2]
10019 [Thread-6] INFO backtype.storm.daemon.executor - Finished loading executor kafka-reader:[2 2]
10037 [Thread-6] INFO backtype.storm.daemon.executor - Loading executor __system:[-1 -1]
10038 [Thread-6] INFO backtype.storm.daemon.executor - Loaded executor tasks __system:[-1 -1]
10048 [Thread-6] INFO backtype.storm.daemon.executor - Finished loading executor __system:[-1 -1]
10060 [Thread-6] INFO backtype.storm.daemon.executor - Loading executor __acker:[1 1]
10062 [Thread-6] INFO backtype.storm.daemon.executor - Loaded executor tasks __acker:[1 1]
10068 [Thread-6] INFO backtype.storm.daemon.executor - Timeouts disabled for executor __acker:[1 1]
10068 [Thread-6] INFO backtype.storm.daemon.executor - Finished loading executor __acker:[1 1]
10078 [Thread-6] INFO backtype.storm.daemon.worker - Worker has topology config {"storm.id" "Main-1-1458493086", "dev.zookeeper.path" "/tmp/dev-storm-zookeeper", "topology.tick.tuple.freq.secs" nil, "topology.builtin.metrics.bucket.size.secs" 60, "topology.fall.back.on.java.serialization" true, "topology.max.error.report.per.interval" 5, "zmq.linger.millis" 0, "topology.skip.missing.kryo.registrations" true, "storm.messaging.netty.client_worker_threads" 1, "ui.childopts" "-Xmx768m", "storm.zookeeper.session.timeout" 20000, "nimbus.reassign" true, "topology.trident.batch.emit.interval.millis" 50, "storm.messaging.netty.flush.check.interval.ms" 10, "nimbus.monitor.freq.secs" 10, "logviewer.childopts" "-Xmx128m", "java.library.path" "/usr/local/lib:/opt/local/lib:/usr/lib", "topology.executor.send.buffer.size" 1024, "storm.local.dir" "C:\\Users\\ADMINI~1\\AppData\\Local\\Temp\\ed96f14b-104c-47c0-99e5-b6e7b3728705", "storm.messaging.netty.buffer_size" 5242880, "supervisor.worker.start.timeout.secs" 120, "topology.enable.message.timeouts" true, "nimbus.cleanup.inbox.freq.secs" 600, "nimbus.inbox.jar.expiration.secs" 3600, "drpc.worker.threads" 64, "storm.meta.serialization.delegate" "backtype.storm.serialization.DefaultSerializationDelegate", "topology.worker.shared.thread.pool.size" 4, "nimbus.host" "localhost", "storm.messaging.netty.min_wait_ms" 100, "storm.zookeeper.port" 2000, "transactional.zookeeper.port" nil, "topology.executor.receive.buffer.size" 1024, "transactional.zookeeper.servers" nil, "storm.zookeeper.root" "/storm", "storm.zookeeper.retry.intervalceiling.millis" 30000, "supervisor.enable" true, "storm.messaging.netty.server_worker_threads" 1, "storm.zookeeper.servers" ["localhost"], "transactional.zookeeper.root" "/transactional", "topology.acker.executors" nil, "topology.kryo.decorators" (), "topology.name" "Main", "topology.transfer.buffer.size" 1024, "topology.worker.childopts" nil, "drpc.queue.size" 128, "worker.childopts" "-Xmx768m", "supervisor.heartbeat.frequency.secs" 5, "topology.error.throttle.interval.secs" 10, "zmq.hwm" 0, "drpc.port" 3772, "supervisor.monitor.frequency.secs" 3, "drpc.childopts" "-Xmx768m", "topology.receiver.buffer.size" 8, "task.heartbeat.frequency.secs" 3, "topology.tasks" nil, "storm.messaging.netty.max_retries" 300, "topology.spout.wait.strategy" "backtype.storm.spout.SleepSpoutWaitStrategy", "nimbus.thrift.max_buffer_size" 1048576, "topology.max.spout.pending" nil, "storm.zookeeper.retry.interval" 1000, "topology.sleep.spout.wait.strategy.time.ms" 1, "nimbus.topology.validator" "backtype.storm.nimbus.DefaultTopologyValidator", "supervisor.slots.ports" (1027 1028 1029), "topology.environment" nil, "topology.debug" false, "nimbus.task.launch.secs" 120, "nimbus.supervisor.timeout.secs" 60, "topology.kryo.register" nil, "topology.message.timeout.secs" 30, "task.refresh.poll.secs" 10, "topology.workers" 1, "supervisor.childopts" "-Xmx256m", "nimbus.thrift.port" 6627, "topology.stats.sample.rate" 0.05, "worker.heartbeat.frequency.secs" 1, "topology.tuple.serializer" "backtype.storm.serialization.types.ListDelegateSerializer", "topology.disruptor.wait.strategy" "com.lmax.disruptor.BlockingWaitStrategy", "topology.multilang.serializer" "backtype.storm.multilang.JsonSerializer", "nimbus.task.timeout.secs" 30, "storm.zookeeper.connection.timeout" 15000, "topology.kryo.factory" "backtype.storm.serialization.DefaultKryoFactory", "drpc.invocations.port" 3773, "logviewer.port" 8000, "zmq.threads" 1, "storm.zookeeper.retry.times" 5, "topology.worker.receiver.thread.count" 1, "storm.thrift.transport" "backtype.storm.security.auth.SimpleTransportPlugin", "topology.state.synchronization.timeout.secs" 60, "supervisor.worker.timeout.secs" 30, "nimbus.file.copy.expiration.secs" 600, "storm.messaging.transport" "backtype.storm.messaging.netty.Context", "logviewer.appender.name" "A1", "storm.messaging.netty.max_wait_ms" 1000, "drpc.request.timeout.secs" 600, "storm.local.mode.zmq" false, "ui.port" 8080, "nimbus.childopts" "-Xmx1024m", "storm.cluster.mode" "local", "topology.max.task.parallelism" 1, "storm.messaging.netty.transfer.batch.size" 262144, "topology.classpath" nil}
10078 [Thread-6] INFO backtype.storm.daemon.worker - Worker 6f4ebcd6-3ad0-41e6-a3e7-718296ffad7f for storm Main-1-1458493086 on 893e68c8-12d4-4ecd-a153-bae3832a2440:1027 has finished loading
10430 [refresh-active-timer] INFO backtype.storm.daemon.worker - All connections are ready for worker 893e68c8-12d4-4ecd-a153-bae3832a2440:1027 with id 6f4ebcd6-3ad0-41e6-a3e7-718296ffad7f
10521 [Thread-9-kafka-reader] INFO backtype.storm.daemon.executor - Opening spout kafka-reader:(2)
10531 [Thread-9-kafka-reader] ERROR backtype.storm.util - Async loop died!
java.lang.NoClassDefFoundError: org/apache/curator/RetryPolicy
at storm.kafka.KafkaSpout.open(KafkaSpout.java:85) ~[storm-kafka-0.9.5.jar:0.9.5]
at backtype.storm.daemon.executor$fn__6579$fn__6594.invoke(executor.clj:522) ~[storm-core-0.9.5.jar:0.9.5]
at backtype.storm.util$async_loop$fn__459.invoke(util.clj:461) ~[storm-core-0.9.5.jar:0.9.5]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Unknown Source) [na:1.8.0_73]
Caused by: java.lang.ClassNotFoundException: org.apache.curator.RetryPolicy
at java.net.URLClassLoader.findClass(Unknown Source) ~[na:1.8.0_73]
at java.lang.ClassLoader.loadClass(Unknown Source) ~[na:1.8.0_73]
at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source) ~[na:1.8.0_73]
at java.lang.ClassLoader.loadClass(Unknown Source) ~[na:1.8.0_73]
... 5 common frames omitted
10532 [Thread-9-kafka-reader] ERROR backtype.storm.daemon.executor -
java.lang.NoClassDefFoundError: org/apache/curator/RetryPolicy
at storm.kafka.KafkaSpout.open(KafkaSpout.java:85) ~[storm-kafka-0.9.5.jar:0.9.5]
at backtype.storm.daemon.executor$fn__6579$fn__6594.invoke(executor.clj:522) ~[storm-core-0.9.5.jar:0.9.5]
at backtype.storm.util$async_loop$fn__459.invoke(util.clj:461) ~[storm-core-0.9.5.jar:0.9.5]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Unknown Source) [na:1.8.0_73]
Caused by: java.lang.ClassNotFoundException: org.apache.curator.RetryPolicy
at java.net.URLClassLoader.findClass(Unknown Source) ~[na:1.8.0_73]
at java.lang.ClassLoader.loadClass(Unknown Source) ~[na:1.8.0_73]
at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source) ~[na:1.8.0_73]
at java.lang.ClassLoader.loadClass(Unknown Source) ~[na:1.8.0_73]
... 5 common frames omitted
10548 [Thread-11-__system] INFO backtype.storm.daemon.executor - Preparing bolt __system:(-1)
10556 [Thread-11-__system] INFO backtype.storm.daemon.executor - Prepared bolt __system:(-1)
10575 [Thread-13-__acker] INFO backtype.storm.daemon.executor - Preparing bolt __acker:(1)
10578 [Thread-13-__acker] INFO backtype.storm.daemon.executor - Prepared bolt __acker:(1)
10635 [Thread-9-kafka-reader] ERROR backtype.storm.util - Halting process: ("Worker died")
java.lang.RuntimeException: ("Worker died")
at backtype.storm.util$exit_process_BANG_.doInvoke(util.clj:325) [storm-core-0.9.5.jar:0.9.5]
at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.5.1.jar:na]
at backtype.storm.daemon.worker$fn__7024$fn__7025.invoke(worker.clj:493) [storm-core-0.9.5.jar:0.9.5]
at backtype.storm.daemon.executor$mk_executor_data$fn__6480$fn__6481.invoke(executor.clj:240) [storm-core-0.9.5.jar:0.9.5]
at backtype.storm.util$async_loop$fn__459.invoke(util.clj:473) [storm-core-0.9.5.jar:0.9.5]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Unknown Source) [na:1.8.0_73]
代码如下：
[mw_shl_code=java,true] public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException {
String zks = "node01:9092,node02:9092,node03:9092";
String topic = "test";
String zkRoot = "/storm"; // default zookeeper root configuration for storm
String id = "word";
BrokerHosts brokerHosts = new ZkHosts(zks);
SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
spoutConf.forceFromStart = false;
spoutConf.zkServers = Arrays.asList(new String[] {"node01", "node02", "node03"});
spoutConf.zkPort = 2181;
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("kafka-reader", new KafkaSpout(spoutConf)); // Kafka我们创建了一个5分区的Topic，这里并行度设置为5
builder.setBolt("rowkeycombine", new AdClickCombinedRowkey()).shuffleGrouping("kafka-reader");
builder.setBolt("update", new AdclickUpdate()).shuffleGrouping("rowkeycombine");
Config conf = new Config();
String name = Main.class.getSimpleName();
if (args != null && args.length > 0) {
// Nimbus host name passed from command line
conf.put(Config.NIMBUS_HOST, args[0]);
conf.setNumWorkers(3);
StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
} else {
conf.setMaxTaskParallelism(1);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(name, conf, builder.createTopology());
Thread.sleep(60000);
cluster.shutdown();
}[/mw_shl_code]
|
/2 
京ICP备2020039040号
简书网举报电话：021-34700000
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡
