部署 JAR 拓扑时 Storm 缺少资源文件夹

Question

我正在尝试部署一个使用 MultiLang Bolt（用 Python 编写）的拓扑：

builder.setBolt("avro-parser", new AvroBolt(), 3).shuffleGrouping("main-kafka-spout");
builder.setBolt("nlp-analyzer", new NLPBolt("/python/analyzer/audio_parser.py"), 2).shuffleGrouping("avro-parser");

我可以毫无问题地提交拓扑，但是检查日志文件我看到了这个错误：

java.lang.RuntimeException: Error when launching multilang subprocess

    at org.apache.storm.utils.ShellProcess.launch(ShellProcess.java:89) ~[storm-core-1.0.1.jar:1.0.1]
    at org.apache.storm.task.ShellBolt.prepare(ShellBolt.java:131) ~[storm-core-1.0.1.jar:1.0.1]
    at org.apache.storm.daemon.executor$fn__7953$fn__7966.invoke(executor.clj:792) ~[storm-core-1.0.1.jar:1.0.1]
    at org.apache.storm.util$async_loop$fn__625.invoke(util.clj:482) [storm-core-1.0.1.jar:1.0.1]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Unknown Source) [?:1.8.0_91]
Caused by: java.io.IOException: Cannot run program "/usr/bin/python" (in directory "/var/lib/storm/supervisor/stormdist/sintonea-topology-main-22-1472285031/resources"): error=2, No such file or directory
    at java.lang.ProcessBuilder.start(Unknown Source) ~[?:1.8.0_91]
    at org.apache.storm.utils.ShellProcess.launch(ShellProcess.java:82) ~[storm-core-1.0.1.jar:1.0.1]
    ... 5 more
Caused by: java.io.IOException: error=2, No such file or directory
    at java.lang.UNIXProcess.forkAndExec(Native Method) ~[?:1.8.0_91]
    at java.lang.UNIXProcess.<init>(Unknown Source) ~[?:1.8.0_91]
    at java.lang.ProcessImpl.start(Unknown Source) ~[?:1.8.0_91]
    at java.lang.ProcessBuilder.start(Unknown Source) ~[?:1.8.0_91]
    at org.apache.storm.utils.ShellProcess

当然，我注意到我在指定的日志跟踪中没有“资源”文件夹：

/var/lib/storm/supervisor/stormdist/sintonea-topology-main-22-1472285031/resources

如果您在该目录中执行“ls”命令，您会看到它：

stormcode.ser  stormconf.ser  stormjar.jar

为什么storm没有在那个目录中创建资源文件夹？

我在 src/main/resources 中有两个资源文件夹：

• schema：包含 avro 方案文件
• python：包含python代码

这些文件夹被复制到根目录中的 JAR 中，而不是 /resources 文件夹中。

这怎么可能？我有一个以这种方式使用模式的 AvroBolt，它没有给出任何问题（使用 DummyBolt 而不是 MultiLang Bolt 进行测试）：

_schema = parser.parse(getClass().getResourceAsStream("/schema/caller_request.avsc"));

编辑

我已将资源文件夹移至

baseDirectory.value / "多语言"

读完后：

http://storm.apache.org/releases/current/Creating-a-new-Storm-project.html

现在资源文件夹已创建并复制，但我仍然遇到同样的问题：

Serializer Exception:
/usr/bin/python: can't open file '/resources/python/analyzer/audio_parser.py': [Errno 2] No such file or directory

编辑 2

我找到了一种解决方法（在不同级别创建两个资源文件夹）：

1. 在 src 的父目录中创建一个 multilang/resources（上一级）。复制里面的python文件夹：{base-directory}/multilang/resources/python/*.py
2. 保留文件夹 src/main/schema：{base-directory}/src/main/resources/schema/caller_request.avsc。

并将其添加到我的 build.sbt 中：

unmanagedResourceDirectories in Compile += { baseDirectory.value / "multilang" }
unmanagedClasspath in Compile += baseDirectory.value / "multilang"

并将我的 ShellBot 设置为（忽略路径中的“资源”或“/”一词）：

NLPBolt nlpBolt = new NLPBolt("python/analyzer/audio_parser.py");

还有：

_schema = parser.parse(getClass().getResourceAsStream("/schema/caller_request.avsc"));

它似乎有效，但我不喜欢这个解决方案。有什么想法吗？

编辑 3

顺便说一下，这段代码：

import storm
#from nltk.stem.snowball import SnowballStemmer
from es_tagger import SpanishTagger

config = utils.load_json('python/analyzer/data/config.json')

class AudioParserBolt(storm.BasicBolt):

产生以下错误：

Serializer Exception:
Traceback (most recent call last):
  File "python/analyzer/audio_parser.py", line 27, in <module>
    class AudioParserBolt(storm.BasicBolt):
AttributeError: 'module' object has no attribute 'BasicBolt'

Answer 1

所以，这基本上就是我为使其正常工作所做的工作：

1. 创建文件夹 ${basedir}/multilang/resources --> 复制 python 代码。
2. 从 ${basedir}/src/main/resources 中删除 python 代码，只保留 Avro 方案。
3. 添加到 build.sbt 文件中：

编译中的 unmanagedResourceDirectories += { baseDirectory.value / "multilang" } 编译中的 unmanagedClasspath += baseDirectory.value / "multilang"

4. 从https://github.com/apache/storm/blob/master/storm-multilang/python/src/main/resources/resources/storm.py 下载storm.py 并将其复制/粘贴到${basedir}/multilang/resources/python 文件夹中。
5. 在 Python 代码中，注释或删除任何打印指令（引发解析元组的 Java 异常，因为 print 写入标准输出）。
6. 如果您的 Python 进程有点慢（我的是 NLP 进程，第一次设置需要一点时间），请告诉您的拓扑再等一会儿：

config.put(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS, );

1. 还有很多尝试和错误 =)

任何时候你需要在你的 Python 代码中使用一个外部文件，比如配置文件，这样调用它（父目录将是我们之前创建的 multilang/resources）：

config = utils.load_json('python/analyzer/data/config.json')