Мой штормовой болт не может десериализоваться в кластерном режиме - PullRequest
Новая
       16

Мой штормовой болт не может десериализоваться в кластерном режиме

1 голос
/ 22 мая 2019

Я использую springboot и storm для демонстрации, он работает в локальном режиме, но сообщает об ошибке в кластерном режиме при отправке jar 

./storm jar storm-demo3-0.0.1-SNAPSHOT.jar org.springframework.boot.loader.JarLauncher simpleBoot

Когда я перемещаю springBoot и упаковываю с maven-тогда компилятор-плагин может хорошо работать 

<build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

это ошибка на супервизоре 

java.lang.RuntimeException: java.lang.ClassNotFoundException: com.fosung.share.stormdemo3.bolt.FilterBolt
        at org.apache.storm.utils.Utils.javaDeserialize(Utils.java:259) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.utils.Utils.getSetComponentObject(Utils.java:507) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$get_task_object.invoke(task.clj:76) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$mk_task_data$fn__6524.invoke(task.clj:180) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.util$assoc_apply_self.invoke(util.clj:931) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$mk_task_data.invoke(task.clj:172) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$mk_task.invoke(task.clj:184) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.executor$mk_executor$fn__10662.invoke(executor.clj:379) ~[storm-core-1.2.2.jar:1.2.2]
        at clojure.core$map$fn__4553.invoke(core.clj:2622) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
        at clojure.lang.RT.seq(RT.java:507) ~[clojure-1.7.0.jar:?]
        at clojure.core$seq__4128.invoke(core.clj:137) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$seq_reduce.invoke(protocols.clj:30) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6506.invoke(protocols.clj:101) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6452$G__6447__6465.invoke(protocols.clj:13) ~[clojure-1.7.0.jar:?]
        at clojure.core$reduce.invoke(core.clj:6519) ~[clojure-1.7.0.jar:?]
        at clojure.core$into.invoke(core.clj:6600) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.executor$mk_executor.invoke(executor.clj:380) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.worker$fn__11300$exec_fn__2470__auto__$reify__11302$iter__11307__11311$fn__11312.invoke(worker.clj:663) ~[storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
        at clojure.lang.RT.seq(RT.java:507) ~[clojure-1.7.0.jar:?]
        at clojure.core$seq__4128.invoke(core.clj:137) ~[clojure-1.7.0.jar:?]
        at clojure.core$dorun.invoke(core.clj:3009) ~[clojure-1.7.0.jar:?]
        at clojure.core$doall.invoke(core.clj:3025) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$fn__11300$exec_fn__2470__auto__$reify__11302.run(worker.clj:663) ~[storm-core-1.2.2.jar:1.2.2]
        at java.security.AccessController.doPrivileged(Native Method) ~[?:1.8.0_152]
        at javax.security.auth.Subject.doAs(Subject.java:422) ~[?:1.8.0_152]
        at org.apache.storm.daemon.worker$fn__11300$exec_fn__2470__auto____11301.invoke(worker.clj:633) ~[storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.AFn.applyToHelper(AFn.java:178) ~[clojure-1.7.0.jar:?]
        at clojure.lang.AFn.applyTo(AFn.java:144) ~[clojure-1.7.0.jar:?]
        at clojure.core$apply.invoke(core.clj:630) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$fn__11300$mk_worker__11391.doInvoke(worker.clj:605) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.RestFn.invoke(RestFn.java:512) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$_main.invoke(worker.clj:798) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.AFn.applyToHelper(AFn.java:165) [clojure-1.7.0.jar:?]
        at clojure.lang.AFn.applyTo(AFn.java:144) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker.main(Unknown Source) [storm-core-1.2.2.jar:1.2.2]
Caused by: java.lang.ClassNotFoundException: com.fosung.share.stormdemo3.bolt.FilterBolt
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[?:1.8.0_152]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[?:1.8.0_152]
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:338) ~[?:1.8.0_152]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[?:1.8.0_152]
        at java.lang.Class.forName0(Native Method) ~[?:1.8.0_152]
        at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:683) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1863) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1746) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2037) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1568) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:428) ~[?:1.8.0_152]
        at org.apache.storm.utils.Utils.javaDeserialize(Utils.java:253) ~[storm-core-1.2.2.jar:1.2.2]
        ... 38 more
2019-05-22 11:09:14.684 o.a.s.util main [ERROR] Halting process: ("Error on initialization")
java.lang.RuntimeException: ("Error on initialization")
        at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$fn__11300$mk_worker__11391.doInvoke(worker.clj:605) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.RestFn.invoke(RestFn.java:512) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$_main.invoke(worker.clj:798) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.AFn.applyToHelper(AFn.java:165) [clojure-1.7.0.jar:?]
        at clojure.lang.AFn.applyTo(AFn.java:144) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker.main(Unknown Source) [storm-core-1.2.2.jar:1.2.2]

