Storm 0.10.0是否重用拓扑设计？

Question

因此，在某种程度上，拓扑描述了一个文件所需要的流，以计数它所拥有的唯一单词。

如果我有两个文件file1和file2，那么一个应该能够调用相同的拓扑并创建该拓扑的两个实例来运行相同的字数。

为了跟踪单词计数是否确实完成，一旦文件处理完毕，单词计数拓扑的实例应该具有完成状态。

对于文件2StormSubmitter.SubmitTopology（“WordCountTopology2”,conf,Builder.CreateTopology（））；

更别提使用storm客户端同样上传jar

Storm jar StormWordCount-1.0.0-jar-with-dependencies.jar com.company.wordCount1main.app“server”“filePath1”

另一个问题是，一旦文件被处理，拓扑就无法完成。在我们对拓扑发出杀戮之前，它们一直活着

暴风杀“WordCountTopology”

我知道，在流媒体世界中，消息来自像Kafka这样的消息队列，没有消息结束，但在实体/消息固定的文件世界中，这又有什么关系。

//可以在Storm中查询当前作业是否完成JobStatus status=StormSubmitter.GetTopologyStatus（conf,tracker）；

Answer 1

对于重复使用同一拓扑两次,您有两种可能：

1）为您的文件喷注使用构造函数参数，并使用不同的参数两次实例化相同的拓扑：

private StormTopology createMyTopology(String filename) {
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("File Spout", new FileSpout(filename));
    // add further spouts and bolts etc.
    return builder.createTopology();
}

public static void main(String[] args) {
    String file1 = "/path/to/file1";
    String file2 = "/path/to/file2";
    Config c = new Config();
    if(useFile1) {
        StormSubmitter.submitTopology("T1", c, createMyTopology(file1));
    } else {
        StormSubmitter.submitTopology("T1", c, createMyTopology(file2));
    }
}

2）作为另一种选择，您可以在open()方法中配置文件喷口。

public class FileSpout extends IRichSpout {
    @Override
    public void open(Map conf, ...) {
        String filenmae = (String)conf.get("FILENAME");
        // ...
    }
    // other methods omitted
}

    public static void main(String[] args) {
    String file1 = "/path/to/file1";
    String file2 = "/path/to/file2";
    Config c = new Config();
    if(useFile1) {
        c.put("FILENAME", file1);
    } else {
        c.put("FILENAME", file2);
    }

    // assembly topology...

    StormSubmitter.submitTopology("T", c, builder.createTopology());
}

Config cfg = new Config();
// set NIMBUS_HOST and NIMBUS_THRIFT_PORT in cfg
Client client = NimbusClient.getConfiguredClient(cfg).getClient();
TopologyInfo info = client.getTopologyInfo("topologyName");
// get emitted tuples...
client.killTopology("topologyName");