Как представить топологию в кластере производства штормов с помощью IDE

Я сталкиваюсь с проблемой: необходимо Must submit topologies using the 'storm' client script so that StormSubmitter knows which jar to upload при отправке топологии в производственный кластер с помощью IDE, а то же самое, если я выполняю в командной строке с помощью команды storm jar, это бегать как небеса. Я видел примеры того же от githublink.

Для представления топологии я использую эти строки

conf.put(Config.NIMBUS_HOST, NIMBUS_NODE);
conf.put(Config.NIMBUS_THRIFT_PORT,6627);
conf.put(Config.STORM_ZOOKEEPER_PORT,2181);
conf.put(Config.STORM_ZOOKEEPER_SERVERS,ZOOKEEPER_ID);
conf.setNumWorkers(20);
conf.setMaxSpoutPending(5000);
StormSubmitter submitter = new StormSubmitter();
submitter.submitTopology("test", conf, builder.createTopology());

Пожалуйста, предложите мне, если это правильный подход к запуску?

Ответы

Ответ 1

Хорошо нашел решение. Когда мы запускали "штормовую банку", он запускал флаг собственности для storm.jar в представленной банке. Поэтому, если мы хотим программно представить банку, тогда просто установите флаг таким образом

System.setProperty("storm.jar", <path-to-jar>);

Например:

System.setProperty("storm.jar", "/Users/programming/apache-storm-1.0.1/lib/storm-core-1.0.1.jar");
StormSubmitter.submitTopology("myTopology", config, builder.createTopology());

Ответ 2

Для отправки топологии удаленному кластеру Storm вам нужно загрузить эту банку в машину nimbus, а затем отправить эту банку в кластер с помощью NimbusClient.
Вы можете сделать это следующим образом:

Map storm_conf = Utils.readStormConfig();
storm_conf.put("nimbus.host", "<Nimbus Machine IP>");
Client client = NimbusClient.getConfiguredClient(storm_conf)
                                .getClient();
String inputJar = "C:\\workspace\\TestStormRunner\\target\\TestStormRunner-0.0.1-SNAPSHOT-jar-with-dependencies.jar";
NimbusClient nimbus = new NimbusClient(storm_conf, "<Nimbus Machine IP>",
                                <Nimbus Machine Port>);
 // upload topology jar to Cluster using StormSubmitter
String uploadedJarLocation = StormSubmitter.submitJar(storm_conf,
                                inputJar);

String jsonConf = JSONValue.toJSONString(storm_conf);
nimbus.getClient().submitTopology("testtopology",
                      <uploadedJarLocation>, jsonConf, builder.createTopology());

Вот рабочий пример: Отправка топологии удаленному кластеру Storm

Ответ 3

Я не запускал Java-код для отправки себя, но я проверил команду штурма - и это файл python, который запускает java и http://nathanmarz.github.com/storm/doc/backtype/storm/StormSubmitter.html class

Единственное, о чем я думаю, что вам следует беспокоиться - включить все необходимые библиотеки при его выполнении.

Ответ 4

Я решил эту проблему на основе ответов @abhi и @Nishu Tayal, я хотел бы опубликовать здесь свой код:

public static void submitLocalTopologyWay1(String topologyName, Config topologyConf, 
        StormTopology topology, String localJar) {
    try {
        //get default storm config
        Map defaultStormConf = Utils.readStormConfig();
        defaultStormConf.putAll(topologyConf);

        //set JAR
        System.setProperty("storm.jar",localJar);

        //submit topology
        StormSubmitter.submitTopology(topologyName, defaultStormConf, topology);

    } catch (Exception e) {
        String errorMsg = "can't deploy topology " + topologyName + ", " + e.getMessage();
        System.out.println(errorMsg);
        e.printStackTrace();
    } 
}

public static void submitLocalTopologyWay2(String topologyName, Config topologyConf, 
        StormTopology topology, String localJar) {
    try {
        //get nimbus client
        Map defaultStormConf = Utils.readStormConfig();
        defaultStormConf.putAll(topologyConf);
        Client client = NimbusClient.getConfiguredClient(defaultStormConf).getClient();

        //upload JAR
        String remoteJar = StormSubmitter.submitJar(defaultStormConf, localJar);

        //submit topology
        client.submitTopology(topologyName, remoteJar, JSONValue.toJSONString(topologyConf), topology);

    } catch (Exception e) {
        String errorMsg = "can't deploy topology " + topologyName + ", " + e.getMessage();
        System.out.println(errorMsg);
        e.printStackTrace();
    } 
}

то вот тест, и вы должны сначала создать свой код в файле JAR.

public void testSubmitTopologySubmitLocalTopologyWay1() {   
    Config config = new Config();
    config.put(Config.NIMBUS_HOST,"9.119.84.179");   
    config.put(Config.NIMBUS_THRIFT_PORT, 6627);
    config.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList("9.119.84.177","9.119.84.178","9.119.84.176")); 
    config.put(Config.STORM_ZOOKEEPER_PORT,2181);

    config.put(Config.TOPOLOGY_WORKERS, 3);

    RemoteSubmitter.submitLocalTopologyWay1("word-count-test-1", config, 
            WordCountTopology.buildTopology(), // your topology
            "C:\\MyWorkspace\\project\\storm-sample-0.0.1-SNAPSHOT-jar-with-dependencies.jar");//the JAR file
}