my pom.xml 

<properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.2.2</version>
            <!--<scope>provided</scope>-->
            <exclusions>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-slf4j-impl</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-1.2-api</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-web</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <artifactId>ring-cors</artifactId>
                    <groupId>ring-cors</groupId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

MyTopology 

public class MyTopology {


    public  static void main(String[] args) {
        System.out.println("MyTopology main start");
        // 定义一个拓扑

        TopologyBuilder builder = new TopologyBuilder();
        // 设置1个Executeor(线程)，默认一个
        DataSpout dataSpout = new DataSpout();
        builder.setSpout("spoutId", dataSpout);
        // shuffleGrouping:表示是随机分组
        // 设置1个Executeor(线程)，和两个task
        FilterBolt filterBolt = new FilterBolt();
        InsertBolt insertBolt = new InsertBolt();

        builder.setBolt("filterBolt", filterBolt).setNumTasks(1).allGrouping("spoutId", "spoutId");
        builder.setBolt("insertBolt", insertBolt).setNumTasks(1).allGrouping("filterBolt", "spoutId");
        Config conf = new Config();

        try {
            // 有参数时，表示向集群提交作业，并把第一个参数当做topology名称
            // 没有参数时，本地提交
            if (args != null && args.length > 0) {
                System.out.println("运行远程模式");
                StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
            } else {
                // 启动本地模式
                System.out.println("运行本地模式");
                LocalCluster cluster = new LocalCluster();
                cluster.submitTopology("TopologyApp", conf, builder.createTopology());
            }
        } catch (Exception e) {
            System.out.println("storm启动失败!程序退出!");
            System.exit(1);
            e.printStackTrace();
        }
//        System.out.println("storm启动成功...");
    }
}

Мой носик 

public class DataSpout extends BaseRichSpout {

    SpoutOutputCollector collector;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
        System.out.println("spout open");
    }

    @Override
    public void nextTuple() {
        /*try {
            Thread.sleep(1000);
            return;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/
        System.out.println("spout nextTuple start");
        int rndomn = (int)Math.random() * 1000;

        collector.emit("spoutId", new Values(rndomn));

        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("spoutId", new Fields("spoutId"));
    }
}

Мой болт 


public class FilterBolt extends BaseRichBolt  {
    OutputCollector collector;
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        System.out.println("filter bolt start");
        Integer o = (Integer) input.getValues().get(0);
        if (o>10){
            collector.emit("spoutId", new Values(o));
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        //定义下个bolt接收streamId
        declarer.declareStream("spoutId", new Fields("spoutId"));

    }
}

1 Ответ

0 голосов
/ 24 мая 2019

Spring (boot) не очень подходит для Storm. Storm - это фреймворк, то есть он отвечает за управление жизненным циклом некоторых классов, таких как ваш болт. Поскольку Storm ничего не знает о Spring, внедрение зависимостей Spring не работает «из коробки». Можно настроить Spring для работы с частями приложения Storm, например, с помощью задача и рабочие хуки , которые могут позволить вам создать контекст Spring в Storm worker. Я не думаю, что я бы порекомендовал это, если у вас нет веских причин, чтобы нуждаться в весне.

Что касается ошибки, которую вы получаете, Storm не может найти один из ваших классов в банке, которую вы отправляете. Поскольку вы не опубликовали свой pom.xml для своей конфигурации Spring, это трудно сказать, но, возможно, вы используете плагин, который перемещает ваши классы. Когда вы отправляете топологию в Storm, Storm выполняет несколько этапов, которые вы должны понимать:

Сначала вы делаете storm jar com.yourcompany.yourMain. Это запустит JVM на вашем локальном компьютере (или везде, где вы запускаете команду), который запускает настройку вашей топологии, в вашем случае MyTopology.main. Затем установка сериализует ваши изливы и болты и отправляет банку и сериализованную топологию в Nimbus (отдельную JVM), которая, в свою очередь, отправляет ее супервизорам (еще одной отдельной JVM). На супервизорах JVM супервизора загружает несколько рабочих JVM для запуска вашей топологии. Каждая рабочая JVM запускается с помощью команды, подобной java -cp your-topology.jar org.apache.storm.Worker. Рабочие JVM загружают сериализованную топологию и классы в вашем банке топологии и загружают потоки для запуска носиков и болтов.

Эти этапы, скорее всего, являются причиной того, что он не работает для вас. Когда вы запускаете код установки топологии, вы делаете это с помощью команды Spring Boot, поэтому Spring Boot получает шанс на запуск. Когда топология запускается на рабочих машинах, JVM запускаются обычным старым вызовом основного метода, отличного от Spring, поэтому Spring не получает возможности для запуска.

Если вы решите не использовать Spring, вы можете найти рабочий пример POM здесь .

Другие ссылки, которые могут представлять интерес: более ранний ответ и проект , выполняющий интеграцию Spring для Storm.

...