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

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 / "multilang"

读完本文后:

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 in Compile += { baseDirectory.value / "multilang" } unmanagedClasspath in Compile += baseDirectory.value / "multilang"

从https://github.com/apache/storm/blob/master/storm-multilang/python/src/main/resources/resources/storm.py下载storm.py并将其复制/粘贴到 ${basedir}/multilang/resources/python 文件夹中。

在 Python 代码中，注释或删除任何 print 指令(由于 print 会写入标准输出，因此会在解析元组时引发 Java 异常)。

如果您的 Python 进程有点慢(我的是 NLP 进程，第一次需要一点设置)，请告诉您的拓扑再稍等一下:

config.put(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS, );

以及大量的尝试和错误 =)

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

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