java - 使用 shell bolt 构建 Storm 拓扑

Question

我目前正在尝试实现与 R 语言集成的 Storm 拓扑。

作为起点，我采用了以下项目 ( https://github.com/allenday/R-Storm )，它通过扩展 ShellBolt 类来实现 R 集成，以及一个 R 库来处理 java 和 R 端之间的通信。

我的问题是，如果我基于常规(仅限 Java) bolt 创建拓扑，我可以毫无问题地将它们链接在一起。然而，当链条中间的一个 bolt 是 R Shell Bolt 时，它就会分崩离析:

5661 [Thread-18] ERROR backtype.storm.util - Async loop died!
java.lang.RuntimeException: java.lang.RuntimeException: java.lang.RuntimeException: Pipe to subprocess seems to be broken! No output read.
Shell Process Exception:


at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:87) ~[storm-0.9.0-wip16.jar:na]
at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:58) ~[storm-0.9.0-wip16.jar:na]
at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62) ~[storm-0.9.0-wip16.jar:na]
at backtype.storm.daemon.executor$fn__3557$fn__3569$fn__3616.invoke(executor.clj:715) ~[storm-0.9.0-wip16.jar:na]
at backtype.storm.util$async_loop$fn__436.invoke(util.clj:377) ~[storm-0.9.0-wip16.jar:na]
at clojure.lang.AFn.run(AFn.java:24) ~[clojure-1.4.0.jar:na]
at java.lang.Thread.run(Unknown Source) ~[na:1.7.0_25]

Caused by: java.lang.RuntimeException: java.lang.RuntimeException: Pipe to subprocess seems to be broken! No output read.

更具体地说，以下拓扑按预期工作:

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("spout", new RandomSentenceSpout(), 1);
builder.setBolt("permutebolt", new PermuteBolt(), 1).shuffleGrouping("spout");

其中 PermuteBolt 是 R Shell Bolt。此示例的日志显示预期输出:

6246 [Thread-18] INFO  backtype.storm.daemon.task - Emitting: spout default [four score and seven years ago]
6246 [Thread-16] INFO  backtype.storm.daemon.executor - Processing received message source: spout:3, stream: default, id: {}, [four score and seven years ago]
6261 [Thread-23] INFO  backtype.storm.daemon.task - Emitting: permutebolt default ["PERMUTE seven years ago and four score"]

但是，如果我添加另一个 bolt 从第一个 bolt 获取数据，例如:

builder.setBolt("permutebolt", new PermuteBolt(), 1).shuffleGrouping("spout");
builder.setBolt("identity", new IdentityBolt(new Fields("identity")), 1).fieldsGrouping("permutebolt", new Fields("permutation"));

上面打印的跟踪失败。另外，奇怪的是第二个失败的例子包含在项目中。

这是任何人以前遇到过的问题吗？

更新:我注意到这只发生在使用 R Shell Bolts 时，我已经尝试启动使用 python 脚本的 bolt 并且能够正常链接它们。

Answer 1

@andrei，这是在今天上传到 github 的 1.01 中修复的: https://github.com/allenday/R-Storm/releases/tag/v1.01

它已经提交给 CRAN，很快就会可用。

感谢您的报告。

-艾伦